graflo.architecture.evolution.state_core.vocabulary

The target shape a lift converts a manifest towards.

Not a manifest. These are the pieces the planner instantiates against whatever types the input already has, which is the difference between a reference model you merge onto and a transformation you apply.

Two conventions the whole package rests on, both stated here so a reader does not have to infer them from the IRIs:

Time is grounded in PROV-O and SOSA, not OWL-Time. time:hasBeginning ranges over a time:Instant, not a literal, so grounding a DATETIME column in it is a claim that becomes false the moment the schema is projected to OWL. prov:generatedAtTime, prov:invalidatedAtTime and sosa:resultTime are literal-ranged and say the same thing truthfully. OWL-Time stays in exact_match at the concept level, where it is about the type.

Units are UCUM tokens, carried per row. An abstract observation type serves temperature and pressure alike, so it cannot name one unit in its contract without lying; result_unit is grounded in qudt:ucumCode and travels with the measurement. UCUM has no currency, so currency falls back to ISO-4217 alpha codes (USD, EUR).

Attributes

AGENT = 'Agent' module-attribute

DCTERMS = 'http://purl.org/dc/terms/' module-attribute

DCTERMS_IS_PART_OF = f'{DCTERMS}isPartOf' module-attribute

EVIDENCE = 'Evidence' module-attribute

OBSERVATION_SUFFIX = 'Observation' module-attribute

OBSERVED_PROPERTY = 'observed_property' module-attribute

PROV = 'http://www.w3.org/ns/prov#' module-attribute

PROV_ACTIVITY = f'{PROV}Activity' module-attribute

PROV_AGENT = f'{PROV}Agent' module-attribute

PROV_ENTITY = f'{PROV}Entity' module-attribute

PROV_GENERATED_AT = f'{PROV}generatedAtTime' module-attribute

PROV_INVALIDATED_AT = f'{PROV}invalidatedAtTime' module-attribute

PROV_SPECIALIZATION_OF = f'{PROV}specializationOf' module-attribute

PROV_WAS_ATTRIBUTED_TO = f'{PROV}wasAttributedTo' module-attribute

PROV_WAS_DERIVED_FROM = f'{PROV}wasDerivedFrom' module-attribute

PROV_WAS_INFLUENCED_BY = f'{PROV}wasInfluencedBy' module-attribute

QUDT = 'http://qudt.org/schema/qudt/' module-attribute

QUDT_UCUM_CODE = f'{QUDT}ucumCode' module-attribute

RESULT_TIME = 'result_time' module-attribute

RESULT_UNIT = 'result_unit' module-attribute

RESULT_VALUE = 'result_value' module-attribute

SOSA = 'http://www.w3.org/ns/sosa/' module-attribute

SOSA_FEATURE_OF_INTEREST = f'{SOSA}FeatureOfInterest' module-attribute

SOSA_HAS_FEATURE_OF_INTEREST = f'{SOSA}hasFeatureOfInterest' module-attribute

SOSA_HAS_SIMPLE_RESULT = f'{SOSA}hasSimpleResult' module-attribute

SOSA_OBSERVATION = f'{SOSA}Observation' module-attribute

SOSA_OBSERVED_PROPERTY = f'{SOSA}observedProperty' module-attribute

SOSA_RESULT_TIME = f'{SOSA}resultTime' module-attribute

SOSA_SENSOR = f'{SOSA}Sensor' module-attribute

SSN = 'http://www.w3.org/ns/ssn/' module-attribute

SSN_PROPERTY = f'{SSN}Property' module-attribute

STATE_SUFFIX = 'State' module-attribute

TIME = 'http://www.w3.org/2006/time#' module-attribute

TIME_PROPER_INTERVAL = f'{TIME}ProperInterval' module-attribute

VALID_FROM = 'valid_from' module-attribute

VALID_TO = 'valid_to' module-attribute

__all__ = ['AGENT', 'DCTERMS_IS_PART_OF', 'EVIDENCE', 'PROV_WAS_INFLUENCED_BY', 'RESULT_TIME', 'RESULT_UNIT', 'VALID_FROM', 'VALID_TO', 'agent_vertex', 'attributed_to_edge', 'derived_from_edge', 'evidence_vertex', 'feature_of_interest_edge', 'observation_type_name', 'observation_vertex', 'specialization_edge', 'state_type_name', 'state_vertex'] module-attribute

Classes

Functions:

agent_vertex()

Who or what acted or measured: a person, a team, a sensor, a system.

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def agent_vertex() -> Vertex:
    """Who or what acted or measured: a person, a team, a sensor, a system."""
    return Vertex(
        name=AGENT,
        description="Who or what acted or measured.",
        semantics=Semantics(iri=PROV_AGENT, exact_match=[PROV_AGENT, SOSA_SENSOR]),
        properties=[
            Field(name="agent_id", type=FieldType.STRING),
            Field(name="label", type=FieldType.STRING),
            Field(name="agent_kind", type=FieldType.STRING),
        ],
        identity=["agent_id"],
    )

attributed_to_edge()

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def attributed_to_edge() -> Edge:
    return _edge(
        EVIDENCE,
        AGENT,
        "wasAttributedTo",
        PROV_WAS_ATTRIBUTED_TO,
        "Who or what produced this evidence.",
    )

