jmri_server.services.meta

 1"""Meta services — hello/ping/version/railroad/node/power.
 2
 3These don't manage beans; they expose server identity and one global piece
 4of state (track power per connection).
 5"""
 6from __future__ import annotations
 7
 8from typing import Any
 9
10from jmri_core.state import PowerState
11
12from jmri_server.service import JsonService, JsonServiceError
13from jmri_server.wire_states import power_from_wire, to_wire
14
15
16class HelloService(JsonService):
17    topic = "hello"
18
19    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
20        m = self.ctx.metadata
21        return {
22            "JMRI": m.jmri_version,
23            "json": m.json_protocol_version,
24            "version": ".".join(m.json_protocol_version.split(".")[:2]),
25            "heartbeat": m.heartbeat_ms,
26            "railroad": m.railroad,
27            "node": m.node,
28            "activeProfile": m.active_profile,
29        }
30
31
32class PingService(JsonService):
33    topic = "ping"
34
35    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
36        # JMRI returns pong as the response type at the envelope level, but
37        # the data slot for ping is empty. The dispatcher writes the envelope;
38        # we just return an empty object.
39        return {}
40
41
42class VersionService(JsonService):
43    topic = "version"
44
45    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
46        m = self.ctx.metadata
47        return {"JMRI": m.jmri_version, "version": m.json_protocol_version}
48
49
50class RailroadService(JsonService):
51    topic = "railroad"
52
53    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
54        return {"name": self.ctx.metadata.railroad}
55
56
57class NodeService(JsonService):
58    topic = "node"
59
60    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
61        return {"node": self.ctx.metadata.node, "formerNodes": []}
62
63
64class PowerService(JsonService):
65    """Track power for the (single) active connection."""
66
67    topic = "power"
68
69    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
70        return self._snapshot()
71
72    async def on_list(self, data: dict[str, Any] | None) -> list[dict[str, Any]]:
73        return [self._snapshot()]
74
75    async def on_post(self, data: dict[str, Any] | None) -> dict[str, Any]:
76        if data is None:
77            raise JsonServiceError(400, "power.post requires data")
78        if "state" in data:
79            await self.ctx.connection.power.set_power(power_from_wire(int(data["state"])))
80        return self._snapshot()
81
82    async def on_put(self, data: dict[str, Any] | None) -> dict[str, Any]:
83        return await self.on_post(data)
84
85    def _snapshot(self) -> dict[str, Any]:
86        conn = self.ctx.connection
87        return {
88            "name": conn.prefix,
89            "default": True,
90            "state": to_wire(
91                int(conn.power.state) if conn.power.state != PowerState.UNKNOWN else 1
92            ),
93        }