jmri_server.services.throttle

  1"""``ThrottleService`` — drives :class:`jmri_core.beans.throttle.Throttle`
  2instances via the active connection's :class:`ThrottleManager`.
  3
  4Wire shape per ``contracts/json-schema/throttle/*-server.json``:
  5
  6* ``name`` (required) — identity string for the throttle handle. We use
  7  the DCC address as a string by default.
  8* ``throttle`` — deprecated alias for ``name``; emitted on every response.
  9* ``address`` — DCC address (int).
 10* ``speed`` — -1.0..1.0 (number). -1 is emergency stop.
 11* ``forward`` — bool.
 12* ``F0``..``F28`` — booleans.
 13* ``release`` — null marker on response when the throttle was released.
 14
 15JSON method semantics:
 16
 17* ``put``  — acquire a throttle for ``address``.
 18* ``get``  — return the current state of the throttle named ``name``.
 19* ``post`` — apply mutations (speed/forward/F*).
 20* ``delete`` — release.
 21* ``list`` — return every active throttle.
 22"""
 23from __future__ import annotations
 24
 25from typing import Any
 26
 27from jmri_core.beans.throttle import NUM_FUNCTIONS, Throttle
 28from jmri_core.beans.throttle_manager import ThrottleManager
 29from jmri_core.state import ThrottleDirection
 30
 31from jmri_server.service import JsonService, JsonServiceError
 32
 33
 34def _throttle_name(throttle: Throttle) -> str:
 35    """Identity string the wire format uses for a throttle handle."""
 36    return str(throttle.address)
 37
 38
 39def _project(throttle: Throttle, *, released: bool = False) -> dict[str, Any]:
 40    name = _throttle_name(throttle)
 41    out: dict[str, Any] = {
 42        "name": name,
 43        "throttle": name,
 44        "address": throttle.address,
 45        "speed": throttle.speed,
 46        "forward": throttle.direction == ThrottleDirection.FORWARD,
 47        "speedSteps": 128,
 48        "clients": 1,
 49    }
 50    for i in range(min(NUM_FUNCTIONS, 29)):
 51        out[f"F{i}"] = throttle.functions[i]
 52    if released:
 53        out["release"] = None
 54    return out
 55
 56
 57def _resolve_name(throttles_mgr: ThrottleManager, data: dict[str, Any]) -> Throttle:
 58    """Find the throttle the client refers to.
 59
 60    Accepts either ``name`` or the deprecated ``throttle`` field. Returns
 61    the live ``Throttle`` instance.
 62    """
 63    name = data.get("name") or data.get("throttle")
 64    if name is None:
 65        raise JsonServiceError(400, "throttle request requires 'name' or 'throttle'")
 66    try:
 67        address = int(name)
 68    except (TypeError, ValueError):
 69        raise JsonServiceError(400, f"throttle name must be a DCC address, got {name!r}") from None
 70    throttle = throttles_mgr.get(address)
 71    if throttle is None:
 72        raise JsonServiceError(404, f"no throttle for address {address}")
 73    return throttle
 74
 75
 76class ThrottleService(JsonService):
 77    topic = "throttle"
 78
 79    @property
 80    def _throttles(self) -> Any:
 81        """The ThrottleManager on the active system connection."""
 82        return self.ctx.connection.throttles
 83
 84    async def on_get(self, data: dict[str, Any] | None) -> dict[str, Any]:
 85        throttle = _resolve_name(self._throttles, data or {})
 86        return _project(throttle)
 87
 88    async def on_list(self, data: dict[str, Any] | None) -> list[dict[str, Any]]:
 89        return [_project(t) for t in self._throttles]
 90
 91    async def on_put(self, data: dict[str, Any] | None) -> dict[str, Any]:
 92        if not data:
 93            raise JsonServiceError(400, "throttle.put requires data")
 94        # Accept either ``address`` directly or ``name`` as a stringified address.
 95        address = data.get("address")
 96        if address is None:
 97            name = data.get("name") or data.get("throttle")
 98            try:
 99                address = int(name) if name is not None else None
100            except (TypeError, ValueError):
101                address = None
102        if not isinstance(address, int):
103            raise JsonServiceError(400, "throttle.put requires 'address' or numeric 'name'")
104        long_address = bool(data.get("isLongAddress", False))
105        throttle = await self._throttles.acquire(address, long_address=long_address)
106        await _apply_mutations(throttle, data)
107        return _project(throttle)
108
109    async def on_post(self, data: dict[str, Any] | None) -> dict[str, Any]:
110        if not data:
111            raise JsonServiceError(400, "throttle.post requires data")
112        throttle = _resolve_name(self._throttles, data)
113        # ``release`` field (any value, including null) triggers release.
114        if "release" in data:
115            await self._throttles.release(throttle)
116            return _project(throttle, released=True)
117        await _apply_mutations(throttle, data)
118        return _project(throttle)
119
120    async def on_delete(self, data: dict[str, Any] | None) -> dict[str, Any]:
121        throttle = _resolve_name(self._throttles, data or {})
122        await self._throttles.release(throttle)
123        return _project(throttle, released=True)
124
125
126async def _apply_mutations(throttle: Throttle, data: dict[str, Any]) -> None:
127    if "speed" in data:
128        speed = float(data["speed"])
129        await throttle.set_speed(speed)
130    if "forward" in data:
131        direction = ThrottleDirection.FORWARD if data["forward"] else ThrottleDirection.REVERSE
132        await throttle.set_direction(direction)
133    for i in range(min(NUM_FUNCTIONS, 29)):
134        key = f"F{i}"
135        if key in data:
136            await throttle.set_function(i, bool(data[key]))