Merge branch 'master' of github.com:OpenCyphal/pycyphal into dev

Author: songmeo

.test_deps/README.md

@@ -8,28 +8,6 @@ Please keep this document in sync with the contents of this directory.
 
 ## Nmap project binaries
 
-### Portable Ncat
-
-Ncat is needed for brokering TCP connections that emulate serial port connections.
-This is needed for testing the Cyphal/serial transport without having to access a physical serial port
-(which would be difficult to set up on a CI server).
-
-The binary comes with the following statement by its developers:
-
-> This is a portable (statically compiled) Win32 version of Ncat.
-> You should be able to take the ncat.exe and run it on other systems without having to also copy over
-> a bunch of DLLs, etc.
->
-> More information on Ncat: http://nmap.org/ncat/
->
-> You can get the version number of this ncat by runnign "ncat --version".
-> We don't create a new Ncat portable for each Ncat release,
-> so you will have to compile your own if you want a newer version.
-> Instructions for doing so are available at: https://secwiki.org/w/Nmap/Ncat_Portable
->
-> Ncat is distributed under the same free and open source license as Nmap.
-> See http://nmap.org/book/man-legal.html.
-
 
 ### Npcap installer
 

CONTRIBUTING.rst

@@ -171,7 +171,7 @@ a list of all the paths where the DSDL root namespace directories are to be foun
 Next, open 2 terminal windows.
 In the first, run::
 
-    ncat --broker --listen -p 50905
+    cyphal-serial-broker -p 50905
 
 In the second one::
 

demo/launch.orc.yaml

@@ -10,8 +10,9 @@ PYCYPHAL_PATH: ".pycyphal_generated"  # This one is optional; the default is "~/
 uavcan:
   # Use Cyphal/UDP via localhost:
   udp.iface: 127.0.0.1
-  # If you have Ncat or some other TCP broker, you can use Cyphal/serial tunneled over TCP (in a heterogeneous
-  # redundant configuration with UDP or standalone). Ncat launch example: ncat --broker --listen --source-port 50905
+  # You can use Cyphal/serial tunneled over TCP (in a heterogeneous redundant configuration with
+  # UDP or standalone). pycyphal includes cyphal-serial-broker for this purpose:
+  #   cyphal-serial-broker --port 50905
   serial.iface: "" # socket://127.0.0.1:50905
   # It is recommended to explicitly assign unused transports to ensure that previously stored transport
   # configurations are not accidentally reused:

noxfile.py

@@ -75,8 +75,8 @@ def test(session):
         session.run("sudo", "setcap", "cap_net_raw+eip", str(Path(session.bin, "python").resolve()), external=True)
 
     # Launch the TCP broker for testing the Cyphal/serial transport.
-    os.system("ncat --version")
-    broker_process = subprocess.Popen(["ncat", "--broker", "--listen", "-p", "50905"])
+    broker_path = shutil.which("cyphal-serial-broker", path=os.pathsep.join(session.bin_paths))
+    broker_process = subprocess.Popen([broker_path, "--port", "50905"])
     time.sleep(1.0)  # Ensure that it has started.
     if broker_process.poll() is not None:
         raise RuntimeError("Could not start the TCP broker")

pycyphal/_version.py

@@ -1 +1 @@
-__version__ = "1.23.0"
+__version__ = "1.23.2"

pycyphal/transport/serial/__init__.py

@@ -82,23 +82,18 @@
 instead of a real serial port name when establishing the connection.
 
 The location specified in the URL must point to the TCP server port that will forward the data
-to and from the other end of the link.
-While such a server can be trivially coded manually by the developer,
-it is possible to avoid the effort by relying on the TCP connection brokering mode available in
-Ncat (which is a part of the `Nmap <https://nmap.org>`_ project, thanks Fyodor).
+to and from the other end of the link. For this purpose PyCyphal includes ``cyphal-serial-broker``.
+Alternatively, ncat (which is a part of the `Nmap <https://nmap.org>`_ project, thanks Fyodor)
+has the broker mode.
 
