Skip to content

fujilib.transport

A transport is a thin lifecycle object that exposes the byte stream the Modbus bus binds to: the real anyserial.SerialPort, so anymodbus keeps its drain-after-send and input-reset behaviour (design §4.1).

fujilib.transport.base

The transport contract and serial framing settings (design §2.1, §4.1).

The ZP series' serial settings are fixed and cannot be changed: 38400 bps, 8 data bits, no parity, 1 stop bit, no flow control. They are still carried as a value so a session can report what it actually used and a second open with incompatible settings can be refused.

A :class:Transport is a thin lifecycle object that exposes its byte stream instead of being one. anymodbus keeps its drain-after-send and input-reset behaviour only when the stream it is given is literally an anyserial.SerialPort, so the Modbus port binds its bus to :attr:Transport.stream directly (design §4.1).

SerialSettings dataclass 

SerialSettings(
    port,
    baudrate=FUJI_BAUDRATE,
    bytesize=ByteSize.EIGHT,
    parity=Parity.NONE,
    stopbits=StopBits.ONE,
    rtscts=False,
    xonxoff=False,
    exclusive=True,
)

Frozen serial framing descriptor. The defaults are the analyzer's fixed 38400 8-N-1.

Transport

Bases: Protocol

An open connection to one serial line.

Implementations: :class:~fujilib.transport.serial.SerialTransport over a real port (or one end of a test pair), and :class:~fujilib.transport.fake.FakeTransport for byte-exact fixture replay.

is_open property 

is_open

Whether the transport is open.

label property 

label

The canonical port name, used in error context and as the ownership key.

settings property 

settings

The serial settings in use.

stream property 

stream

The byte stream the Modbus bus binds to; a real SerialPort where there is one.

aclose async 

aclose()

Close the transport. Idempotent.

Source code in src/fujilib/transport/base.py 
74
75
76
async def aclose(self) -> None:
    """Close the transport. Idempotent."""
    ...

fujilib.transport.serial

The serial transport: an anyserial.SerialPort and the settings it was opened with.

:attr:SerialTransport.stream is the real SerialPort, never a wrapper, so anymodbus drains each request before listening and clears stale input before each request (design §4.1). The same class wraps one end of anyserial.testing.serial_port_pair(), whose ends are real SerialPort objects too, so tests exercise the hardware code path.

SerialTransport 

SerialTransport(port, settings)

An open serial port. Satisfies :class:~fujilib.transport.base.Transport.

Wrap an already open port; :meth:open is the usual constructor.

Source code in src/fujilib/transport/serial.py 
52
53
54
55
def __init__(self, port: SerialPort, settings: SerialSettings) -> None:
    """Wrap an already open ``port``; :meth:`open` is the usual constructor."""
    self._port = port
    self._settings = settings

is_open property 

is_open

Whether the port is open.

label property 

label

The canonical port name.

settings property 

settings

The settings the port was opened with.

stream property 

stream

The real SerialPort.

aclose async 

aclose()

Close the port. Idempotent, and completes even when the caller is cancelled.

Source code in src/fujilib/transport/serial.py 
106
107
108
109
110
111
async def aclose(self) -> None:
    """Close the port. Idempotent, and completes even when the caller is cancelled."""
    if not self._port.is_open:
        return
    with anyio.CancelScope(shield=True):
        await self._port.aclose()

open async classmethod 

open(settings)

Open the port named by settings.port, under its canonical name.

The canonical name is anyserial's, so every spelling of one port agrees: COM8, com8 and \\.\COM8 on Windows, a symlink and its target elsewhere (design §4.1).

Raises:

Type Description
FujiValidationError

settings.port is empty; nothing was opened.
FujiConfigurationError

anyserial rejects the settings.
FujiConnectionError

the port does not exist, is busy or cannot be opened.
Source code in src/fujilib/transport/serial.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
@classmethod
async def open(cls, settings: SerialSettings) -> Self:
    r"""Open the port named by ``settings.port``, under its canonical name.

    The canonical name is ``anyserial``'s, so every spelling of one port
    agrees: ``COM8``, ``com8`` and ``\\.\COM8`` on Windows, a symlink and
    its target elsewhere (design §4.1).

    Raises:
        FujiValidationError: ``settings.port`` is empty; nothing was opened.
        FujiConfigurationError: ``anyserial`` rejects the settings.
        FujiConnectionError: the port does not exist, is busy or cannot be opened.
    """
    stripped = settings.port.strip()
    if not stripped:
        msg = "the serial port name is empty"
        raise FujiValidationError(msg, context=ErrorContext(port=settings.port))
    name = canonical_port_name(stripped)
    context = ErrorContext(port=name, command_name="open")
    try:
        port = await open_serial_port(name, serial_config(settings))
    except ValueError as exc:  # anyserial's ConfigurationError is also a ValueError
        msg = f"cannot open {name} with these settings: {exc}"
        raise FujiConfigurationError(msg, context=context) from exc
    except OSError as exc:  # anyserial's SerialError is an OSError
        msg = f"cannot open {name}: {exc}"
        raise FujiConnectionError(msg, context=context) from exc
    return cls(port, replace(settings, port=name))

