sartoriuslib.transport¶

def add_script(self, command: bytes, reply: ScriptedReply) -> None:
    """Register or overwrite a scripted reply for ``command``."""
    self._script[bytes(command)] = reply

def feed(self, data: bytes) -> None:
    """Push unsolicited bytes into the read buffer.

    Useful for simulating a device that was left in SBI autoprint
    mode, or garbage on the line that the session must drain on
    recovery.
    """
    self._read_buffer.extend(data)

def force_read_timeout(self, enabled: bool = True) -> None:
    """Force the next read to raise ``SartoriusTimeoutError``."""
    self._force_read_timeout = enabled

def force_reopen_error(self, enabled: bool = True) -> None:
    """Force the next :meth:`reopen` to raise ``SartoriusConnectionError``."""
    self._force_reopen_error = enabled

def force_write_timeout(self, enabled: bool = True) -> None:
    """Force the next :meth:`write` to raise ``SartoriusTimeoutError``."""
    self._force_write_timeout = enabled

async def reopen(
    self,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
) -> None:
    """Simulate a serial-setting change — close, record, reopen.

    If ``force_reopen_error()`` has been called the reopen raises
    :class:`SartoriusConnectionError` after the close, leaving the
    transport closed. That is the "reopen wedged" path used by
    :meth:`Balance.configure_protocol` tests for the BROKEN-state
    transition.
    """
    await self.close()
    self._reopen_count += 1
    if baudrate is not None:
        self._last_reopen_baud = baudrate
    if parity is not None:
        self._last_reopen_parity = parity
    if stopbits is not None:
        self._last_reopen_stopbits = stopbits
    if self._force_reopen_error:
        raise SartoriusConnectionError(
            f"forced reopen error on {self._label}",
            context=ErrorContext(port=self._label),
        )
    await self.open()

async def reopen(
    self,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
) -> None:
    """Close and reopen the port with any subset of overrides.

    Any argument left as ``None`` keeps the existing setting. Used for
    the WZA SBI→xBPI flip (swaps both baud and parity) and for
    ``set_baud_rate`` retuning after the device has already switched.
    The cached :class:`SerialSettings` is updated so the new settings
    survive subsequent close / open round-trips.

    If :meth:`open` fails on the new settings the transport is left
    closed; the caller is responsible for surfacing that as a
    ``BROKEN`` session state with recovery guidance.
    """
    await self.close()
    new_settings = self._settings
    if baudrate is not None:
        new_settings = replace(new_settings, baudrate=baudrate)
    if parity is not None:
        new_settings = replace(new_settings, parity=parity)
    if stopbits is not None:
        new_settings = replace(new_settings, stopbits=stopbits)
    self._settings = new_settings
    await self.open()

async def close(self) -> None:
    """Close the underlying port. Safe to call when already closed."""
    ...

async def drain_input(self) -> None:
    """Discard any buffered input bytes. Best-effort; never raises."""
    ...

async def open(self) -> None:
    """Open the underlying port. Idempotent re-calls are an error."""
    ...

async def read_available(
    self,
    *,
    idle_timeout: float,
    max_bytes: int | None = None,
) -> bytes:
    """Read until the line goes idle for ``idle_timeout`` seconds.

    Never raises on idle expiry — an idle timeout is the *expected*
    exit. Returns whatever was accumulated (possibly empty). Used
    for best-effort drain, passive SBI autoprint sniffing during
    protocol detection, and stream-stop recovery.
    """
    ...

async def read_exact(self, n: int, *, timeout: float) -> bytes:
    """Read exactly ``n`` bytes.

    The canonical shape for xBPI framing: read one byte to discover
    the length, then read that many bytes. Raises
    :class:`sartoriuslib.errors.SartoriusTimeoutError` if fewer than
    ``n`` bytes arrive before ``timeout``. Partial buffers are retained
    for the next call — implementations must not discard them.
    """
    ...

async def read_until(self, separator: bytes, *, timeout: float) -> bytes:
    """Read bytes up to and including the next occurrence of ``separator``.

    Raises :class:`sartoriuslib.errors.SartoriusTimeoutError` if the
    separator does not arrive before ``timeout``. Bytes received
    after the separator remain buffered for the next call —
    implementations must not discard them.
    """
    ...

async def reopen(
    self,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
) -> None:
    """Close and re-open the port, optionally changing serial framing.

    Used by :meth:`Balance.configure_protocol` for the WZA SBI→xBPI
    flip, which swaps both baud rate and parity, and by operations
    like ``set_baud_rate`` that retune the port after the device has
    already switched mid-sequence. Any argument left as ``None`` keeps
    the existing value.

    Implementations must leave the transport in a consistent state:
    either fully reopened with the new settings, or clearly closed so
    callers can recognise a failure. Silent partial states are the
    worst failure mode for this method.

    Non-serial transports (e.g. a future TCP adapter) may raise
    :class:`NotImplementedError` — baud/parity don't apply there.
    """
    ...

async def write(self, data: bytes, *, timeout: float) -> None:
    """Write every byte of ``data``.

    Raises :class:`sartoriuslib.errors.SartoriusTimeoutError` on
    expiry. A bounded write timeout is mandatory because sends can
    block on RS-485 hardware flow control or a stuck device. Callers
    that block indefinitely hide real hangs.
    """
    ...

