jmri_logixng.serialize

  1"""YAML/JSON serialization for LogixNG trees.
  2
  3The on-disk format is a nested dict tree where every node has a
  4``type:`` tag matched against :mod:`jmri_logixng.registry`. Each
  5concrete class implements :meth:`extra_fields` to declare its own
  6serialized fields; deserialization runs the inverse via a small,
  7explicit decoder per node class.
  8
  9Why explicit decoders and not reflection? Because actions/expressions
 10hold *typed* children (``Action``, ``Expression[bool]``, ``TurnoutState``,
 11…) that need careful reconstruction, not blind kwarg passing.
 12"""
 13from __future__ import annotations
 14
 15import json
 16import pathlib
 17from typing import Any
 18
 19import yaml
 20from jmri_core.state import LightState, SensorState, TurnoutState
 21
 22from jmri_logixng.actions.digital import (
 23    IfThenElse,
 24    LogAction,
 25    SetLightState,
 26    SetMemory,
 27    SetSensorState,
 28    SetTurnoutState,
 29)
 30from jmri_logixng.actions.flow import Delay, Sequence
 31from jmri_logixng.expressions.digital import (
 32    And,
 33    ConstantBool,
 34    LightIsOn,
 35    MemoryEquals,
 36    Not,
 37    Or,
 38    SensorIsActive,
 39    TurnoutIsClosed,
 40)
 41from jmri_logixng.logix import ConditionalNG, LogixNG
 42from jmri_logixng.tree import Action, Expression, Node
 43
 44
 45def to_yaml(node: Node) -> str:
 46    """Serialize *node* to a YAML string. Lossless."""
 47    return yaml.safe_dump(node.to_dict(), sort_keys=False)
 48
 49
 50def to_json(node: Node, *, indent: int | None = 2) -> str:
 51    """Serialize *node* to a JSON string."""
 52    return json.dumps(node.to_dict(), indent=indent)
 53
 54
 55def save_yaml(node: Node, path: str | pathlib.Path) -> None:
 56    pathlib.Path(path).write_text(to_yaml(node), encoding="utf-8")
 57
 58
 59def load_yaml(path: str | pathlib.Path) -> Node:
 60    return from_dict(yaml.safe_load(pathlib.Path(path).read_text(encoding="utf-8")))
 61
 62
 63def from_yaml(text: str) -> Node:
 64    return from_dict(yaml.safe_load(text))
 65
 66
 67def from_json(text: str) -> Node:
 68    return from_dict(json.loads(text))
 69
 70
 71# ---- decoders --------------------------------------------------------------
 72
 73
 74def from_dict(payload: dict[str, Any]) -> Node:
 75    """Dispatch on ``type`` to the right decoder."""
 76    tag = payload.get("type")
 77    if not isinstance(tag, str):
 78        raise ValueError(f"serialized node missing 'type' tag: {payload!r}")
 79    decoder = _DECODERS.get(tag)
 80    if decoder is None:
 81        raise KeyError(f"no decoder for tag {tag!r}")
 82    node = decoder(payload)
 83    assert isinstance(node, Node)
 84    return node
 85
 86
 87def _node_kwargs(payload: dict[str, Any]) -> dict[str, Any]:
 88    """Common ``user_name`` + ``node_id`` extraction."""
 89    return {
 90        "user_name": payload.get("user_name", ""),
 91        "node_id": payload.get("id"),
 92    }
 93
 94
 95def _decode_conditional(p: dict[str, Any]) -> ConditionalNG:
 96    return ConditionalNG(
 97        actions=[_decode_action(a) for a in p.get("actions", [])],
 98        enabled=p.get("enabled", True),
 99        **_node_kwargs(p),
100    )
101
102
103def _decode_logixng(p: dict[str, Any]) -> LogixNG:
104    return LogixNG(
105        conditionals=[_decode_conditional(c) for c in p.get("conditionals", [])],
106        enabled=p.get("enabled", True),
107        **_node_kwargs(p),
108    )
109
110
111# ---- action decoders ------------------------------------------------------
112
113
114def _decode_action(p: dict[str, Any]) -> Action:
115    node = from_dict(p)
116    if not isinstance(node, Action):
117        raise TypeError(f"expected Action, got {type(node).__name__}")
118    return node
119
120
121def _decode_expression(p: dict[str, Any]) -> Expression[Any]:
122    node = from_dict(p)
123    if not isinstance(node, Expression):
124        raise TypeError(f"expected Expression, got {type(node).__name__}")
125    return node
126
127
128def _decode_set_turnout(p: dict[str, Any]) -> SetTurnoutState:
129    return SetTurnoutState(
130        turnout=p["turnout"], state=TurnoutState[p["state"]],
131        **_node_kwargs(p),
132    )
133
134
135def _decode_set_sensor(p: dict[str, Any]) -> SetSensorState:
136    return SetSensorState(
137        sensor=p["sensor"], state=SensorState[p["state"]],
138        **_node_kwargs(p),
139    )
140
141
142def _decode_set_light(p: dict[str, Any]) -> SetLightState:
143    return SetLightState(
144        light=p["light"], state=LightState[p["state"]],
145        **_node_kwargs(p),
146    )
147
148
149def _decode_set_memory(p: dict[str, Any]) -> SetMemory:
150    return SetMemory(memory=p["memory"], value=p.get("value"), **_node_kwargs(p))
151
152
153def _decode_if_then(p: dict[str, Any]) -> IfThenElse:
154    else_payload = p.get("else")
155    return IfThenElse(
156        condition=_decode_expression(p["condition"]),
157        then_action=_decode_action(p["then"]),
158        else_action=_decode_action(else_payload) if else_payload else None,
159        **_node_kwargs(p),
160    )
161
162
163def _decode_log(p: dict[str, Any]) -> LogAction:
164    return LogAction(
165        message=p.get("message", ""), level=p.get("level", "info"),
166        **_node_kwargs(p),
167    )
168
169
170def _decode_sequence(p: dict[str, Any]) -> Sequence:
171    return Sequence(
172        actions=[_decode_action(a) for a in p.get("actions", [])],
173        **_node_kwargs(p),
174    )
175
176
177def _decode_delay(p: dict[str, Any]) -> Delay:
178    return Delay(seconds=float(p["seconds"]), **_node_kwargs(p))
179
180
181# ---- expression decoders --------------------------------------------------
182
183
184def _decode_const_bool(p: dict[str, Any]) -> ConstantBool:
185    return ConstantBool(value=bool(p["value"]), **_node_kwargs(p))
186
187
188def _decode_sensor_active(p: dict[str, Any]) -> SensorIsActive:
189    return SensorIsActive(sensor=p["sensor"], **_node_kwargs(p))
190
191
192def _decode_turnout_closed(p: dict[str, Any]) -> TurnoutIsClosed:
193    return TurnoutIsClosed(turnout=p["turnout"], **_node_kwargs(p))
194
195
196def _decode_light_on(p: dict[str, Any]) -> LightIsOn:
197    return LightIsOn(light=p["light"], **_node_kwargs(p))
198
199
200def _decode_memory_equals(p: dict[str, Any]) -> MemoryEquals:
201    return MemoryEquals(memory=p["memory"], value=p.get("value"), **_node_kwargs(p))
202
203
204def _decode_and(p: dict[str, Any]) -> And:
205    return And(
206        children=[_decode_expression(c) for c in p.get("children", [])],
207        **_node_kwargs(p),
208    )
209
210
211def _decode_or(p: dict[str, Any]) -> Or:
212    return Or(
213        children=[_decode_expression(c) for c in p.get("children", [])],
214        **_node_kwargs(p),
215    )
216
217
218def _decode_not(p: dict[str, Any]) -> Not:
219    return Not(child=_decode_expression(p["child"]), **_node_kwargs(p))
220
221
222# Tag → decoder dispatch table.
223_DECODERS: dict[str, Any] = {
224    "logixng:conditional": _decode_conditional,
225    "logixng:root": _decode_logixng,
226    "action:set-turnout": _decode_set_turnout,
227    "action:set-sensor": _decode_set_sensor,
228    "action:set-light": _decode_set_light,
229    "action:set-memory": _decode_set_memory,
230    "action:if-then-else": _decode_if_then,
231    "action:log": _decode_log,
232    "action:sequence": _decode_sequence,
233    "action:delay": _decode_delay,
234    "expr:const-bool": _decode_const_bool,
235    "expr:sensor-active": _decode_sensor_active,
236    "expr:turnout-closed": _decode_turnout_closed,
237    "expr:light-on": _decode_light_on,
238    "expr:memory-equals": _decode_memory_equals,
239    "expr:and": _decode_and,
240    "expr:or": _decode_or,
241    "expr:not": _decode_not,
242}