refactor

Author: pavel-kirienko

noxfile.py

@@ -1,31 +1,47 @@
 from __future__ import annotations
 
+import shutil
+from pathlib import Path
 import nox
 
 nox.options.sessions = ["test", "mypy", "lint", "format"]
 
+PYTHONS = ["3.11", "3.12", "3.13"]
 
-@nox.session(python=["3.11", "3.12", "3.13"])
+
+@nox.session(python=False, default=False)
+def clean(session):
+    pats = ["dist", "build", "html*", ".coverage*", ".*cache", "src/*.egg-info", "*.log", "*.tmp", ".nox"]
+    for w in pats:
+        for f in Path.cwd().glob(w):
+            session.log(f"Removing: {f}")
+            if f.is_dir():
+                shutil.rmtree(f, ignore_errors=True)
+            else:
+                f.unlink(missing_ok=True)
+
+
+@nox.session(python=PYTHONS)
 def test(session: nox.Session) -> None:
     session.install(".[test]", "coverage")
     session.run("coverage", "run", "-m", "pytest", "tests/", *session.posargs)
     session.run("coverage", "report")
     session.run("coverage", "html")
 
 
-@nox.session(python="3.12")
+@nox.session(python=PYTHONS[0])
 def mypy(session: nox.Session) -> None:
     session.install(".[dev]")
     session.run("mypy", "src/pycyphal")
 
 
-@nox.session(python="3.12")
+@nox.session(python=PYTHONS[0])
 def lint(session: nox.Session) -> None:
     session.install("ruff")
     session.run("ruff", "check", "src", "tests")
 
 
-@nox.session(python="3.12")
+@nox.session(python=PYTHONS[0])
 def format(session: nox.Session) -> None:
     session.install("black")
     session.run("black", "--check", "--diff", "src", "tests")

src/pycyphal/__init__.py

@@ -2,7 +2,7 @@
 PyCyphal -- Python implementation of the Cyphal v1.1 session layer.
 
 All public entities of the library are directly exposed at the top level, here;
-implementations of the interfaces shown here are hidden and private.
+implementations of the public interfaces are hidden and private.
 One exception applies to transport-specific direct submodules like ``pycyphal.udp``, ``pycyphal.can`` --
 the user should import only the needed transports manually.
 
@@ -11,299 +11,9 @@
 
 from __future__ import annotations
 
-from abc import ABC, abstractmethod
-from dataclasses import dataclass
-import random
-import socket
-
-# We do not import anything from implementation modules here do there's no risk of circular dependencies.
-from ._common import DeliveryError as DeliveryError
-from ._common import Error as Error
-from ._common import Instant as Instant
-from ._common import LivenessError as LivenessError
-from ._common import NackError as NackError
-from ._common import SendError as SendError
-from ._common import Priority as Priority
-from ._common import Closable as Closable
-from ._common import SUBJECT_ID_PINNED_MAX as SUBJECT_ID_PINNED_MAX
+from ._api import *
 from ._transport import Transport as Transport
 from ._transport import TransportArrival as TransportArrival
 from ._transport import SubjectWriter as SubjectWriter
 
 __version__ = "2.0.0.dev0"
