Add file_client and file_server examples

examples/_file_types.py

@@ -0,0 +1,67 @@
+"""Wire types shared by the file RPC examples."""
+
+from __future__ import annotations
+
+import struct
+from dataclasses import dataclass
+from typing import ClassVar
+
+
+@dataclass(frozen=True)
+class FileReadRequest:
+    read_offset: int
+    file_path: str
+
+    _HEADER_FORMAT: ClassVar[str] = "<QH"
+    _HEADER_SIZE: ClassVar[int] = struct.calcsize(_HEADER_FORMAT)
+    _PATH_MAX_LEN: ClassVar[int] = 2048
+
+    def serialize(self) -> bytes:
+        encoded_path = self.file_path.encode("utf8")
+        if len(encoded_path) > self._PATH_MAX_LEN:
+            raise ValueError(f"File path length {len(encoded_path)} is too long")
+        return struct.pack(self._HEADER_FORMAT, self.read_offset, len(encoded_path)) + encoded_path
+
+    @staticmethod
+    def deserialize(payload: bytes) -> FileReadRequest | None:
+        if len(payload) < FileReadRequest._HEADER_SIZE:
+            return None
+        read_offset, path_len = struct.unpack_from(FileReadRequest._HEADER_FORMAT, payload)
+        if path_len == 0 or path_len > FileReadRequest._PATH_MAX_LEN:
+            return None
+        path_end = FileReadRequest._HEADER_SIZE + path_len
+        if len(payload) != path_end:
+            return None
+        try:
+            file_path = payload[FileReadRequest._HEADER_SIZE : path_end].decode("utf8")
+        except UnicodeDecodeError:
+            return None
+        return FileReadRequest(read_offset, file_path)
+
+
+@dataclass(frozen=True)
+class FileReadResponse:
+    error: int
+    data: bytes
+
+    _HEADER_FORMAT: ClassVar[str] = "<IH"
+    _HEADER_SIZE: ClassVar[int] = struct.calcsize(_HEADER_FORMAT)
+    DATA_CAPACITY: ClassVar[int] = 4096
+
+    def serialize(self) -> bytes:
+        if len(self.data) > self.DATA_CAPACITY:
+            raise ValueError(f"Response data is too large: {len(self.data)}")
+        return struct.pack(self._HEADER_FORMAT, self.error, len(self.data)) + self.data
+
+    @staticmethod
+    def deserialize(payload: bytes) -> FileReadResponse | None:
+        if len(payload) < FileReadResponse._HEADER_SIZE:
+            return None
+        error, data_len = struct.unpack_from(FileReadResponse._HEADER_FORMAT, payload)
+        if data_len > FileReadResponse.DATA_CAPACITY:
+            return None
+        data_start = FileReadResponse._HEADER_SIZE
+        data_end = data_start + data_len
+        if len(payload) != data_end:
+            return None
+        return FileReadResponse(error, payload[data_start:data_end])

examples/file_client.py

