diff --git a/examples/_file_types.py b/examples/_file_types.py new file mode 100644 index 000000000..4f6e84c14 --- /dev/null +++ b/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] = " 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] = " 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]) diff --git a/examples/file_client.py b/examples/file_client.py new file mode 100755 index 000000000..efde122ba --- /dev/null +++ b/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 > 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() diff --git a/examples/file_server.py b/examples/file_server.py new file mode 100755 index 000000000..ba0465574 --- /dev/null +++ b/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() diff --git a/noxfile.py b/noxfile.py index 3adb23268..18003ec7e 100644 --- a/noxfile.py +++ b/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() diff --git a/src/pycyphal2/__init__.py b/src/pycyphal2/__init__.py index 3db3f3bdd..280b4b0e3 100644 --- a/src/pycyphal2/__init__.py +++ b/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__ = [