jmri_server.services.oblock
OBlockService — JSON topic oblock.
Wire shape per jmri/server/json/oblock/oblock-server.json.
Round-trip only in this pass. Warrants + portals + paths come with
Panel Editor (Tier 1 #3).
1"""``OBlockService`` — JSON topic ``oblock``. 2 3Wire shape per ``jmri/server/json/oblock/oblock-server.json``. 4Round-trip only in this pass. Warrants + portals + paths come with 5Panel Editor (Tier 1 #3). 6""" 7from __future__ import annotations 8 9from typing import Any 10 11from jmri_core.beans.oblock import OBlock, OBlockManager 12 13from jmri_server.services.beans import _BeanService, _common_named_bean 14 15 16class OBlockService(_BeanService): 17 topic = "oblock" 18 19 def manager(self) -> OBlockManager: 20 return self.ctx.connection.oblocks 21 22 def project(self, bean: OBlock) -> dict[str, Any]: 23 return _common_named_bean(bean) | { 24 "status": bean.status, 25 "value": bean.value, 26 "sensor": bean.sensor_name, 27 "warrant": bean.warrant_name, 28 "train": bean.train_name, 29 } 30 31 async def apply(self, bean: OBlock, data: dict[str, Any]) -> None: 32 if "value" in data: 33 v = data["value"] 34 bean.value = str(v) if v is not None else None 35 if "sensor" in data: 36 v = data["sensor"] 37 bean.sensor_name = str(v) if v is not None else None 38 if "warrant" in data: 39 v = data["warrant"] 40 bean.warrant_name = str(v) if v is not None else None 41 if "train" in data: 42 v = data["train"] 43 bean.train_name = str(v) if v is not None else None 44 if "status" in data: 45 await bean.set_status(int(data["status"]))
17class OBlockService(_BeanService): 18 topic = "oblock" 19 20 def manager(self) -> OBlockManager: 21 return self.ctx.connection.oblocks 22 23 def project(self, bean: OBlock) -> dict[str, Any]: 24 return _common_named_bean(bean) | { 25 "status": bean.status, 26 "value": bean.value, 27 "sensor": bean.sensor_name, 28 "warrant": bean.warrant_name, 29 "train": bean.train_name, 30 } 31 32 async def apply(self, bean: OBlock, data: dict[str, Any]) -> None: 33 if "value" in data: 34 v = data["value"] 35 bean.value = str(v) if v is not None else None 36 if "sensor" in data: 37 v = data["sensor"] 38 bean.sensor_name = str(v) if v is not None else None 39 if "warrant" in data: 40 v = data["warrant"] 41 bean.warrant_name = str(v) if v is not None else None 42 if "train" in data: 43 v = data["train"] 44 bean.train_name = str(v) if v is not None else None 45 if "status" in data: 46 await bean.set_status(int(data["status"]))
Base class encapsulating get/list/put/post/delete plumbing.
Concrete bean services set topic and override manager(),
project(), and (when mutations are supported) apply().
32 async def apply(self, bean: OBlock, data: dict[str, Any]) -> None: 33 if "value" in data: 34 v = data["value"] 35 bean.value = str(v) if v is not None else None 36 if "sensor" in data: 37 v = data["sensor"] 38 bean.sensor_name = str(v) if v is not None else None 39 if "warrant" in data: 40 v = data["warrant"] 41 bean.warrant_name = str(v) if v is not None else None 42 if "train" in data: 43 v = data["train"] 44 bean.train_name = str(v) if v is not None else None 45 if "status" in data: 46 await bean.set_status(int(data["status"]))
Apply mutations from a client request. Default: no-op.