jmri_ncemonitor.parity9bit
9-bit-via-parity → high-bit-marker adapter.
Some NCE Cab Bus adapters carry the address-vs-data distinction on the 9th bit but expose it to userland as mark/space parity rather than as the high bit of the received byte. Operators who go that route need to feed the codec (which expects the high-bit-marker convention) a translated stream.
This module provides Parity9BitDecoder, a
~jmri_ncemonitor.transport.port.BytePort-compatible wrapper
that takes a parity-tagged byte source and produces the standard
high-bit-marker byte stream.
Usage sketch::
async def read_tagged() -> tuple[int, bool]:
# ... driver-specific parity-aware read
return byte_value, is_address_byte
port = Parity9BitDecoder(read_tagged, base_port_for_close)
monitor = CabBusMonitor(port)
Kept in a separate module so the pyserial branch (which does the raw parity-aware read) can live behind an optional dependency without hurting the pure decoder logic.
1"""9-bit-via-parity → high-bit-marker adapter. 2 3Some NCE Cab Bus adapters carry the address-vs-data distinction on 4the 9th bit but expose it to userland as mark/space parity rather 5than as the high bit of the received byte. Operators who go that 6route need to feed the codec (which expects the high-bit-marker 7convention) a translated stream. 8 9This module provides :class:`Parity9BitDecoder`, a 10:class:`~jmri_ncemonitor.transport.port.BytePort`-compatible wrapper 11that takes a parity-tagged byte source and produces the standard 12high-bit-marker byte stream. 13 14Usage sketch:: 15 16 async def read_tagged() -> tuple[int, bool]: 17 # ... driver-specific parity-aware read 18 return byte_value, is_address_byte 19 20 port = Parity9BitDecoder(read_tagged, base_port_for_close) 21 monitor = CabBusMonitor(port) 22 23Kept in a separate module so the pyserial branch (which does the 24raw parity-aware read) can live behind an optional dependency 25without hurting the pure decoder logic. 26""" 27from __future__ import annotations 28 29from collections.abc import Awaitable, Callable 30 31from jmri_ncemonitor.transport.port import BytePort 32 33 34class Parity9BitDecoder(BytePort): 35 """Translate a parity-tagged byte stream into the high-bit-marker form. 36 37 Constructor takes: 38 39 * ``read_tagged`` — awaitable returning ``(byte_value, is_address)``. 40 ``byte_value`` is the 7-bit payload; ``is_address`` indicates 41 whether this byte was received with mark parity (or however the 42 operator's driver signals address-ness). 43 * ``base_port`` — the underlying byte port. ``aclose`` is 44 delegated so lifecycle stays sane; ``read_byte`` is overridden 45 by this class. 46 47 The returned bytes carry the high bit set when ``is_address`` is 48 true, matching :func:`jmri_ncemonitor.codec.is_address_byte`. 49 """ 50 51 def __init__( 52 self, 53 read_tagged: Callable[[], Awaitable[tuple[int, bool]]], 54 base_port: BytePort, 55 ) -> None: 56 self._read_tagged = read_tagged 57 self._base = base_port 58 59 async def read_byte(self) -> int: 60 value, is_address = await self._read_tagged() 61 # Force 7-bit payload; some drivers pass through a full byte 62 # that already has bit 7 clear/set — we mask so the semantic 63 # comes only from ``is_address``. 64 value &= 0x7F 65 return (value | 0x80) if is_address else value 66 67 async def write(self, data: bytes) -> None: # pragma: no cover - passive 68 raise RuntimeError( 69 "NCE Cab Bus monitor is passive — writes are not supported" 70 ) 71 72 async def aclose(self) -> None: 73 await self._base.aclose() 74 75 @property 76 def is_open(self) -> bool: 77 return self._base.is_open
35class Parity9BitDecoder(BytePort): 36 """Translate a parity-tagged byte stream into the high-bit-marker form. 37 38 Constructor takes: 39 40 * ``read_tagged`` — awaitable returning ``(byte_value, is_address)``. 41 ``byte_value`` is the 7-bit payload; ``is_address`` indicates 42 whether this byte was received with mark parity (or however the 43 operator's driver signals address-ness). 44 * ``base_port`` — the underlying byte port. ``aclose`` is 45 delegated so lifecycle stays sane; ``read_byte`` is overridden 46 by this class. 47 48 The returned bytes carry the high bit set when ``is_address`` is 49 true, matching :func:`jmri_ncemonitor.codec.is_address_byte`. 50 """ 51 52 def __init__( 53 self, 54 read_tagged: Callable[[], Awaitable[tuple[int, bool]]], 55 base_port: BytePort, 56 ) -> None: 57 self._read_tagged = read_tagged 58 self._base = base_port 59 60 async def read_byte(self) -> int: 61 value, is_address = await self._read_tagged() 62 # Force 7-bit payload; some drivers pass through a full byte 63 # that already has bit 7 clear/set — we mask so the semantic 64 # comes only from ``is_address``. 65 value &= 0x7F 66 return (value | 0x80) if is_address else value 67 68 async def write(self, data: bytes) -> None: # pragma: no cover - passive 69 raise RuntimeError( 70 "NCE Cab Bus monitor is passive — writes are not supported" 71 ) 72 73 async def aclose(self) -> None: 74 await self._base.aclose() 75 76 @property 77 def is_open(self) -> bool: 78 return self._base.is_open
Translate a parity-tagged byte stream into the high-bit-marker form.
Constructor takes:
read_tagged— awaitable returning(byte_value, is_address).byte_valueis the 7-bit payload;is_addressindicates whether this byte was received with mark parity (or however the operator's driver signals address-ness).base_port— the underlying byte port.acloseis delegated so lifecycle stays sane;read_byteis overridden by this class.
The returned bytes carry the high bit set when is_address is
true, matching jmri_ncemonitor.codec.is_address_byte().
Parity9BitDecoder( read_tagged: Callable[[], Awaitable[tuple[int, bool]]], base_port: jmri_ncemonitor.BytePort)
async def
read_byte(self) -> int:
60 async def read_byte(self) -> int: 61 value, is_address = await self._read_tagged() 62 # Force 7-bit payload; some drivers pass through a full byte 63 # that already has bit 7 clear/set — we mask so the semantic 64 # comes only from ``is_address``. 65 value &= 0x7F 66 return (value | 0x80) if is_address else value