jmri_server.services.signal_system
SignalSystemService — JSON topic signalSystem.
Read-only enumeration of the JMRI signal systems parsed by
~jmri_core.persistence.signal_systems.SignalSystemLoader.
Mobile clients use list to populate a picker; the panel viewer
uses get to look up an individual system's aspect table.
Wire shape::
{
"id": "basic", // directory name under xml/signals
"name": "Basic Model Signals", // <name> from aspects.xml
"reference": "...", // <reference> from aspects.xml
"aspects": [
{"name": "Clear", "indication": "Proceed",
"speed": "Normal", "speed2": "Normal", "route": ""},
...
]
}
No list of systems is 503 unless the loader itself couldn't be
constructed (which shouldn't happen). Callers configuring the env
var wrongly get an empty list back — the /signalsystems web
page renders a hint in that case.
1"""``SignalSystemService`` — JSON topic ``signalSystem``. 2 3Read-only enumeration of the JMRI signal systems parsed by 4:class:`~jmri_core.persistence.signal_systems.SignalSystemLoader`. 5Mobile clients use ``list`` to populate a picker; the panel viewer 6uses ``get`` to look up an individual system's aspect table. 7 8Wire shape:: 9 10 { 11 "id": "basic", // directory name under xml/signals 12 "name": "Basic Model Signals", // <name> from aspects.xml 13 "reference": "...", // <reference> from aspects.xml 14 "aspects": [ 15 {"name": "Clear", "indication": "Proceed", 16 "speed": "Normal", "speed2": "Normal", "route": ""}, 17 ... 18 ] 19 } 20 21No ``list`` of systems is 503 unless the loader itself couldn't be 22constructed (which shouldn't happen). Callers configuring the env 23var wrongly get an empty list back — the ``/signalsystems`` web 24page renders a hint in that case. 25""" 26from __future__ import annotations 27 28from typing import Any 29 30from jmri_core.env import signal_xml_root 31from jmri_core.persistence.signal_systems import ( 32 SignalSystem, 33 SignalSystemLoader, 34) 35 36from jmri_server.service import JsonService, JsonServiceError 37 38 39def _project(system: SignalSystem) -> dict[str, Any]: 40 return { 41 "id": system.id, 42 "name": system.name, 43 "reference": system.reference, 44 "aspects": [ 45 { 46 "name": a.name, 47 "indication": a.indication, 48 "speed": a.speed, 49 "speed2": a.speed2, 50 "route": a.route, 51 } 52 for a in system.aspects 53 ], 54 } 55 56 57def _loader(ctx: Any) -> SignalSystemLoader: 58 """Lazy singleton on the ctx. Constructed the first time a 59 caller asks for it; env-driven root path so operators don't have 60 to edit code to point at their JMRI install.""" 61 existing = getattr(ctx, "_signal_system_loader", None) 62 if isinstance(existing, SignalSystemLoader): 63 return existing 64 root = getattr(ctx, "_signal_xml_root", None) or signal_xml_root() 65 loader = SignalSystemLoader(root) 66 ctx._signal_system_loader = loader # type: ignore[attr-defined] 67 return loader 68 69 70class SignalSystemService(JsonService): 71 topic = "signalSystem" 72 73 async def on_list( 74 self, data: dict[str, Any] | None, 75 ) -> list[dict[str, Any]]: 76 return [_project(s) for s in _loader(self.ctx).systems()] 77 78 async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]: 79 data = data or {} 80 system_id = data.get("id") or data.get("name") 81 if not isinstance(system_id, str) or not system_id: 82 raise JsonServiceError( 83 400, "signalSystem.get requires 'id' (or legacy 'name')", 84 ) 85 system = _loader(self.ctx).get(system_id) 86 if system is None: 87 raise JsonServiceError( 88 404, f"no signal system with id {system_id!r}", 89 ) 90 return _project(system)
71class SignalSystemService(JsonService): 72 topic = "signalSystem" 73 74 async def on_list( 75 self, data: dict[str, Any] | None, 76 ) -> list[dict[str, Any]]: 77 return [_project(s) for s in _loader(self.ctx).systems()] 78 79 async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]: 80 data = data or {} 81 system_id = data.get("id") or data.get("name") 82 if not isinstance(system_id, str) or not system_id: 83 raise JsonServiceError( 84 400, "signalSystem.get requires 'id' (or legacy 'name')", 85 ) 86 system = _loader(self.ctx).get(system_id) 87 if system is None: 88 raise JsonServiceError( 89 404, f"no signal system with id {system_id!r}", 90 ) 91 return _project(system)
Base class for a topic's service handlers.
Subclasses set topic and override the methods they support;
unsupported methods raise 405 by default. Subclasses can also
override REQUIRED_ROLE to change per-method role gates.
async def
on_get(self, data: dict[str, typing.Any] | None) -> dict[str, typing.Any]:
79 async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]: 80 data = data or {} 81 system_id = data.get("id") or data.get("name") 82 if not isinstance(system_id, str) or not system_id: 83 raise JsonServiceError( 84 400, "signalSystem.get requires 'id' (or legacy 'name')", 85 ) 86 system = _loader(self.ctx).get(system_id) 87 if system is None: 88 raise JsonServiceError( 89 404, f"no signal system with id {system_id!r}", 90 ) 91 return _project(system)