derived_from_edge(source)

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def derived_from_edge(source: str) -> Edge:
    return _edge(
        source,
        EVIDENCE,
        "wasDerivedFrom",
        PROV_WAS_DERIVED_FROM,
        "What this fact was read from.",
    )

evidence_vertex()

What a fact was read from, and the attachment point for extracted data.

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def evidence_vertex() -> Vertex:
    """What a fact was read from, and the attachment point for extracted data."""
    return Vertex(
        name=EVIDENCE,
        description=(
            "What a fact was read from: a document, an API response, a table row."
        ),
        semantics=Semantics(iri=PROV_ENTITY, exact_match=[PROV_ENTITY]),
        properties=[
            Field(name="evidence_id", type=FieldType.STRING),
            Field(name="source_uri", type=FieldType.STRING),
            Field(name="media_type", type=FieldType.STRING),
            Field(
                name="retrieved_at",
                type=FieldType.DATETIME,
                semantics=FieldSemantics(iri=PROV_GENERATED_AT),
            ),
        ],
        identity=["evidence_id"],
    )

feature_of_interest_edge(subject)

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def feature_of_interest_edge(subject: str) -> Edge:
    return _edge(
        observation_type_name(subject),
        subject,
        "hasFeatureOfInterest",
        SOSA_HAS_FEATURE_OF_INTEREST,
        f"The {subject} this observation measured.",
    )

observation_type_name(vertex)

Device -> DeviceObservation.

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def observation_type_name(vertex: str) -> str:
    """``Device`` -> ``DeviceObservation``."""
    return f"{vertex}{OBSERVATION_SUFFIX}"

observation_vertex(subject, key_fields)

The Observation type for one subject: a measurement at a time.

result_unit is a property rather than a contract-level declaration because this type is abstract: one measuring temperature and one measuring pressure are the same type here, so the unit has to travel with the row.

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def observation_vertex(subject: str, key_fields: list[Field]) -> Vertex:
    """The ``Observation`` type for one subject: a measurement at a time.

    ``result_unit`` is a property rather than a contract-level declaration
    because this type is abstract: one measuring temperature and one measuring
    pressure are the same type here, so the unit has to travel with the row.
    """
    return Vertex(
        name=observation_type_name(subject),
        description=f"A measurement of a {subject} at a time.",
        semantics=Semantics(iri=SOSA_OBSERVATION, exact_match=[SOSA_OBSERVATION]),
        properties=[
            *(field.model_copy(deep=True) for field in key_fields),
            Field(
                name=OBSERVED_PROPERTY,
                type=FieldType.STRING,
                semantics=FieldSemantics(
                    exact_match=[SOSA_OBSERVED_PROPERTY, SSN_PROPERTY]
                ),
            ),
            Field(
                name=RESULT_VALUE,
                type=FieldType.FLOAT,
                semantics=FieldSemantics(exact_match=[SOSA_HAS_SIMPLE_RESULT]),
            ),
            Field(
                name=RESULT_UNIT,
                type=FieldType.STRING,
                semantics=FieldSemantics(iri=QUDT_UCUM_CODE),
            ),
            Field(
                name=RESULT_TIME,
                type=FieldType.DATETIME,
                semantics=FieldSemantics(iri=SOSA_RESULT_TIME),
            ),
        ],
        hash_identity_properties=[
            *(f.name for f in key_fields),
            OBSERVED_PROPERTY,
            RESULT_TIME,
        ],
    )

specialization_edge(subject)

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def specialization_edge(subject: str) -> Edge:
    return _edge(
        state_type_name(subject),
        subject,
        "specializationOf",
        PROV_SPECIALIZATION_OF,
        f"The {subject} this state is a state of.",
    )

state_type_name(vertex)

Device -> DeviceState.

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def state_type_name(vertex: str) -> str:
    """``Device`` -> ``DeviceState``."""
    return f"{vertex}{STATE_SUFFIX}"

state_vertex(subject, key_fields, moved)

The State type for one subject: its key, its mutable facts, a validity interval.

Keyed on the subject's own key plus valid_from: the same property of the same entity holds many values over time, and those are different facts rather than revisions of one. Closing valid_to instead of overwriting is what makes the history queryable at all.

Source code in graflo/architecture/evolution/state_core/vocabulary.py

def state_vertex(subject: str, key_fields: list[Field], moved: list[Field]) -> Vertex:
    """The ``State`` type for one subject: its key, its mutable facts, a validity interval.

    Keyed on the subject's own key *plus* ``valid_from``: the same property of
    the same entity holds many values over time, and those are different facts
    rather than revisions of one. Closing ``valid_to`` instead of overwriting is
    what makes the history queryable at all.
    """
    return Vertex(
        name=state_type_name(subject),
        description=(
            f"Mutable facts about a {subject}, each holding over one interval. "
            "Closing `valid_to` rather than overwriting is what makes history "
            "queryable."
        ),
        semantics=Semantics(
            iri=PROV_ENTITY, exact_match=[PROV_ENTITY, TIME_PROPER_INTERVAL]
        ),
        properties=[
            *(field.model_copy(deep=True) for field in key_fields),
            *(field.model_copy(deep=True) for field in moved),
            *_validity_fields(),
        ],
        hash_identity_properties=[*(f.name for f in key_fields), VALID_FROM],
    )