Skip to content

graflo.architecture.evolution.state_core.spec

What a caller must declare for a lift, and what the lift works out itself.

The split is not arbitrary. A lift can see structure -- which edges leave directed unstated, which types have no grounding, where a property is a float with no unit -- but it cannot see meaning. Nothing in a schema says that ConfigurationItem denotes a sosa:FeatureOfInterest, or that temp_c is degrees Celsius rather than a count, or that status is a fact that changes over time while serial_number is not. Those are the four things this spec carries, and the reason it is required rather than optional.

Everything else is mechanical and never appears here: undeclared directionality, the scaffolding types and their groundings, the provenance edges.

Attributes

__all__ = ['EdgeGrounding', 'Grounding', 'LiftSpec'] module-attribute

Classes

EdgeGrounding

Bases: Grounding

A grounding addressed at one edge triple.

Source code in graflo/architecture/evolution/state_core/spec.py 
47
48
49
50
51
52
class EdgeGrounding(Grounding):
    """A grounding addressed at one edge triple."""

    source: str = PydanticField(..., description="Source vertex type name.")
    target: str = PydanticField(..., description="Target vertex type name.")
    relation: str | None = PydanticField(default=None, description="Relation name.")

Attributes

relation = PydanticField(default=None, description='Relation name.') class-attribute instance-attribute
source = PydanticField(..., description='Source vertex type name.') class-attribute instance-attribute
target = PydanticField(..., description='Target vertex type name.') class-attribute instance-attribute

Grounding

Bases: ConfigBaseModel

An external-vocabulary anchor a caller asserts for one element.

Source code in graflo/architecture/evolution/state_core/spec.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class Grounding(ConfigBaseModel):
    """An external-vocabulary anchor a caller asserts for one element."""

    iri: str | None = PydanticField(
        default=None, description="IRI of the concept this element denotes."
    )
    exact_match: list[str] = PydanticField(
        default_factory=list, description="IRIs asserted equivalent to it."
    )
    synonyms: list[str] = PydanticField(
        default_factory=list, description="Alternative labels an agent may meet."
    )

    @model_validator(mode="after")
    def _validate_says_something(self) -> Grounding:
        if not self.iri and not self.exact_match and not self.synonyms:
            raise ValueError(
                "a grounding must carry at least one of iri, exact_match, synonyms"
            )
        return self

Attributes

exact_match = PydanticField(default_factory=list, description='IRIs asserted equivalent to it.') class-attribute instance-attribute
iri = PydanticField(default=None, description='IRI of the concept this element denotes.') class-attribute instance-attribute
synonyms = PydanticField(default_factory=list, description='Alternative labels an agent may meet.') class-attribute instance-attribute

LiftSpec

Bases: ConfigBaseModel

The semantic input to :func:~graflo.architecture.evolution.state_core.plan_lift.

Source code in graflo/architecture/evolution/state_core/spec.py 
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 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
 92
 93
 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
class LiftSpec(ConfigBaseModel):
    """The semantic input to :func:`~graflo.architecture.evolution.state_core.plan_lift`."""

    grounding: dict[str, Grounding] = PydanticField(
        default_factory=dict,
        description="Per-vertex grounding: ``{type_name: Grounding}``.",
    )
    edge_grounding: list[EdgeGrounding] = PydanticField(
        default_factory=list,
        description="Per-edge grounding, addressed by triple.",
    )
    identity: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description=(
            "Explicit identity for types that would otherwise fall back to "
            "``identity_from_all_properties``. Required only for those."
        ),
    )
    stateful: dict[str, list[str]] = PydanticField(
        default_factory=dict,
        description=(
            "Per-type mutable properties: ``{type_name: [property, ...]}``. Each "
            "named type gains a ``<Type>State`` carrying those properties over a "
            "validity interval."
        ),
    )
    observed: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Types that gain a ``<Type>Observation`` scaffold for measurements "
            "taken of them at a time."
        ),
    )
    measured: dict[str, str] = PydanticField(
        default_factory=dict,
        description=(
            "Units for existing properties, as ``{'Type.property': ucum_token}``. "
            "UCUM has no currency, so currency uses ISO-4217 (``USD``)."
        ),
    )
    provenance: bool = PydanticField(
        default=True,
        description=(
            "Add ``Evidence`` and ``Agent`` with the provenance edges that make "
            "'where did this fact come from' answerable."
        ),
    )
    retire: Literal["move", "keep"] = PydanticField(
        default="move",
        description=(
            "What happens to a property named in ``stateful``. ``move`` removes it "
            "from the entity -- the honest lift, since a fact that changes over "
            "time does not belong on the thing it is about. ``keep`` leaves it as "
            "a denormalized current value beside the history."
        ),
    )

    @model_validator(mode="after")
    def _validate_measured_addresses(self) -> LiftSpec:
        bad = sorted(key for key in self.measured if key.count(".") != 1)
        if bad:
            raise ValueError(f"measured keys must read 'Type.property', got {bad}")
        return self

    def measured_for(self, vertex: str) -> dict[str, str]:
        """``{property: unit}`` for one type."""
        return {
            key.split(".", 1)[1]: unit
            for key, unit in self.measured.items()
            if key.split(".", 1)[0] == vertex
        }

Attributes

edge_grounding = PydanticField(default_factory=list, description='Per-edge grounding, addressed by triple.') class-attribute instance-attribute
grounding = PydanticField(default_factory=dict, description='Per-vertex grounding: ``{type_name: Grounding}``.') class-attribute instance-attribute
identity = PydanticField(default_factory=dict, description='Explicit identity for types that would otherwise fall back to ``identity_from_all_properties``. Required only for those.') class-attribute instance-attribute
measured = PydanticField(default_factory=dict, description="Units for existing properties, as ``{'Type.property': ucum_token}``. UCUM has no currency, so currency uses ISO-4217 (``USD``).") class-attribute instance-attribute
observed = PydanticField(default_factory=list, description='Types that gain a ``<Type>Observation`` scaffold for measurements taken of them at a time.') class-attribute instance-attribute
provenance = PydanticField(default=True, description="Add ``Evidence`` and ``Agent`` with the provenance edges that make 'where did this fact come from' answerable.") class-attribute instance-attribute
retire = PydanticField(default='move', description='What happens to a property named in ``stateful``. ``move`` removes it from the entity -- the honest lift, since a fact that changes over time does not belong on the thing it is about. ``keep`` leaves it as a denormalized current value beside the history.') class-attribute instance-attribute
stateful = PydanticField(default_factory=dict, description='Per-type mutable properties: ``{type_name: [property, ...]}``. Each named type gains a ``<Type>State`` carrying those properties over a validity interval.') class-attribute instance-attribute

Methods:

measured_for(vertex)

{property: unit} for one type.

Source code in graflo/architecture/evolution/state_core/spec.py 
119
120
121
122
123
124
125
def measured_for(self, vertex: str) -> dict[str, str]:
    """``{property: unit}`` for one type."""
    return {
        key.split(".", 1)[1]: unit
        for key, unit in self.measured.items()
        if key.split(".", 1)[0] == vertex
    }