mbus-ffi 0.13.0

"""
Type stubs for the ``modbus_rs._modbus_rs`` native extension module.

All classes are generated by the Rust / PyO3 backend.  Async methods return
awaitables that can be ``await``-ed with any asyncio-compatible event loop.
"""

from __future__ import annotations
from typing import Any, Optional, Awaitable

# ---------------------------------------------------------------------------
# Exceptions
# ---------------------------------------------------------------------------

class ModbusError(Exception):
    """Base exception for all modbus-rs errors."""

class ModbusTimeout(ModbusError):
    """Request timed out waiting for a response."""

class ModbusConnectionError(ModbusError):
    """Connection could not be established or was lost."""

class ModbusProtocolError(ModbusError):
    """The remote device returned a malformed or unexpected PDU."""

class ModbusDeviceException(ModbusProtocolError):
    """The remote device returned a Modbus exception response."""

class ModbusConfigError(ModbusError):
    """Invalid configuration (address, baud rate, etc.)."""

class ModbusInvalidArgument(ModbusError):
    """Invalid argument passed to a Modbus call (out-of-range address, count, etc.)."""

# ---------------------------------------------------------------------------
# Sync TCP client
# ---------------------------------------------------------------------------

class TcpClient:
    """Synchronous (blocking) Modbus TCP client.

    Each method releases the GIL while waiting for the network response so
    that other Python threads continue to run.

    Usage::

        with TcpClient("192.168.1.10", unit_id=1) as client:
            regs = client.read_holding_registers(0, 5)
    """

    def __init__(
        self,
        host: str,
        port: int = 502,
        unit_id: int = 1,
        timeout_ms: int = 1000,
    ) -> None: ...

    def connect(self) -> None:
        """Establish the TCP connection."""

    def disconnect(self) -> None:
        """Disconnect from the server."""

    def has_pending_requests(self) -> bool:
        """Return ``True`` if there are in-flight requests."""

    def __enter__(self) -> TcpClient: ...
    def __exit__(self, *args: Any) -> bool: ...

    # Coils
    def read_coils(self, address: int, quantity: int) -> list[bool]: ...
    def write_coil(self, address: int, value: bool) -> tuple[int, bool]: ...
    def write_coils(self, address: int, values: list[bool]) -> None: ...

    # Discrete inputs
    def read_discrete_inputs(self, address: int, quantity: int) -> list[bool]: ...

    # Holding registers
    def read_holding_registers(self, address: int, quantity: int) -> list[int]: ...
    def write_register(self, address: int, value: int) -> tuple[int, int]: ...
    def write_registers(self, address: int, values: list[int]) -> None: ...
    def mask_write_register(
        self, address: int, and_mask: int, or_mask: int
    ) -> tuple[int, int, int]: ...
    def read_write_registers(
        self,
        read_address: int,
        read_quantity: int,
        write_address: int,
        write_values: list[int],
    ) -> list[int]: ...

    # Input registers
    def read_input_registers(self, address: int, quantity: int) -> list[int]: ...

    # FIFO queue
    def read_fifo_queue(self, address: int) -> list[int]: ...

    # File records
    def read_file_record(self, file_number: int, record_number: int, record_length: int) -> list[int]: ...
    def write_file_record(self, file_number: int, record_number: int, data: list[int]) -> None: ...

    # Diagnostics
    def get_device_identification(self, object_id: int) -> dict[int, bytes]: ...

# ---------------------------------------------------------------------------
# Async TCP client
# ---------------------------------------------------------------------------

class AsyncTcpClient:
    """Asyncio Modbus TCP client.

    Usage::

        async with AsyncTcpClient("192.168.1.10", unit_id=1) as client:
            regs = await client.read_holding_registers(0, 5)
    """

    def __init__(
        self,
        host: str,
        port: int = 502,
        unit_id: int = 1,
        timeout_ms: int = 1000,
    ) -> None: ...

    def connect(self) -> Any:
        """Awaitable — establish the TCP connection."""

    def has_pending_requests(self) -> bool:
        """Return ``True`` if there are in-flight requests."""

    async def __aenter__(self) -> AsyncTcpClient: ...
    async def __aexit__(self, *args: Any) -> bool: ...

    # Coils
    def read_coils(self, address: int, quantity: int) -> Any: ...
    def write_coil(self, address: int, value: bool) -> Any: ...
    def write_coils(self, address: int, values: list[bool]) -> Any: ...

    # Discrete inputs
    def read_discrete_inputs(self, address: int, quantity: int) -> Any: ...

    # Holding registers
    def read_holding_registers(self, address: int, quantity: int) -> Any: ...
    def write_register(self, address: int, value: int) -> Any: ...
    def write_registers(self, address: int, values: list[int]) -> Any: ...
    def mask_write_register(self, address: int, and_mask: int, or_mask: int) -> Any: ...
    def read_write_registers(
        self,
        read_address: int,
        read_quantity: int,
        write_address: int,
        write_values: list[int],
    ) -> Any: ...

    # Input registers
    def read_input_registers(self, address: int, quantity: int) -> Any: ...

    # FIFO queue
    def read_fifo_queue(self, address: int) -> Any: ...

    # File records
    def read_file_record(self, file_number: int, record_number: int, record_length: int) -> Any: ...
    def write_file_record(self, file_number: int, record_number: int, data: list[int]) -> Any: ...

    # Diagnostics
    def get_device_identification(self, object_id: int) -> Any: ...