@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+"""
+Read a file from a Cyphal/UDP file server and write it to stdout.
+Usage:
+    python examples/file_client.py <file> > copy.bin
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import logging
+import os
+import sys
+from pathlib import Path
+
+from _file_types import FileReadRequest, FileReadResponse
+
+from pycyphal2 import DeliveryError, Instant, LivenessError, Node, ResponseStream, SendError
+from pycyphal2.udp import UDPTransport
+
+NAME = f"{Path(__file__).stem}/"  # The trailing separator ensures that a random ID will be added.
+TOPIC = "file/read"
+RESPONSE_TIMEOUT = 30.0
+REQUEST_DELIVERY_TIMEOUT = RESPONSE_TIMEOUT / 2.0
+_logger = logging.getLogger(__name__)
+
+
+def _format_remote_error(error: int) -> str:
+    try:
+        message = os.strerror(error)
+    except (OverflowError, ValueError):
+        message = "unknown error"
+    return f"Remote error {error}: {message}"
+
+
+async def _receive_response(stream: ResponseStream, expected_server_id: int | None) -> tuple[int, FileReadResponse]:
+    async for response in stream:
+        if expected_server_id is not None and response.remote_id != expected_server_id:
+            _logger.info(
+                "ignoring response from redundant server %016x, expected %016x",
+                response.remote_id,
+                expected_server_id,
+            )
+            continue
+        decoded = FileReadResponse.deserialize(response.message)
+        if decoded is None:
+            _logger.debug("dropping malformed response from %016x seq=%d", response.remote_id, response.seqno)
+            continue
+        return response.remote_id, decoded
+    raise LivenessError("Response stream closed")
+
+
+async def _receive_valid_response(
+    stream: ResponseStream, expected_server_id: int | None, response_timeout: float
+) -> tuple[int, FileReadResponse]:
+    try:
+        return await asyncio.wait_for(_receive_response(stream, expected_server_id), timeout=response_timeout)
+    except asyncio.TimeoutError as ex:
+        raise LivenessError("Response timeout") from ex
+
+
+async def run(file_path: str) -> int:
+    transport = UDPTransport.new()
+    node = Node.new(transport, NAME)
+    pub = node.advertise(TOPIC)
+    discovered_server_id: int | None = None
+    read_offset = 0
+    _logger.info("file client ready on %r via %s", TOPIC, transport)
+    try:
+        while True:
+            _logger.info("requesting offset %d", read_offset)
+            stream: ResponseStream | None = None
+            try:
+                request = FileReadRequest(read_offset, file_path).serialize()
+                stream = await pub.request(Instant.now() + REQUEST_DELIVERY_TIMEOUT, RESPONSE_TIMEOUT, request)
+                server_id, response = await _receive_valid_response(stream, discovered_server_id, RESPONSE_TIMEOUT)
+            except ValueError as ex:
+                sys.stderr.write(f"{ex}\n")
+                return 1
+            except DeliveryError:
+                sys.stderr.write("Request delivery failed\n")
+                return 1
+            except LivenessError:
+                sys.stderr.write("Response timeout\n")
+                return 1
+            except SendError as ex:
+                sys.stderr.write(f"Request send failed: {ex}\n")
+                return 1
+            finally:
+                if stream is not None:
+                    stream.close()
+
+            if discovered_server_id is None:
+                discovered_server_id = server_id
+                _logger.info("discovered server UID: %016x", discovered_server_id)
+            _logger.info("received response: offset %d", read_offset)
+
+            if response.error != 0:
+                sys.stderr.write(_format_remote_error(response.error) + "\n")
+                return 1
+            if len(response.data) > 0:
+                sys.stdout.buffer.write(response.data)
+                sys.stdout.buffer.flush()
+                read_offset += len(response.data)
+                continue
+
+            _logger.info("finished transferring %d bytes", read_offset)
+            return 0
+    finally:
+        pub.close()
+        node.close()
+        transport.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Read a file from a Cyphal/UDP file server.")
+    parser.add_argument("file", help="File path to read on the server")
+    parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging")
+    args = parser.parse_args()
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(levelname)s: %(message)s")
+    try:
+        raise SystemExit(asyncio.run(run(args.file)))
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()

examples/file_server.py

@@ -0,0 +1,94 @@
+#!/usr/bin/env python3
+"""
+Serve file chunks over a tiny Cyphal/UDP RPC.
+Usage:
+    python examples/file_server.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import errno
+import logging
+from pathlib import Path
+
+from _file_types import FileReadRequest, FileReadResponse
+
+from pycyphal2 import Arrival, DeliveryError, NackError, Node, SendError
+from pycyphal2.udp import UDPTransport
+
+NAME = f"{Path(__file__).stem}/"  # The trailing separator ensures that a random ID will be added.
+TOPIC = "file/read"
+RESPONSE_DEADLINE = 10.0
+
+_logger = logging.getLogger(__name__)
+
+
+def _errno_from_exception(ex: BaseException) -> int:
+    if isinstance(ex, OSError) and ex.errno is not None:
+        return ex.errno
+    if isinstance(ex, OverflowError):
+        return getattr(errno, "EOVERFLOW", errno.EINVAL)
+    return errno.EINVAL
+
+
+def _read_chunk(file_path: str, offset: int) -> FileReadResponse:
+    try:
+        with open(file_path, "rb") as file:
+            file.seek(offset)
+            data = file.read(FileReadResponse.DATA_CAPACITY)
+    except (OSError, ValueError, OverflowError) as ex:
+        return FileReadResponse(_errno_from_exception(ex), b"")
+    return FileReadResponse(0, data)
+
+
+async def _serve_request(arrival: Arrival, request: FileReadRequest) -> None:
+    response = _read_chunk(request.file_path, request.read_offset)
+    payload = response.serialize()
+    _logger.info(
+        "responding: file=%r offset=%d size=%d error=%d",
+        request.file_path,
+        request.read_offset,
+        len(response.data),
+        response.error,
+    )
+    try:
+        await arrival.breadcrumb(arrival.timestamp + RESPONSE_DEADLINE, payload, reliable=True)
+    except NackError:
+        _logger.info("client rejected response: remote=%016x file=%r", arrival.breadcrumb.remote_id, request.file_path)
+    except DeliveryError:
+        _logger.info(
+            "client did not acknowledge: remote=%016x file=%r", arrival.breadcrumb.remote_id, request.file_path
+        )
+    except SendError as ex:
+        _logger.warning("response send failed: remote=%016x error=%s", arrival.breadcrumb.remote_id, ex)
+
+
+async def run() -> None:
+    transport = UDPTransport.new()
+    node = Node.new(transport, NAME)
+    sub = node.subscribe(TOPIC)
+    _logger.info("file server ready on %r via %s", TOPIC, transport)
+    try:
+        async for arrival in sub:
+            request = FileReadRequest.deserialize(arrival.message)
+            if request is None:
+                _logger.debug("dropping malformed request of size %d", len(arrival.message))
+                continue
+            await _serve_request(arrival, request)
+    finally:
+        sub.close()
+        node.close()
+        transport.close()
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
+    try:
+        asyncio.run(run())
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == "__main__":
+    main()

