Skip to content

graflo.architecture.evolution.aspects.state.vocabulary

The state aspect's types: how a fact changes over time, and what was measured.

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 compose onto and a transformation you apply. Time and unit conventions are stated in :mod:graflo.architecture.evolution.aspects.iris.

Attributes

OBSERVATION_SUFFIX = 'Observation' module-attribute

OBSERVED_PROPERTY = 'observed_property' module-attribute

RESULT_TIME = 'result_time' module-attribute

RESULT_UNIT = 'result_unit' module-attribute

RESULT_VALUE = 'result_value' module-attribute

STATE_SUFFIX = 'State' module-attribute

VALID_FROM = 'valid_from' module-attribute

VALID_TO = 'valid_to' module-attribute

__all__ = ['OBSERVED_PROPERTY', 'RESULT_TIME', 'RESULT_UNIT', 'RESULT_VALUE', 'VALID_FROM', 'VALID_TO', 'feature_of_interest_edge', 'observation_type_name', 'observation_vertex', 'specialization_edge', 'state_type_name', 'state_vertex'] module-attribute

Classes

Functions:

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],
    )