serial_config 

serial_config(settings)

The anyserial configuration for settings.

Source code in src/fujilib/transport/serial.py 
35
36
37
38
39
40
41
42
43
44
def serial_config(settings: SerialSettings) -> SerialConfig:
    """The ``anyserial`` configuration for ``settings``."""
    return SerialConfig(
        baudrate=settings.baudrate,
        byte_size=settings.bytesize,
        parity=settings.parity,
        stop_bits=settings.stopbits,
        flow_control=FlowControl(xon_xoff=settings.xonxoff, rts_cts=settings.rtscts),
        exclusive=settings.exclusive,
    )

fujilib.transport.fake

A scripted transport for byte-exact fixture replay (design §4.1, §10).

:class:FakeTransport is its own byte stream. Each request the client sends is looked up in a script of exact request bytes; a match queues the scripted reply, anything else is recorded and gets no reply. anymodbus sends each request in a single send call, so the lookup sees whole frames.

It is used only to prove that the client puts the manual's exact bytes on the wire. It is not a SerialPort, so anymodbus skips drain and input reset with it; those paths are covered by the simulated analyzer on a real port pair (:mod:fujilib.testing).

FakeTransport 

FakeTransport(script=None, *, label='fake://fixture')

Bases: ByteStream

Replays scripted replies. Satisfies :class:~fujilib.transport.base.Transport.

Create a transport that answers each request in script with its reply.

Parameters:

Name Type Description Default
script Mapping[bytes, bytes] | None

Exact request bytes mapped to the reply bytes to send back.

 None
label str

The name reported as the port.

 'fake://fixture'
Source code in src/fujilib/transport/fake.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    script: Mapping[bytes, bytes] | None = None,
    *,
    label: str = "fake://fixture",
) -> None:
    """Create a transport that answers each request in ``script`` with its reply.

    Args:
        script: Exact request bytes mapped to the reply bytes to send back.
        label: The name reported as the port.
    """
    self._script: Mapping[bytes, bytes] = MappingProxyType(dict(script or {}))
    self._settings = SerialSettings(port=label)
    self._inbound = bytearray()
    self._waiter: anyio.Event | None = None
    self._closed = False
    self.writes: list[bytes] = []
    """Every request sent, in order."""
    self.unmatched: list[bytes] = []
    """Requests the script had no reply for."""

is_open property 

is_open

Whether :meth:aclose has not been called.

label property 

label

The name reported as the port.

settings property 

settings

The default ZP settings, under :attr:label.

stream property 

stream

The transport itself.

unmatched instance-attribute 

unmatched = []

Requests the script had no reply for.

writes instance-attribute 

writes = []

Every request sent, in order.

aclose async 

aclose()

Close the transport and wake any pending receive. Idempotent.

Source code in src/fujilib/transport/fake.py 
108
109
110
111
async def aclose(self) -> None:
    """Close the transport and wake any pending receive. Idempotent."""
    self._closed = True
    self._wake()

receive async 

receive(max_bytes=65536)

Return up to max_bytes queued reply bytes, waiting until there are some.

Source code in src/fujilib/transport/fake.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
async def receive(self, max_bytes: int = 65536) -> bytes:
    """Return up to ``max_bytes`` queued reply bytes, waiting until there are some."""
    while not self._inbound:
        if self._closed:
            raise anyio.ClosedResourceError
        self._waiter = anyio.Event()
        await self._waiter.wait()
    chunk = bytes(self._inbound[:max_bytes])
    del self._inbound[:max_bytes]
    return chunk

send async 

send(item)

Record a request and queue its scripted reply, if it has one.

Source code in src/fujilib/transport/fake.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
async def send(self, item: bytes) -> None:
    """Record a request and queue its scripted reply, if it has one."""
    if self._closed:
        raise anyio.ClosedResourceError
    await anyio.lowlevel.checkpoint()
    request = bytes(item)
    self.writes.append(request)
    reply = self._script.get(request)
    if reply is None:
        self.unmatched.append(request)
        return
    self._inbound.extend(reply)
    self._wake()

send_eof async 

send_eof()

Nothing to do: the script decides what comes back.

Source code in src/fujilib/transport/fake.py 
105
106
async def send_eof(self) -> None:
    """Nothing to do: the script decides what comes back."""