# ---------------------------------------------------------------------------
# Sync serial client
# ---------------------------------------------------------------------------

class SerialClient:
    """Synchronous (blocking) Modbus serial client (RTU or ASCII).

    Usage::

        with SerialClient("/dev/ttyUSB0", baud_rate=9600, unit_id=1) as client:
            regs = client.read_holding_registers(0, 5)
    """

    def __init__(
        self,
        port: str,
        baud_rate: int = 9600,
        unit_id: int = 1,
        mode: str = "rtu",
        timeout_ms: int = 1000,
    ) -> None: ...

    def connect(self) -> None: ...
    def disconnect(self) -> None: ...
    def has_pending_requests(self) -> bool: ...

    def __enter__(self) -> SerialClient: ...
    def __exit__(self, *args: Any) -> bool: ...

    # Coils
    def read_coils(self, address: int, quantity: int) -> list[bool]: ...
    def write_coil(self, address: int, value: bool) -> tuple[int, bool]: ...
    def write_coils(self, address: int, values: list[bool]) -> None: ...

    # Discrete inputs
    def read_discrete_inputs(self, address: int, quantity: int) -> list[bool]: ...

    # Holding registers
    def read_holding_registers(self, address: int, quantity: int) -> list[int]: ...
    def write_register(self, address: int, value: int) -> tuple[int, int]: ...
    def write_registers(self, address: int, values: list[int]) -> None: ...
    def mask_write_register(self, address: int, and_mask: int, or_mask: int) -> tuple[int, int, int]: ...
    def read_write_registers(
        self,
        read_address: int,
        read_quantity: int,
        write_address: int,
        write_values: list[int],
    ) -> list[int]: ...

    # Input registers
    def read_input_registers(self, address: int, quantity: int) -> list[int]: ...

    # FIFO queue
    def read_fifo_queue(self, address: int) -> list[int]: ...

    # File records
    def read_file_record(self, file_number: int, record_number: int, record_length: int) -> list[int]: ...
    def write_file_record(self, file_number: int, record_number: int, data: list[int]) -> None: ...

    # Diagnostics
    def get_device_identification(self, object_id: int) -> dict[int, bytes]: ...

# ---------------------------------------------------------------------------
# Async serial client
# ---------------------------------------------------------------------------

class AsyncSerialClient:
    """Asyncio Modbus serial client (RTU or ASCII).

    Usage::

        async with AsyncSerialClient("/dev/ttyUSB0", unit_id=1) as client:
            regs = await client.read_holding_registers(0, 5)
    """

    def __init__(
        self,
        port: str,
        baud_rate: int = 9600,
        unit_id: int = 1,
        mode: str = "rtu",
        timeout_ms: int = 1000,
    ) -> None: ...

    def connect(self) -> Any: ...
    def has_pending_requests(self) -> bool: ...

    async def __aenter__(self) -> AsyncSerialClient: ...
    async def __aexit__(self, *args: Any) -> bool: ...

    # Coils
    def read_coils(self, address: int, quantity: int) -> Any: ...
    def write_coil(self, address: int, value: bool) -> Any: ...
    def write_coils(self, address: int, values: list[bool]) -> Any: ...

    # Discrete inputs
    def read_discrete_inputs(self, address: int, quantity: int) -> Any: ...

    # Holding registers
    def read_holding_registers(self, address: int, quantity: int) -> Any: ...
    def write_register(self, address: int, value: int) -> Any: ...
    def write_registers(self, address: int, values: list[int]) -> Any: ...
    def mask_write_register(self, address: int, and_mask: int, or_mask: int) -> Any: ...
    def read_write_registers(
        self,
        read_address: int,
        read_quantity: int,
        write_address: int,
        write_values: list[int],
    ) -> Any: ...

    # Input registers
    def read_input_registers(self, address: int, quantity: int) -> Any: ...

    # FIFO queue
    def read_fifo_queue(self, address: int) -> Any: ...

    # File records
    def read_file_record(self, file_number: int, record_number: int, record_length: int) -> Any: ...
    def write_file_record(self, file_number: int, record_number: int, data: list[int]) -> Any: ...

    # Diagnostics
    def get_device_identification(self, object_id: int) -> Any: ...