noxfile.py

@@ -63,6 +63,7 @@ def examples(session: nox.Session) -> None:
     import json as _json
     import subprocess
     import sys
+    import tempfile
     import time
 
     if sys.platform == "darwin":
@@ -145,9 +146,54 @@ def run_streaming_case() -> None:
             assert "remaining" in obj
             assert "sent_at" in obj
 
+    def run_file_case() -> None:
+        session.log("--- examples smoke: file RPC ---")
+        server_proc = None
+        client_proc = None
+        missing_proc = None
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            file_path = Path(tmp_dir) / "source.bin"
+            payload = bytes(range(256)) * 20
+            file_path.write_bytes(payload)
+            try:
+                server_proc = subprocess.Popen(
+                    [python, "examples/file_server.py"],
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                time.sleep(1)
+                client_proc = subprocess.Popen(
+                    [python, "examples/file_client.py", str(file_path)],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                )
+                client_stdout, client_stderr = client_proc.communicate(timeout=30)
+                assert client_proc.returncode == 0, f"File client failed: {client_stderr.decode()}"
+                assert client_stdout == payload
+
+                missing_proc = subprocess.Popen(
+                    [python, "examples/file_client.py", str(file_path.with_name("missing.bin"))],
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                missing_stdout, missing_stderr = missing_proc.communicate(timeout=30)
+                assert missing_proc.returncode == 1
+                assert missing_stdout == ""
+                assert "Remote error" in missing_stderr
+
+                time.sleep(1)
+                assert server_proc.poll() is None, "File server exited unexpectedly"
+            finally:
+                terminate_process(client_proc)
+                terminate_process(missing_proc)
+                terminate_process(server_proc)
+
     run_case("udp", [])
     if sys.platform == "linux" and Path("/sys/class/net/vcan0").exists():
         run_case("socketcan:vcan0", ["--transport", "socketcan:vcan0"])
     else:
         session.log("Skipping socketcan:vcan0 case (vcan0 not available)")
     run_streaming_case()
+    run_file_case()

src/pycyphal2/__init__.py

@@ -155,7 +155,7 @@ async def main():
 from ._transport import Transport as Transport
 from ._transport import TransportArrival as TransportArrival
 
-__version__ = "2.0.0.dev6"
+__version__ = "2.0.0.dev7"
 
 # pdoc needs __all__ to display re-exported members.
 __all__ = [