Skip to content

graflo.architecture.evolution.state_core

state-core: lifting an arbitrary manifest into a twin-ready schema.

A meta-layer rather than a model. Given any manifest and a statement of what its types mean, :func:plan_lift emits the operations that add temporal validity and provenance to it: state reified onto its own types with a validity interval, measurements carrying their unit, facts attributable to evidence and an agent.

The output is an op list, so the conversion is replayable, invertible and reviewable through the machinery every other manifest change already uses.

Modules:

Name Description
plan

Planning the operations that lift a manifest into a twin-ready shape.
spec

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

The target shape a lift converts a manifest towards.

Attributes

__all__ = ['EdgeGrounding', 'Grounding', 'LiftError', 'LiftSpec', 'plan_lift'] 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

LiftError

Bases: ValueError

The lift cannot be planned, and the message says which declaration is missing.

Raised rather than guessed. Every case here is one where the planner can see that something is wrong but not what the right answer is -- an ungrounded type, a name it would have to overwrite, a property it was told to move that is part of the key.

Source code in graflo/architecture/evolution/state_core/plan.py 
43
44
45
46
47
48
49
50
class LiftError(ValueError):
    """The lift cannot be planned, and the message says which declaration is missing.

    Raised rather than guessed. Every case here is one where the planner can see
    that something is wrong but not what the right answer is -- an ungrounded
    type, a name it would have to overwrite, a property it was told to move that
    is part of the key.
    """

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
    }

Functions:

plan_lift(manifest, spec, *, authored=None)

The ops that lift manifest into a twin-ready schema.

Parameters:

Name Type Description Default
manifest GraphManifest

the manifest to lift, already finish_init()-ed.

 required
spec LiftSpec

the semantic declarations the planner cannot infer.

 required
authored dict[str, Any] | None

the manifest as written, when available. Only an authored document distinguishes "this vertex declared no identity" from "this vertex declared every property as its identity", so without it the planner cannot detect the fallback and does not try.

 None

Returns:

Type Description
list[ManifestOp]

Ops in application order: grounding first, then structure, then the
list[ManifestOp]

scaffolding, then the removals. Ordering matters -- a property is moved
list[ManifestOp]

onto its State before it is removed from the entity.

Raises:

Type Description
LiftError

a declaration the planner needs is missing, or a name it would mint is already taken.
Source code in graflo/architecture/evolution/state_core/plan.py 
 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
def plan_lift(
    manifest: GraphManifest,
    spec: LiftSpec,
    *,
    authored: dict[str, Any] | None = None,
) -> list[ManifestOp]:
    """The ops that lift *manifest* into a twin-ready schema.

    Args:
        manifest: the manifest to lift, already ``finish_init()``-ed.
        spec: the semantic declarations the planner cannot infer.
        authored: the manifest as written, when available. Only an authored
            document distinguishes "this vertex declared no identity" from
            "this vertex declared every property as its identity", so without
            it the planner cannot detect the fallback and does not try.

    Returns:
        Ops in application order: grounding first, then structure, then the
        scaffolding, then the removals. Ordering matters -- a property is moved
        onto its ``State`` before it is removed from the entity.

    Raises:
        LiftError: a declaration the planner needs is missing, or a name it
            would mint is already taken.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise LiftError(
            "cannot lift a manifest with no schema block: there are no types to "
            "ground and nothing to attach state to"
        )

    core = schema.core_schema
    vertex_config = core.vertex_config
    existing = set(vertex_config.vertex_set)
    by_name = {vertex.name: vertex for vertex in vertex_config.vertices}

    unknown = sorted(
        (set(spec.grounding) | set(spec.stateful) | set(spec.observed)) - existing
    )
    if unknown:
        raise LiftError(f"spec names types the manifest does not have: {unknown}")

    ops: list[ManifestOp] = []
    ops += _identity_ops(spec, vertex_config, authored)
    ops += _grounding_ops(spec, core)
    ops += _directionality_ops(core)
    ops += _unit_ops(spec, by_name)
    scaffold_ops, scaffolded = _scaffold_ops(spec, by_name, existing)
    ops += scaffold_ops
    ops += _provenance_ops(spec, existing, scaffolded)
    ops += _retire_ops(spec, by_name)
    return ops