# ---------------------------------------------------------------------------
# Server application base class
# ---------------------------------------------------------------------------

class ModbusApp:
    """Base class for Modbus server application handlers.

    Subclass this and override the ``handle_*`` methods to implement your
    server logic.  Raise :class:`ModbusError` (or a subclass) to return an
    exception response to the client.

    Usage::

        class MyApp(modbus_rs.ModbusApp):
            def handle_read_holding_registers(self, address, count):
                return [0] * count
    """

    def handle_read_coils(self, address: int, count: int) -> list[bool] | Awaitable[list[bool]]:
        """Return a list of ``count`` coil values starting at ``address``."""
        ...

    def handle_write_coil(self, address: int, value: bool) -> None | Awaitable[None]:
        """Set the coil at ``address`` to ``value``."""
        ...

    def handle_write_coils(self, address: int, values: list[bool]) -> None | Awaitable[None]:
        """Write multiple coils (FC15)."""
        ...

    def handle_read_discrete_inputs(self, address: int, count: int) -> list[bool] | Awaitable[list[bool]]:
        """Return a list of ``count`` discrete input values starting at ``address``."""
        ...

    def handle_read_holding_registers(self, address: int, count: int) -> list[int] | Awaitable[list[int]]:
        """Return a list of ``count`` holding register values."""
        ...

    def handle_read_input_registers(self, address: int, count: int) -> list[int] | Awaitable[list[int]]:
        """Return a list of ``count`` input register values."""
        ...

    def handle_write_register(self, address: int, value: int) -> None | Awaitable[None]:
        """Write a single holding register."""
        ...

    def handle_write_registers(self, address: int, values: list[int]) -> None | Awaitable[None]:
        """Write multiple holding registers (FC16)."""
        ...

    def handle_mask_write_register(
        self, address: int, and_mask: int, or_mask: int
    ) -> None | Awaitable[None]:
        """Apply bit masks to a holding register (FC22)."""
        ...

    def handle_read_write_registers(
        self,
        read_address: int,
        read_count: int,
        write_address: int,
        write_values: list[int],
    ) -> list[int] | Awaitable[list[int]]:
        """Read/write holding registers in one transaction (FC23)."""
        ...

    def handle_read_fifo_queue(self, pointer_address: int) -> list[int] | Awaitable[list[int]]:
        """Return FIFO queue contents starting at ``pointer_address`` (FC24)."""
        ...

    def handle_read_exception_status(self) -> int | Awaitable[int]:
        """Return the 8-bit exception status byte (FC07)."""
        ...

    def handle_get_comm_event_counter(self) -> tuple[int, int] | Awaitable[tuple[int, int]]:
        """Return ``(status, event_count)`` tuple (FC11)."""
        ...

# ---------------------------------------------------------------------------
# Async TCP server
# ---------------------------------------------------------------------------

class AsyncTcpServer:
    """Asyncio Modbus TCP server.

    Pass a :class:`ModbusApp` subclass instance; the server dispatches each
    incoming request to the corresponding ``handle_*`` method.

    Usage::

        class MyApp(modbus_rs.ModbusApp):
            def handle_read_holding_registers(self, address, count):
                return [0] * count

        async def main():
            async with AsyncTcpServer("0.0.0.0", MyApp(), port=502, unit_id=1) as server:
                await server.serve_forever()
    """

    def __init__(
        self,
        host: str,
        app: ModbusApp,
        port: int = 502,
        unit_id: int = 1,
    ) -> None: ...

    def serve_forever(self) -> Any:
        """Awaitable — bind and serve until an error occurs."""

    async def __aenter__(self) -> AsyncTcpServer: ...
    async def __aexit__(self, *args: Any) -> bool: ...

# ---------------------------------------------------------------------------
# Sync TCP server
# ---------------------------------------------------------------------------

