Skip to content

serializer

serializer #

Flatten a Procedure DAG to the cross-platform wire form.

Three steps: walk the DAG depth-first, giving each node a short hex id; ask the loop which of those nodes it clones per iteration (loop_templates); emit {"schema", "root", "nodes"}:

  • schema is the target Struct's wire form — the a-priori data schema, emitted once, that replaces the runtime manifest.
  • root is the id of the root node; nodes maps id → node.
  • each node records its kind (class name), return_type, label, settings, inputs (field name → id), and substeps (list of ids); falsy entries are omitted. Path/Void return types serialize compactly ({"type": "path"} / {"type": "void"}) since their structure is recoverable from schema plus the access chain.

procedure_to_dict(procedure, target_schema=None) #

Serialize a Procedure DAG to {"schema"?, "root", "nodes"}.

Parameters:

Name Type Description Default
procedure Procedure[CGType]

The root Procedure node to serialize.

 required
target_schema type[CGType] | None

The target Struct whose wire form is embedded as "schema" (omitted when not provided).

 None

Returns:

Type Description
dict[str, Any]

The serialized procedure graph.

Raises:

Type Description
ValueError

If a ProcedureWhile violates loop invariants.
Source code in capturegraph-lib/capturegraph/procedures/procedure/exporting/serializer.py 
 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
def procedure_to_dict(
    procedure: Procedure[CGType],
    target_schema: type[CGType] | None = None,
) -> dict[str, Any]:
    """Serialize a Procedure DAG to ``{"schema"?, "root", "nodes"}``.

    Args:
        procedure: The root Procedure node to serialize.
        target_schema: The target ``Struct`` whose wire form is embedded as
            ``"schema"`` (omitted when not provided).

    Returns:
        The serialized procedure graph.

    Raises:
        ValueError: If a ProcedureWhile violates loop invariants.
    """
    graph = _reachable(procedure)
    ids = _short_ids(graph)
    nodes = {ids[uuid]: _node_json(node, ids) for uuid, node in graph.items()}

    for loop_uuid, template in loop_templates(graph).items():
        settings = nodes[ids[loop_uuid]].setdefault("settings", {})
        settings["template_nodes"] = sorted(ids[uuid] for uuid in template)

    result: dict[str, Any] = {}
    if target_schema is not None:
        result["schema"] = schema(target_schema)
    result["root"] = ids[procedure.uuid]
    result["nodes"] = nodes
    return result

procedure_to_json(procedure, indent=4, target_schema=None, **json_kwargs) #

Serialize a Procedure DAG to a JSON string (see [procedure_to_dict][]).

Source code in capturegraph-lib/capturegraph/procedures/procedure/exporting/serializer.py 
129
130
131
132
133
134
135
136
137
138
139
140
def procedure_to_json(
    procedure: Procedure[CGType],
    indent: int = 4,
    target_schema: type[CGType] | None = None,
    **json_kwargs: Unpack[_JSONDumpsKwargs],
) -> str:
    """Serialize a Procedure DAG to a JSON string (see [procedure_to_dict][])."""
    return json.dumps(
        procedure_to_dict(procedure, target_schema=target_schema),
        indent=indent,
        **json_kwargs,
    )