-For example, one could set up the TCP broker as follows
-(add ``-v`` to see what's happening; more info at https://nmap.org/ncat/guide/ncat-broker.html)
-(the port number is chosen at random here)::
+For example, one could use ``cyphal-serial-broker`` as follows (the port number is chosen at random here)::
 
-    ncat --broker --listen -p 50905
+    cyphal-serial-broker -p 50906
 
 And then use a serial transport with ``socket://127.0.0.1:50905``
 (N.B.: using ``localhost`` may significantly increase initialization latency on Windows due to slow DNS lookup).
 All nodes whose transports are configured like that will be able to communicate with each other,
 as if they were connected to the same bus.
-Essentially, this can be seen as a virtualized RS-485 bus,
-where same concerns regarding medium access coordination apply.
 
 The location of the URI doesn't have to be local, of course --
 one can use this approach to link Cyphal nodes via conventional IP networks.

pycyphal/transport/serial/_serial.py

@@ -339,7 +339,7 @@ def _handle_received_out_of_band_data(self, timestamp: Timestamp, data: memoryvi
             printable = printable.decode("utf8")
         except ValueError:
             pass
-        _logger.warning("%s: Out-of-band received at %s: %r", self._serial_port.name, timestamp, printable)
+        _logger.info("%s: Out-of-band received at %s: %r", self._serial_port.name, timestamp, printable)
 
     def _handle_received_item_and_update_stats(
         self, timestamp: Timestamp, item: typing.Union[SerialFrame, memoryview], in_bytes_count: int

pycyphal/util/_broker.py

@@ -0,0 +1,137 @@
+"""
+Cyphal/Serial-over-TCP broker.
+
+Cyphal/Serial uses COBS-encoded frames with a zero byte as frame delimiter. When
+brokering a byte-stream ncat --broker does know about the frame delimiter and
+might interleave frames from different clients.
+This broker is similar in functionality to :code:`ncat --broker`, but reads the
+whole frame before passing it on to other clients, avoiding interleaved frames
+and potential frame/data loss.
+"""
+
+import argparse
+import asyncio
+import logging
+import socket
+import typing as t
+
+
+class Client:
+    """
+    Represents a client connected to the broker, wrapping StreamReader and
+    StreamWriter to conveniently read/write zero-terminated frames.
+    """
+
+    def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        self._buffer = bytearray()
+        self._reader = reader
+        self._writer = writer
+
+    async def __aenter__(self) -> "Client":
+        return self
+
+    async def __aexit__(self, *_: t.Any) -> bool:
+        self._writer.close()
+        await self._writer.wait_closed()
+        return True
+
+    async def read(self) -> t.AsyncGenerator[bytes, None]:
+        """
+        async generator yielding complete frames, including terminating \x00.
+        """
+        buffer = bytearray()
+        while not self._reader.at_eof():
+            buffer += await self._reader.readuntil(separator=b"\x00")
+            # don't pass on a leading zero-byte on its own.
+            if len(buffer) == 1:
+                continue
+            yield buffer
+            buffer = bytearray()
+
+    def write(self, frame: bytes) -> None:
+        """
+        Writes a frame to the stream, unless the stream is closing.
+
+        :param frame: Frame to send to this client.
+        """
+        if self._writer.is_closing():
+            return
+        self._writer.write(frame)
+
+    async def drain(self) -> None:
+        """
+        Flushes the stream.
+        """
+        if self._writer.is_closing():
+            return
+        await self._writer.drain()
+
+
+async def serve_forever(host: str, port: int) -> None:
+    """
+    pybroker core server loop.
+
+    Accept clients on :code:`host`::code:`port` and broadcast any frame
+    received from any client to all other clients.
+
+    :param host: IP, where the broker will be reachable on.
+    :param port: port, on which the broker will listen on.
+    """
+    clients: list[Client] = []
+    list_lock = asyncio.Lock()
+
+    async def _run_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        async with Client(reader, writer) as client:
+            async with list_lock:
+                logging.info("Client connected.")
+                clients.append(client)
+            try:
+                async for frame in client.read():
+                    logging.debug("Received frame %s", frame)
+                    for c in clients:
+                        if c != client:
+                            c.write(frame)
+                    async with list_lock:
+                        # not sure if flushing is required.
+                        for c in clients:
+                            await c.drain()
+
+            finally:
+                async with list_lock:
+                    clients.remove(client)
+                    logging.info("Client disconnected.")
+
+    logging.info("Broker started on %s:%s", host, port)
+    reuse_port = hasattr(socket, "SO_REUSEPORT") and socket.SO_REUSEPORT
+    await asyncio.start_server(
+        _run_client,
+        host,
+        port,
+        family=socket.AF_INET,
+        reuse_address=True,
+        reuse_port=reuse_port,
+    )
+
+
+def main() -> None:
+    """
+    TCP-broker which forwards complete, zero-terminated frames/datagrams among
+    all connected clients.
+    """
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("-i", "--host", default="127.0.0.1", help="Interface to listen on for incoming connections.")
+    parser.add_argument("-p", "--port", default=50905, help="Clients connect to this port.")
+    parser.add_argument("--verbose", default=False, action="store_true", help="Increase logging verbosity.")
+
+    args = parser.parse_args()
+
+    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO)
+
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(serve_forever(args.host, args.port))
+    loop.run_forever()

setup.cfg

@@ -36,6 +36,10 @@ classifiers =
     Operating System :: OS Independent
     Typing :: Typed
 
+[options.entry_points]
+console_scripts =
+    cyphal-serial-broker = pycyphal.util._broker:main
+
 [options.extras_require]
 # Key name format: "transport-<transport-name>-<media-name>"; e.g.: "transport-ieee802154-xbee".
 # If there is no media sub-layer, or the media dependencies are shared, or it is desired to have a common