Skip to content

graflo.architecture.evolution.state_core.vocabulary

IRIs are in :mod:graflo.architecture.evolution.aspects.iris, Agent and the PROV-O edges in :mod:~graflo.architecture.evolution.aspects.provenance.vocabulary, the state and observation scaffolds in :mod:~graflo.architecture.evolution.aspects.state.vocabulary. EVIDENCE and evidence_vertex resolve to SOURCE_RECORD and source_record_vertex, the vertex a lift now mints in its place.

Attributes

AGENT = 'Agent' module-attribute

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

EVIDENCE = 'SourceRecord' module-attribute

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

RESULT_TIME = 'result_time' module-attribute

RESULT_UNIT = 'result_unit' 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

__getattr__ = forwarder(__name__, {'DCTERMS_IS_PART_OF': (_IRIS, 'DCTERMS_IS_PART_OF'), 'PROV_WAS_INFLUENCED_BY': (_IRIS, 'PROV_WAS_INFLUENCED_BY'), 'AGENT': (_PROVENANCE, 'AGENT'), 'EVIDENCE': (_PROVENANCE, 'SOURCE_RECORD'), 'agent_vertex': (_PROVENANCE, 'agent_vertex'), 'attributed_to_edge': (_PROVENANCE, 'attributed_to_edge'), 'derived_from_edge': (_PROVENANCE, 'derived_from_edge'), 'evidence_vertex': (_PROVENANCE, 'source_record_vertex'), **{name: (_STATE, name) for name in ('RESULT_TIME', 'RESULT_UNIT', 'VALID_FROM', 'VALID_TO', 'feature_of_interest_edge', 'observation_type_name', 'observation_vertex', 'specialization_edge', 'state_type_name', 'state_vertex')}}) module-attribute

Functions:

agent_vertex()

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

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
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,
                semantics=FieldSemantics(iri=RDFS_LABEL),
            ),
            Field(name="agent_kind", type=FieldType.STRING),
        ],
        identity=["agent_id"],
    )

attributed_to_edge()

SourceRecord -wasAttributedTo-> Agent, with the agent's role on the edge.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
129
130
131
132
133
134
135
136
137
138
def attributed_to_edge() -> Edge:
    """``SourceRecord -wasAttributedTo-> Agent``, with the agent's role on the edge."""
    return grounded_edge(
        SOURCE_RECORD,
        AGENT,
        "wasAttributedTo",
        PROV_WAS_ATTRIBUTED_TO,
        "Who or what produced this record, and in which role.",
        properties=[Field(name=ROLE, type=FieldType.STRING)],
    )

derived_from_edge(source)

<source> -wasDerivedFrom-> SourceRecord.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
107
108
109
110
111
112
113
114
115
def derived_from_edge(source: str) -> Edge:
    """``<source> -wasDerivedFrom-> SourceRecord``."""
    return grounded_edge(
        source,
        SOURCE_RECORD,
        "wasDerivedFrom",
        PROV_WAS_DERIVED_FROM,
        "What this fact was read from.",
    )

evidence_vertex()

The record a fact was read from: its address and hash, never its content.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def source_record_vertex() -> Vertex:
    """The record a fact was read from: its address and hash, never its content."""
    return Vertex(
        name=SOURCE_RECORD,
        description=(
            "The record a fact was read from: a document, a message, an API "
            "response, a table row. Carries where the record is and a hash of "
            "what it held, not the content itself."
        ),
        semantics=Semantics(iri=PROV_ENTITY, exact_match=[PROV_ENTITY]),
        properties=[
            Field(name=SOURCE_RECORD_ID, type=FieldType.STRING),
            Field(name=SOURCE_URI, type=FieldType.STRING),
            Field(name=MEDIA_TYPE, type=FieldType.STRING),
            Field(name=CONTENT_HASH, type=FieldType.STRING),
            Field(
                name=RETRIEVED_AT,
                type=FieldType.DATETIME,
                semantics=FieldSemantics(iri=PROV_GENERATED_AT),
            ),
        ],
        identity=[SOURCE_RECORD_ID],
    )

feature_of_interest_edge(subject)

<Subject>Observation -hasFeatureOfInterest-> <Subject>.

Source code in graflo/architecture/evolution/aspects/state/vocabulary.py 
149
150
151
152
153
154
155
156
157
def feature_of_interest_edge(subject: str) -> Edge:
    """``<Subject>Observation -hasFeatureOfInterest-> <Subject>``."""
    return grounded_edge(
        observation_type_name(subject),
        subject,
        "hasFeatureOfInterest",
        SOSA_HAS_FEATURE_OF_INTEREST,
        f"The {subject} this observation measured.",
    )

observation_type_name(vertex)

Machine -> MachineObservation.

Source code in graflo/architecture/evolution/aspects/state/vocabulary.py 
47
48
49
def observation_type_name(vertex: str) -> str:
    """``Machine`` -> ``MachineObservation``."""
    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/aspects/state/vocabulary.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
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 `{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)

<Subject>State -specializationOf-> <Subject>.

Source code in graflo/architecture/evolution/aspects/state/vocabulary.py 
138
139
140
141
142
143
144
145
146
def specialization_edge(subject: str) -> Edge:
    """``<Subject>State -specializationOf-> <Subject>``."""
    return grounded_edge(
        state_type_name(subject),
        subject,
        "specializationOf",
        PROV_SPECIALIZATION_OF,
        f"The {subject} this state is a state of.",
    )

state_type_name(vertex)

Machine -> MachineState.

Source code in graflo/architecture/evolution/aspects/state/vocabulary.py 
42
43
44
def state_type_name(vertex: str) -> str:
    """``Machine`` -> ``MachineState``."""
    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/aspects/state/vocabulary.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
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 `{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],
    )