jmri_core.beans.throttle_manager

ThrottleManager — acquire/release throttle handles per DCC address.

Throttle objects aren't NamedBeans (they're transient holds on a slot in the command station), so they don't fit the regular Manager[T] shape. A ThrottleManager instead provides:

  • acquire(address, long_address) → Throttle — get an existing throttle for address if one is already held, or create a new one.
  • release(throttle) — let go of the throttle; subsequent acquire with the same address will create a fresh one.
  • iteration yields every active throttle.

Drivers override _construct() to return their hardware-aware Throttle subclass (e.g. LnThrottle). The base in-memory impl is used by InternalConnection and by tests.

 1"""``ThrottleManager`` — acquire/release throttle handles per DCC address.
 2
 3Throttle objects aren't NamedBeans (they're transient holds on a slot in
 4the command station), so they don't fit the regular ``Manager[T]`` shape.
 5A ``ThrottleManager`` instead provides:
 6
 7* ``acquire(address, long_address)`` → :class:`Throttle` — get an existing
 8  throttle for *address* if one is already held, or create a new one.
 9* ``release(throttle)`` — let go of the throttle; subsequent ``acquire``
10  with the same address will create a fresh one.
11* iteration yields every active throttle.
12
13Drivers override :meth:`_construct` to return their hardware-aware
14``Throttle`` subclass (e.g. ``LnThrottle``). The base in-memory impl is
15used by ``InternalConnection`` and by tests.
16"""
17from __future__ import annotations
18
19from collections.abc import Iterator
20
21from jmri_core.beans.throttle import Throttle
22from jmri_core.events import Signal
23
24
25class ThrottleManager:
26    """Acquire/release throttle handles by DCC address."""
27
28    def __init__(self, system_prefix: str) -> None:
29        if not system_prefix:
30            raise ValueError("system_prefix must be a non-empty string")
31        self.system_prefix = system_prefix
32        self._throttles: dict[int, Throttle] = {}
33        self.acquired: Signal[Throttle] = Signal()
34        self.released: Signal[Throttle] = Signal()
35
36    async def acquire(self, address: int, *, long_address: bool = False) -> Throttle:
37        if address in self._throttles:
38            return self._throttles[address]
39        throttle = await self._construct(address, long_address=long_address)
40        self._throttles[address] = throttle
41        await self.acquired.emit(throttle)
42        return throttle
43
44    async def release(self, throttle: Throttle) -> None:
45        existing = self._throttles.get(throttle.address)
46        if existing is not throttle:
47            return
48        del self._throttles[throttle.address]
49        await self.released.emit(throttle)
50
51    def get(self, address: int) -> Throttle | None:
52        return self._throttles.get(address)
53
54    def __iter__(self) -> Iterator[Throttle]:
55        return iter(self._throttles.values())
56
57    def __len__(self) -> int:
58        return len(self._throttles)
59
60    async def _construct(self, address: int, *, long_address: bool) -> Throttle:
61        """Build the throttle for *address*. Driver hook."""
62        return Throttle(address=address, long_address=long_address)
class ThrottleManager: 
26class ThrottleManager:
27    """Acquire/release throttle handles by DCC address."""
28
29    def __init__(self, system_prefix: str) -> None:
30        if not system_prefix:
31            raise ValueError("system_prefix must be a non-empty string")
32        self.system_prefix = system_prefix
33        self._throttles: dict[int, Throttle] = {}
34        self.acquired: Signal[Throttle] = Signal()
35        self.released: Signal[Throttle] = Signal()
36
37    async def acquire(self, address: int, *, long_address: bool = False) -> Throttle:
38        if address in self._throttles:
39            return self._throttles[address]
40        throttle = await self._construct(address, long_address=long_address)
41        self._throttles[address] = throttle
42        await self.acquired.emit(throttle)
43        return throttle
44
45    async def release(self, throttle: Throttle) -> None:
46        existing = self._throttles.get(throttle.address)
47        if existing is not throttle:
48            return
49        del self._throttles[throttle.address]
50        await self.released.emit(throttle)
51
52    def get(self, address: int) -> Throttle | None:
53        return self._throttles.get(address)
54
55    def __iter__(self) -> Iterator[Throttle]:
56        return iter(self._throttles.values())
57
58    def __len__(self) -> int:
59        return len(self._throttles)
60
61    async def _construct(self, address: int, *, long_address: bool) -> Throttle:
62        """Build the throttle for *address*. Driver hook."""
63        return Throttle(address=address, long_address=long_address)

Acquire/release throttle handles by DCC address.

ThrottleManager(system_prefix: str) 
29    def __init__(self, system_prefix: str) -> None:
30        if not system_prefix:
31            raise ValueError("system_prefix must be a non-empty string")
32        self.system_prefix = system_prefix
33        self._throttles: dict[int, Throttle] = {}
34        self.acquired: Signal[Throttle] = Signal()
35        self.released: Signal[Throttle] = Signal()
system_prefix
async def acquire( self, address: int, *, long_address: bool = False) -> jmri_core.beans.Throttle: 
37    async def acquire(self, address: int, *, long_address: bool = False) -> Throttle:
38        if address in self._throttles:
39            return self._throttles[address]
40        throttle = await self._construct(address, long_address=long_address)
41        self._throttles[address] = throttle
42        await self.acquired.emit(throttle)
43        return throttle
async def release(self, throttle: jmri_core.beans.Throttle) -> None: 
45    async def release(self, throttle: Throttle) -> None:
46        existing = self._throttles.get(throttle.address)
47        if existing is not throttle:
48            return
49        del self._throttles[throttle.address]
50        await self.released.emit(throttle)
def get(self, address: int) -> jmri_core.beans.Throttle | None: 
52    def get(self, address: int) -> Throttle | None:
53        return self._throttles.get(address)