Skip to content

graflo.architecture.evolution.semantics

Applying grounding changes to a manifest.

Grounding was authorable only at the moment a type was written: the schema models carry semantics blocks, but no operation could attach one to an element that already existed. A manifest that arrived ungrounded -- an inferred one, or anything authored before the block existed -- could therefore never be grounded through the op system at all, only rewritten by hand, which is not replayable and not reviewable.

These three handlers close that. They are deliberately not in physical.py: grounding is never consulted at execution time and has no physical projection, so it changes what a reader can learn from the schema and nothing else.

Attributes

__all__ = ['apply_set_edge_semantics', 'apply_set_field_semantics', 'apply_set_vertex_descriptions', 'apply_set_vertex_semantics'] module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Functions:

apply_set_edge_semantics(manifest, op)

Attach, replace or clear the grounding on selected edge relations.

Source code in graflo/architecture/evolution/semantics.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
def apply_set_edge_semantics(manifest: GraphManifest, op: SetEdgeSemanticsOp) -> None:
    """Attach, replace or clear the grounding on selected edge relations."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_edge_semantics requires graph_schema")

    by_edge_id = {edge.edge_id: edge for edge in schema.core_schema.edge_config.edges}
    unknown = sorted(
        str(selector.edge_id())
        for selector in op.edges
        if selector.edge_id() not in by_edge_id
    )
    if unknown:
        raise ValueError(f"set_edge_semantics: unknown edges: {unknown}")

    for selector in op.edges:
        edge = by_edge_id[selector.edge_id()]
        edge.semantics = (
            op.semantics.model_copy(deep=True) if op.semantics is not None else None
        )

    schema.finish_init()

apply_set_field_semantics(manifest, op)

Attach, replace or clear the grounding on selected vertex or edge properties.

Source code in graflo/architecture/evolution/semantics.py 
 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
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def apply_set_field_semantics(manifest: GraphManifest, op: SetFieldSemanticsOp) -> None:
    """Attach, replace or clear the grounding on selected vertex or edge properties."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_field_semantics requires graph_schema")

    vertex_config = schema.core_schema.vertex_config
    vertex_targets = [t for t in op.targets if isinstance(t, FieldSemanticsTarget)]
    edge_targets = [t for t in op.targets if isinstance(t, EdgeFieldSemanticsTarget)]

    unknown_vertices = sorted(
        {target.vertex for target in vertex_targets} - vertex_config.vertex_set
    )
    if unknown_vertices:
        raise ValueError(f"set_field_semantics: unknown vertices: {unknown_vertices}")
    by_edge_id = {edge.edge_id: edge for edge in schema.core_schema.edge_config.edges}
    unknown_edges = sorted(
        str(target.edge_id())
        for target in edge_targets
        if target.edge_id() not in by_edge_id
    )
    if unknown_edges:
        raise ValueError(f"set_field_semantics: unknown edges: {unknown_edges}")

    by_name = {vertex.name: vertex for vertex in vertex_config.vertices}
    holders = [(target, by_name[target.vertex]) for target in vertex_targets] + [
        (target, by_edge_id[target.edge_id()]) for target in edge_targets
    ]
    missing: list[str] = []
    for target, holder in holders:
        if target.field not in {field.name for field in holder.properties}:
            label = (
                target.vertex
                if isinstance(target, FieldSemanticsTarget)
                else str(target.edge_id())
            )
            missing.append(f"{label}.{target.field}")
    if missing:
        # Naming the property rather than the vertex: the likely mistake is a
        # stale field name after a rename, and "unknown vertex" would send the
        # reader looking in the wrong place.
        raise ValueError(f"set_field_semantics: unknown properties: {sorted(missing)}")

    for target, holder in holders:
        for field in holder.properties:
            if field.name != target.field:
                continue
            field.semantics = (
                target.semantics.model_copy(deep=True)
                if target.semantics is not None
                else None
            )

    schema.finish_init()

apply_set_vertex_descriptions(manifest, op)

Set, replace or clear the description on selected vertex types.

Source code in graflo/architecture/evolution/semantics.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def apply_set_vertex_descriptions(
    manifest: GraphManifest, op: SetVertexDescriptionsOp
) -> None:
    """Set, replace or clear the description on selected vertex types."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_vertex_descriptions requires graph_schema")

    vertex_config = schema.core_schema.vertex_config
    unknown = sorted(set(op.descriptions) - vertex_config.vertex_set)
    if unknown:
        raise ValueError(f"set_vertex_descriptions: unknown vertices: {unknown}")

    for vertex in vertex_config.vertices:
        if vertex.name in op.descriptions:
            vertex.description = op.descriptions[vertex.name]

    schema.finish_init()

apply_set_vertex_semantics(manifest, op)

Attach, replace or clear the grounding on selected vertex types.

Source code in graflo/architecture/evolution/semantics.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def apply_set_vertex_semantics(
    manifest: GraphManifest, op: SetVertexSemanticsOp
) -> None:
    """Attach, replace or clear the grounding on selected vertex types."""
    schema = manifest.graph_schema
    if schema is None:
        raise ValueError("set_vertex_semantics requires graph_schema")

    vertex_config = schema.core_schema.vertex_config
    unknown = sorted(set(op.semantics) - vertex_config.vertex_set)
    if unknown:
        raise ValueError(f"set_vertex_semantics: unknown vertices: {unknown}")

    for vertex in vertex_config.vertices:
        if vertex.name not in op.semantics:
            continue
        value = op.semantics[vertex.name]
        vertex.semantics = value.model_copy(deep=True) if value is not None else None

    schema.finish_init()