-
-
-class Topic(ABC):
-    """
-    Topics are managed automatically by the library, created and destroyed as necessary.
-    This is just a compact view to expose some auxiliary information.
-    """
-
-    @property
-    @abstractmethod
-    def hash(self) -> int:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def name(self) -> str:
-        raise NotImplementedError
-
-    @abstractmethod
-    def match(self, pattern: str) -> list[tuple[str, int]] | None:
-        """
-        If the pattern matches the topic name, returns the name segment substitutions needed to achieve the match.
-        None if there is no match. Empty list for verbatim subscribers (match only one topic), where pattern==name.
-        Each substitution is the segment and the index of the substitution character in the pattern.
-        """
-        raise NotImplementedError
-
-
-@dataclass(frozen=True)
-class Response:
-    """
-    Represents a response to a published message from the specified remote node.
-    Seqno is managed by the remote, incrementing by one with each response starting from zero.
-    """
-
-    timestamp: Instant
-    remote_id: int
-    seqno: int
-    message: bytes
-
-
-class ResponseStream(Closable, ABC):
-    """
-    Represents the expectation of response arrivals, used for one-off RPC and streaming.
-    Generates multiple async results, one per received response or per error.
-    Async iterator will continue to yield new messages until close()d, even after delivery/liveness errors.
-    """
-
-    def __aiter__(self) -> ResponseStream:
-        return self
-
-    async def __anext__(self) -> Response:
-        """
-        Blocks until response is received.
-        Raises DeliveryError if request could not be delivered by the deadline, LivenessError on response timeout,
-        SendError if failed to send the response before the deadline.
-        """
-        raise NotImplementedError
-
-
-class Publisher(Closable, ABC):
-    """
-    Represents the intent to send messages on a topic.
-    This is callable; an invocation triggers publication.
-    For publications that expect a response, use the ``request`` method.
-    """
-
-    @property
-    @abstractmethod
-    def topic(self) -> Topic:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def priority(self) -> Priority:
-        raise NotImplementedError
-
-    @priority.setter
-    @abstractmethod
-    def priority(self, priority: Priority) -> None:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def ack_timeout(self) -> float:
-        """
-        The ack timeout controls the baseline retransmission delay.
-        The number of attempts is controlled by the deadline specified at publish time.
-        """
-        raise NotImplementedError
-
-    @ack_timeout.setter
-    @abstractmethod
-    def ack_timeout(self, duration: float) -> None:
-        raise NotImplementedError
-
-    @abstractmethod
-    async def __call__(self, deadline: Instant, message: memoryview | bytes, *, reliable: bool = False) -> None:
-        """
-        Blocks at most until the deadline. Raises SendError if couldn't be sent before the deadline.
-        If reliable, DeliveryError will be raised unless acked by the remote subscribers before deadline.
-        """
-        raise NotImplementedError
-
-    @abstractmethod
-    async def request(
-        self, delivery_deadline: Instant, response_timeout: float, message: memoryview | bytes
-    ) -> ResponseStream:
-        """
-        Publish a message and expect responses. See ResponseStream for details.
-        """
-        raise NotImplementedError
-
-
-class Breadcrumb(ABC):
-    """
-    The breadcrumb can be used, optionally, to send responses RPC-style or streamed back to a message publisher.
-    Instances can be retained after message reception for as long as necessary.
-    One instance is shared across all subscribers receiving the same message, ensuring contiguous seqno across
-    all responses.
-
-    Responses are always sent at the same priority as that of the request.
-    Internally, the library tracks the seqno that starts at zero and is incremented with every response.
-
-    The set of (remote-ID, topic hash, message tag) forms a globally unique stream identification triplet,
-    which can be hashed down to a single number for convenience.
-    """
-
-    @property
-    @abstractmethod
-    def remote_id(self) -> int:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def topic(self) -> Topic:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def tag(self) -> int:
-        raise NotImplementedError
-
-    @abstractmethod
-    async def __call__(self, deadline: Instant, message: memoryview | bytes, *, reliable: bool = False) -> None:
-        """
-        Invoke multiple times to stream multiple responses.
-        Blocks at most until the deadline. Raises SendError if couldn't be sent before the deadline.
-        If reliable:
-        - DeliveryError will be raised unless acked by the remote subscribers before deadline.
-        - NackError will be raised if the remote is no longer accepting responses.
-        """
-        raise NotImplementedError
-
-
-@dataclass(frozen=True)
-class Arrival:
-    """
-    Represents a message received from a topic.
-    The breadcrumb allows sending responses back to the publisher, thus enabling RPC and streaming.
-    """
-
-    timestamp: Instant
-    breadcrumb: Breadcrumb
-    message: bytes
-
-
-class Subscriber(Closable, ABC):
-    @property
-    @abstractmethod
-    def pattern(self) -> str:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def verbatim(self) -> bool:
-        """
-        True if the pattern does not contain substitution segments named `*` and `>`.
-        """
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def timeout(self) -> float:
-        """
-        By default, the timeout is infinite, meaning that LivenessError will never be returned.
-        The user can override this as needed. Setting a non-finite timeout disables this feature.
-        """
-        raise NotImplementedError
-
-    @timeout.setter
-    @abstractmethod
-    def timeout(self, duration: float) -> None:
-        raise NotImplementedError
-
-    @abstractmethod
-    def substitutions(self, topic: Topic) -> list[tuple[str, int]] | None:
-        """
-        Pattern name segment substitutions needed to match the name of this subscriber to the name of the
-        specified topic. None if no match. Empty list for verbatim subscribers (match only one topic).
-        """
-        raise NotImplementedError
-
-    def __aiter__(self) -> Subscriber:
-        return self
-
-    @abstractmethod
-    async def __anext__(self) -> Arrival:
-        """
-        Raises LivenessError if messages cease arriving for longer than the timeout, unless timeout is non-finite.
-        """
-        raise NotImplementedError
-
-
-class Node(ABC):
-    """
-    The top-level entity that represents a node in the network.
-    """
-
-    @property
-    @abstractmethod
-    def home(self) -> str:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def namespace(self) -> str:
-        raise NotImplementedError
-
-    @abstractmethod
-    def advertise(self, name: str) -> Publisher:
-        """
-        Begin publishing on a topic; this also includes sending RPC requests.
-        """
-        raise NotImplementedError
-
-    @abstractmethod
-    def subscribe(self, name: str, *, reordering_window: float | None = None) -> Subscriber:
-        """
-        Receive messages from a single topic or multiple, if ``name`` is a pattern.
-        If the reordering window is set, ordered subscription is used that guarantees monotonically
-        increasing message tags, otherwise messages arrive ASAP.
-        """
-        raise NotImplementedError
-
-
-def new(transport: Transport, *, home: str = "", namespace: str = "") -> Node:
-    """
-    Construct a new node using the specified transport. This is the main entry point of the library.
-
-    The transport is constructed using one of the stock transport implementations like ``pycyphal.udp``,
-    depending on the needs of the application, or it could be custom.
-
-    Every node needs a unique nonempty home. If the home string is not provided, a random home will be generated.
-    If home ends with a `/`, a unique string will be automatically appended to generate a prefixed unique home;
-    e.g., `my_node_123` stays as-is; `my_node_123/` becomes like `my_node_123/123xyz`,
-    an empty string becomes a random string.
-
-    If namespace is not set, it is read from the CYPHAL_NAMESPACE environment variable if available,
-    otherwise it remains empty.
-    """
-    from ._hash import rapidhash
-    from ._node import NodeImpl
-
-    # Add random suffix if requested or generate pure random home. Leading/trailing separators will be normalized away.
-    home = home.strip() or "/"
-    if home.endswith("/"):
-        # A simple hex identifier where, for observability/diagnostic purposes, the first 20 bits (5 hex digits) are
-        # host-specific (same for all nodes running on this host), and the remaining 44 bits (11 digits) are random.
-        # The protocol doesn't care about this structure, it is just an optional default convention.
-        # noinspection PyTypeChecker
-        host = rapidhash(socket.gethostname().encode()) & 0xFFFFF
-        rand = random.getrandbits(44)
-        home += f"{host:05x}{rand:011x}"
-
-    return NodeImpl(transport, home=home, namespace=namespace)