async def close(self) -> None:
    """Close the underlying port. Safe to call when already closed."""
    ...

async def drain_input(self) -> None:
    """Discard any buffered input bytes. Best-effort; never raises."""
    ...

async def open(self) -> None:
    """Open the underlying port. Idempotent re-calls are an error."""
    ...

async def read_available(
    self,
    *,
    idle_timeout: float,
    max_bytes: int | None = None,
) -> bytes:
    """Read until the line goes idle for ``idle_timeout`` seconds.

    Never raises on idle expiry — an idle timeout is the *expected*
    exit. Returns whatever was accumulated (possibly empty). Used
    for best-effort drain, passive SBI autoprint sniffing during
    protocol detection, and stream-stop recovery.
    """
    ...

async def read_exact(self, n: int, *, timeout: float) -> bytes:
    """Read exactly ``n`` bytes.

    The canonical shape for xBPI framing: read one byte to discover
    the length, then read that many bytes. Raises
    :class:`sartoriuslib.errors.SartoriusTimeoutError` if fewer than
    ``n`` bytes arrive before ``timeout``. Partial buffers are retained
    for the next call — implementations must not discard them.
    """
    ...

async def read_until(self, separator: bytes, *, timeout: float) -> bytes:
    """Read bytes up to and including the next occurrence of ``separator``.

    Raises :class:`sartoriuslib.errors.SartoriusTimeoutError` if the
    separator does not arrive before ``timeout``. Bytes received
    after the separator remain buffered for the next call —
    implementations must not discard them.
    """
    ...

async def reopen(
    self,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
) -> None:
    """Close and re-open the port, optionally changing serial framing.

    Used by :meth:`Balance.configure_protocol` for the WZA SBI→xBPI
    flip, which swaps both baud rate and parity, and by operations
    like ``set_baud_rate`` that retune the port after the device has
    already switched mid-sequence. Any argument left as ``None`` keeps
    the existing value.

    Implementations must leave the transport in a consistent state:
    either fully reopened with the new settings, or clearly closed so
    callers can recognise a failure. Silent partial states are the
    worst failure mode for this method.

    Non-serial transports (e.g. a future TCP adapter) may raise
    :class:`NotImplementedError` — baud/parity don't apply there.
    """
    ...

async def write(self, data: bytes, *, timeout: float) -> None:
    """Write every byte of ``data``.

    Raises :class:`sartoriuslib.errors.SartoriusTimeoutError` on
    expiry. A bounded write timeout is mandatory because sends can
    block on RS-485 hardware flow control or a stuck device. Callers
    that block indefinitely hide real hangs.
    """
    ...

async def reopen(
    self,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
) -> None:
    """Close and reopen the port with any subset of overrides.

    Any argument left as ``None`` keeps the existing setting. Used for
    the WZA SBI→xBPI flip (swaps both baud and parity) and for
    ``set_baud_rate`` retuning after the device has already switched.
    The cached :class:`SerialSettings` is updated so the new settings
    survive subsequent close / open round-trips.

    If :meth:`open` fails on the new settings the transport is left
    closed; the caller is responsible for surfacing that as a
    ``BROKEN`` session state with recovery guidance.
    """
    await self.close()
    new_settings = self._settings
    if baudrate is not None:
        new_settings = replace(new_settings, baudrate=baudrate)
    if parity is not None:
        new_settings = replace(new_settings, parity=parity)
    if stopbits is not None:
        new_settings = replace(new_settings, stopbits=stopbits)
    self._settings = new_settings
    await self.open()

def add_script(self, command: bytes, reply: ScriptedReply) -> None:
    """Register or overwrite a scripted reply for ``command``."""
    self._script[bytes(command)] = reply

def feed(self, data: bytes) -> None:
    """Push unsolicited bytes into the read buffer.

    Useful for simulating a device that was left in SBI autoprint
    mode, or garbage on the line that the session must drain on
    recovery.
    """
    self._read_buffer.extend(data)

def force_read_timeout(self, enabled: bool = True) -> None:
    """Force the next read to raise ``SartoriusTimeoutError``."""
    self._force_read_timeout = enabled

def force_reopen_error(self, enabled: bool = True) -> None:
    """Force the next :meth:`reopen` to raise ``SartoriusConnectionError``."""
    self._force_reopen_error = enabled

def force_write_timeout(self, enabled: bool = True) -> None:
    """Force the next :meth:`write` to raise ``SartoriusTimeoutError``."""
    self._force_write_timeout = enabled

async def reopen(
    self,
    *,
    baudrate: int | None = None,
    parity: Parity | None = None,
    stopbits: StopBits | None = None,
) -> None:
    """Simulate a serial-setting change — close, record, reopen.

    If ``force_reopen_error()`` has been called the reopen raises
    :class:`SartoriusConnectionError` after the close, leaving the
    transport closed. That is the "reopen wedged" path used by
    :meth:`Balance.configure_protocol` tests for the BROKEN-state
    transition.
    """
    await self.close()
    self._reopen_count += 1
    if baudrate is not None:
        self._last_reopen_baud = baudrate
    if parity is not None:
        self._last_reopen_parity = parity
    if stopbits is not None:
        self._last_reopen_stopbits = stopbits
    if self._force_reopen_error:
        raise SartoriusConnectionError(
            f"forced reopen error on {self._label}",
            context=ErrorContext(port=self._label),
        )
    await self.open()