class TcpServer:
    """Synchronous (blocking) Modbus TCP server.

    Blocks the calling thread until an error or shutdown.

    Usage::

        server = TcpServer("0.0.0.0", MyApp(), port=502, unit_id=1)
        server.serve_forever()  # blocks
    """

    def __init__(
        self,
        host: str,
        app: ModbusApp,
        port: int = 502,
        unit_id: int = 1,
    ) -> None: ...

    def serve_forever(self) -> None:
        """Block the calling thread, serving requests until an error."""

    def __enter__(self) -> TcpServer: ...
    def __exit__(self, *args: Any) -> bool: ...

# ---------------------------------------------------------------------------
# Async serial server
# ---------------------------------------------------------------------------

class AsyncSerialServer:
    """Asyncio Modbus serial server (RTU or ASCII).

    Usage::

        async with AsyncSerialServer("/dev/ttyUSB0", MyApp(), unit_id=1) as srv:
            await srv.serve_forever()
    """

    def __init__(
        self,
        port: str,
        app: ModbusApp,
        baud_rate: int = 9600,
        unit_id: int = 1,
        mode: str = "rtu",
        timeout_ms: int = 1000,
    ) -> None: ...

    def serve_forever(self) -> Any:
        """Awaitable — run the server until the port closes."""

    async def __aenter__(self) -> AsyncSerialServer: ...
    async def __aexit__(self, *args: Any) -> bool: ...

# ---------------------------------------------------------------------------
# Sync serial server
# ---------------------------------------------------------------------------

class SerialServer:
    """Synchronous (blocking) Modbus serial server (RTU or ASCII).

    Usage::

        server = SerialServer("/dev/ttyUSB0", MyApp(), baud_rate=9600, unit_id=1)
        server.serve_forever()  # blocks
    """

    def __init__(
        self,
        port: str,
        app: ModbusApp,
        baud_rate: int = 9600,
        unit_id: int = 1,
        mode: str = "rtu",
        timeout_ms: int = 1000,
    ) -> None: ...

    def serve_forever(self) -> None:
        """Block the calling thread, serving requests until the port closes."""

    def __enter__(self) -> SerialServer: ...
    def __exit__(self, *args: Any) -> bool: ...


# ─── Gateway (feature: python-gateway) ─────────────────────────────────────

class GatewayEventHandler:
    """
    Forward-compatible base class for gateway lifecycle events.

    All methods are no-ops by default. Override only what you need.
    """

    def __init__(self) -> None: ...
    def on_forward(self, session_id: int, unit_id: int, channel_idx: int) -> None: ...
    def on_response_returned(self, session_id: int, upstream_txn: int) -> None: ...
    def on_routing_miss(self, session_id: int, unit_id: int) -> None: ...
    def on_downstream_timeout(self, session_id: int, internal_txn: int) -> None: ...
    def on_upstream_disconnect(self, session_id: int) -> None: ...


class AsyncTcpGateway:
    """Asyncio Modbus TCP→TCP gateway."""

    def __init__(
        self,
        bind_addr: str,
        event_handler: Optional[GatewayEventHandler] = None,
    ) -> None: ...

    def bind_address(self) -> str: ...
    def add_tcp_downstream(self, host: str, port: int = 502) -> int:
        """Register a downstream TCP endpoint; returns its channel index."""
    def add_unit_route(self, unit: int, channel: int) -> None: ...
    def add_range_route(self, min: int, max: int, channel: int) -> None: ...

    async def serve_forever(self) -> None:
        """Bind the listener and serve until :meth:`stop` is called."""

    def stop(self) -> None: ...

    async def __aenter__(self) -> "AsyncTcpGateway": ...
    async def __aexit__(self, *args: Any) -> bool: ...


class TcpGateway:
    """Blocking Modbus TCP→TCP gateway."""

    def __init__(
        self,
        bind_addr: str,
        event_handler: Optional[GatewayEventHandler] = None,
    ) -> None: ...

    def bind_address(self) -> str: ...
    def add_tcp_downstream(self, host: str, port: int = 502) -> int: ...
    def add_unit_route(self, unit: int, channel: int) -> None: ...
    def add_range_route(self, min: int, max: int, channel: int) -> None: ...

    def serve_forever(self) -> None:
        """Block the calling thread, serving until :meth:`stop` is called."""

    def stop(self) -> None: ...

    def __enter__(self) -> "TcpGateway": ...
    def __exit__(self, *args: Any) -> bool: ...