jmri_server.services.id_tag
IdTagService — JSON topic idTag.
Wire shape per jmri/server/json/idtag/idTag-server.json. Read-only
from the layout's perspective: the report_seen / report_unseen
transitions are driven by reporter beans, not by JSON post. But the
JSON contract does support a post that sets the whole state manually
(scripts occasionally do this for simulation), so we honour it.
1"""``IdTagService`` — JSON topic ``idTag``. 2 3Wire shape per ``jmri/server/json/idtag/idTag-server.json``. Read-only 4from the layout's perspective: the ``report_seen`` / ``report_unseen`` 5transitions are driven by reporter beans, not by JSON post. But the 6JSON contract does support a post that sets the whole state manually 7(scripts occasionally do this for simulation), so we honour it. 8""" 9from __future__ import annotations 10 11from typing import Any 12 13from jmri_core.beans.id_tag import ( 14 ID_TAG_STATE_SEEN, 15 ID_TAG_STATE_UNKNOWN, 16 ID_TAG_STATE_UNSEEN, 17 IdTag, 18 IdTagManager, 19) 20 21from jmri_server.service import JsonServiceError 22from jmri_server.services.beans import _BeanService, _common_named_bean 23 24 25class IdTagService(_BeanService): 26 topic = "idTag" 27 28 def manager(self) -> IdTagManager: 29 return self.ctx.connection.id_tags 30 31 def project(self, bean: IdTag) -> dict[str, Any]: 32 return _common_named_bean(bean) | { 33 "state": bean.state, 34 "reporter": bean.reporter_name, 35 "time": bean.time, 36 } 37 38 async def apply(self, bean: IdTag, data: dict[str, Any]) -> None: 39 # Scripts + tests can set state + reporter + time directly. The 40 # reporter reference is opaque (a bean name string); the JSON 41 # layer doesn't cross-check against the reporter manager 42 # because a report can name a reporter on a different profile. 43 if "state" in data: 44 state = int(data["state"]) 45 if state not in ( 46 ID_TAG_STATE_UNKNOWN, 47 ID_TAG_STATE_SEEN, 48 ID_TAG_STATE_UNSEEN, 49 ): 50 raise JsonServiceError( 51 400, f"invalid idTag state: {state}", 52 ) 53 when = str(data.get("time") or "") 54 reporter = data.get("reporter") 55 if state == ID_TAG_STATE_SEEN: 56 if not isinstance(reporter, str) or not reporter: 57 raise JsonServiceError( 58 400, 59 "idTag.state=SEEN requires a 'reporter' name", 60 ) 61 await bean.report_seen(reporter, when) 62 elif state == ID_TAG_STATE_UNSEEN: 63 await bean.report_unseen(when) 64 else: 65 # UNKNOWN: reset everything. 66 bean._state = state # noqa: SLF001 — internal reset 67 bean._reporter_name = None # noqa: SLF001 68 bean._time = None # noqa: SLF001 69 await bean.state_changed.emit(state)
26class IdTagService(_BeanService): 27 topic = "idTag" 28 29 def manager(self) -> IdTagManager: 30 return self.ctx.connection.id_tags 31 32 def project(self, bean: IdTag) -> dict[str, Any]: 33 return _common_named_bean(bean) | { 34 "state": bean.state, 35 "reporter": bean.reporter_name, 36 "time": bean.time, 37 } 38 39 async def apply(self, bean: IdTag, data: dict[str, Any]) -> None: 40 # Scripts + tests can set state + reporter + time directly. The 41 # reporter reference is opaque (a bean name string); the JSON 42 # layer doesn't cross-check against the reporter manager 43 # because a report can name a reporter on a different profile. 44 if "state" in data: 45 state = int(data["state"]) 46 if state not in ( 47 ID_TAG_STATE_UNKNOWN, 48 ID_TAG_STATE_SEEN, 49 ID_TAG_STATE_UNSEEN, 50 ): 51 raise JsonServiceError( 52 400, f"invalid idTag state: {state}", 53 ) 54 when = str(data.get("time") or "") 55 reporter = data.get("reporter") 56 if state == ID_TAG_STATE_SEEN: 57 if not isinstance(reporter, str) or not reporter: 58 raise JsonServiceError( 59 400, 60 "idTag.state=SEEN requires a 'reporter' name", 61 ) 62 await bean.report_seen(reporter, when) 63 elif state == ID_TAG_STATE_UNSEEN: 64 await bean.report_unseen(when) 65 else: 66 # UNKNOWN: reset everything. 67 bean._state = state # noqa: SLF001 — internal reset 68 bean._reporter_name = None # noqa: SLF001 69 bean._time = None # noqa: SLF001 70 await bean.state_changed.emit(state)
Base class encapsulating get/list/put/post/delete plumbing.
Concrete bean services set topic and override manager(),
project(), and (when mutations are supported) apply().
39 async def apply(self, bean: IdTag, data: dict[str, Any]) -> None: 40 # Scripts + tests can set state + reporter + time directly. The 41 # reporter reference is opaque (a bean name string); the JSON 42 # layer doesn't cross-check against the reporter manager 43 # because a report can name a reporter on a different profile. 44 if "state" in data: 45 state = int(data["state"]) 46 if state not in ( 47 ID_TAG_STATE_UNKNOWN, 48 ID_TAG_STATE_SEEN, 49 ID_TAG_STATE_UNSEEN, 50 ): 51 raise JsonServiceError( 52 400, f"invalid idTag state: {state}", 53 ) 54 when = str(data.get("time") or "") 55 reporter = data.get("reporter") 56 if state == ID_TAG_STATE_SEEN: 57 if not isinstance(reporter, str) or not reporter: 58 raise JsonServiceError( 59 400, 60 "idTag.state=SEEN requires a 'reporter' name", 61 ) 62 await bean.report_seen(reporter, when) 63 elif state == ID_TAG_STATE_UNSEEN: 64 await bean.report_unseen(when) 65 else: 66 # UNKNOWN: reset everything. 67 bean._state = state # noqa: SLF001 — internal reset 68 bean._reporter_name = None # noqa: SLF001 69 bean._time = None # noqa: SLF001 70 await bean.state_changed.emit(state)
Apply mutations from a client request. Default: no-op.