Add io configs

Author: pavel-kirienko

cynic.py

@@ -75,6 +75,12 @@
 Raw encoding disables escape expansion for DATA after `--`; raw decoding emits undelimited payload bytes.
 With an RPC DSDL, `req` encodes the request and decodes the response; `pub` and `sub` use the request part.
 
+Per-topic `--io` defaults can be configured in the TOML `io` table; this is convenient because topics usually don't
+change their data types.
+Each topic name key here is a glob pattern so that a single entry can match multiple topics.
+The keys are matched against fully resolved topic names (after namespace prefixing) so that this works with
+pattern subscribers as well.
+
 ### DSDL
 
 Use `--enc=path/to/File.1.1.dsdl`, `--dec=path/to/File.1.1.dsdl`, or `--io=path/to/File.1.1.dsdl`.
@@ -93,6 +99,17 @@
 The following are equivalent: `{foo: [1,2,3], bar: 456}`, `[[1,2,3], 456]`.
 Likewise, for single-field objects: `{foo: 456}`, `[456]`, `456`.
 
+`cynic.toml` example:
+
+```toml
+# cynic.toml
+# ...skip...
+dsdl_root = ["/path/to/types"]
+[io]
+"command.i8" = "/path/to/types/primitive/Integer8.1.0.dsdl"
+"sensor.*"   = "/path/to/types/sensor/Sample.1.0.dsdl"
+```
+
 ## Serve files
 
 Run a file server on the specified topic(s) using `zubax.file.Read` out of the current working directory.
@@ -102,6 +119,23 @@
 cn fs my/file/topic /other/topic
 ```
 
+## Examples of cynic.toml
+
+A few practical config examples to help one get started are collected here.
+
+```toml
+can = "slcan0"
+dsdl_root = [
+    "/home/pavel/zubax/zubax_dsdl/zubax/",
+]
+[io]
+"feedback" = "zubax/zubax_dsdl/zubax/fluxgrip/Feedback.1.1.dsdl"
+"/com.zubax.fluxgrip/*/zubax.manage" = "zubax/zubax_dsdl/zubax/fluxgrip/Manage.0.1.dsdl"
+```
+
+With the above config, one can go like `cn sub feedback` or
+`cn req com.zubax.fluxgrip/0ef4cccfc40c9393/zubax.manage -- 'get_info: {}'`, very concise.
+
 ---
 
 Distributed under the MIT License. Author: Pavel Kirienko `pavel@opencyphal.org`
@@ -121,9 +155,10 @@
 import tomllib
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from functools import partial
+from fnmatch import fnmatchcase
+from functools import cache, partial
 from pathlib import Path
-from typing import Any, Callable
+from typing import Any, Callable, Protocol
 import pydsdl  # type: ignore[import-untyped]
 import yaml  # type: ignore[import-untyped]
 from pycyphal2 import Arrival, Error, Instant, LivenessError, Node, Publisher, Response, Subscriber, Topic, Transport
@@ -137,7 +172,11 @@
 DEFAULT_TIMEOUT = 10.0
 SCOUT_PATTERN = "/>"
 
-Input = Callable[[], bytes]
+
+class Input(Protocol):
+    def __call__(self, topics: tuple[str, ...] | None = None) -> bytes: ...
+
+
 Output = Callable[[Arrival | Response | Exception, Topic | None], None]
 
 
@@ -206,6 +245,28 @@ def unescape(text: str) -> bytes:
     return text.encode("latin-1").decode("unicode_escape").encode("latin-1")
 
 
+def match_topic_io(defaults: dict[str, str], topic: str) -> str | None:
+    """Select a configured I/O default for a concrete wire topic name."""
+    if topic in defaults:
+        return defaults[topic]
+    matches = [(pattern, selector) for pattern, selector in defaults.items() if fnmatchcase(topic, pattern)]
+    selectors = {selector for _, selector in matches}
+    if len(selectors) > 1:
+        details = ", ".join(f"{pattern!r}={selector!r}" for pattern, selector in matches)
+        raise ValueError(f"ambiguous I/O defaults for topic {topic!r}: {details}")
+    return next(iter(selectors), None)
+
+
+def match_input_io(defaults: dict[str, str], topics: tuple[str, ...]) -> str | None:
+    """Select the one configured input default shared by every topic in an invocation."""
+    matches = [(topic, match_topic_io(defaults, topic)) for topic in topics]
+    selectors = {selector for _, selector in matches}
+    if len(selectors) > 1:
+        details = ", ".join(f"{topic!r}={selector!r}" for topic, selector in matches)
+        raise ValueError(f"topics select different input defaults: {details}; use -i or --io")
+    return next(iter(selectors), None)
+
+
 def load_dsdl(file: Path, roots: list[Path]) -> pydsdl.CompositeType:
     file, roots = file.resolve(), [x.resolve() for x in roots]
     if not file.is_file():
@@ -402,7 +463,7 @@ async def cmd_pub(node: Node, config: PubConfig) -> int:
     pubs = [node.advertise(x) for x in config.topics]
     try:
         deadline = Instant.now() + config.timeout
-        message = config.input()
+        message = config.input(tuple(pub.topic.name for pub in pubs))
         await asyncio.gather(*(pub(deadline, message, reliable=True) for pub in pubs))
         logger.info("Published on %s", config.topics)
     except Exception as ex:
@@ -437,7 +498,7 @@ async def request(pub: Publisher, config: ReqConfig, message: bytes) -> int:
 async def cmd_req(node: Node, config: ReqConfig) -> int:
     pubs = [node.advertise(x) for x in config.topics]
     try:
-        message = config.input()
+        message = config.input(tuple(pub.topic.name for pub in pubs))
         counts = await asyncio.gather(*(request(pub, config, message) for pub in pubs))
     except Exception as ex:
         config.output(ex, None)
@@ -660,15 +721,15 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
     sub.add_parser("ls", parents=[options], help="discover topics, print each once as it appears; exits when idle")
 
     argv = sys.argv[1:]
-    source: Input = sys.stdin.buffer.read
-    input_ = source
+    source: Callable[[], bytes] = sys.stdin.buffer.read
+    expand_escapes = False
     if data_supplied := "--" in argv:
         separator = argv.index("--")
         if len(argv) != separator + 2:
             parser.error("expected exactly one DATA argument after --")
         data = os.fsencode(argv[-1])
         source = lambda: data
-        input_ = lambda: unescape(data.decode("latin-1"))
+        expand_escapes = True
         argv = argv[:separator]
     args = parser.parse_args(argv)
     if args.command is None:
@@ -707,6 +768,11 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
         isinstance(values["dsdl_root"], list) and all(isinstance(x, str) for x in values["dsdl_root"])
     ):
         parser.error(f"{path!r}: 'dsdl_root' has an invalid type")
+    if "io" in values and not (
+        isinstance(values["io"], dict)
+        and all(isinstance(pattern, str) and isinstance(selector, str) for pattern, selector in values["io"].items())
+    ):
+        parser.error(f"{path!r}: 'io' has an invalid type")
 
     env_bitrate: int | None = None
     if value := os.environ.get("CYNIC_BITRATE"):
@@ -727,6 +793,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
     timeout = float(cli.get("timeout", values.get("timeout", DEFAULT_TIMEOUT)))
     verbose = cli.get("verbose", values.get("verbose", 0))
     command = cli["command"]
+    io_defaults: dict[str, str] = values.get("io", {})
     enc, dec, io = cli.get("enc"), cli.get("dec"), cli.get("io")
     for name, selector, commands in (
         ("enc", enc, ("pub", "req")),
@@ -746,8 +813,13 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
     roots: list[Path] | None = None
     schemas: dict[Path, pydsdl.CompositeType] = {}
 
+    directional_schemas: dict[tuple[str, bool], pydsdl.CompositeType] = {}
+
     def resolve_schema(selector: str, output_direction: bool) -> pydsdl.CompositeType:
         nonlocal roots
+        key = (selector, output_direction)
+        if key in directional_schemas:
+            return directional_schemas[key]
         try:
             file = Path(selector).resolve()
             if roots is None:
@@ -768,22 +840,58 @@ def resolve_schema(selector: str, output_direction: bool) -> pydsdl.CompositeTyp
                 schemas[file] = load_dsdl(file, roots)
             schema = schemas[file]
         except (OSError, RuntimeError, ValueError, pydsdl.Error) as ex:
-            parser.error(f"cannot load DSDL {selector!r}: {ex}")
+            raise ValueError(f"cannot load DSDL {selector!r}: {ex}") from ex
         if isinstance(schema, pydsdl.ServiceType):
-            return schema.response_type if command == "req" and output_direction else schema.request_type
+            schema = schema.response_type if command == "req" and output_direction else schema.request_type
+        directional_schemas[key] = schema
         return schema
 
-    if enc == "raw":
-        input_ = source
-    elif enc is not None:
-        input_schema = resolve_schema(enc, False)
-        input_ = lambda: pydsdl.serialize(input_schema, yaml.safe_load(source()), relaxed=True)
+    # Explicit CLI selectors retain eager validation. Configured selectors are resolved only if a matching wire
+    # topic is actually used, which keeps unrelated entries from breaking a command.
+    for selector, output_direction in ((enc, False), (dec, True)):
+        if selector is not None and selector != "raw":
+            try:
+                resolve_schema(selector, output_direction)
+            except ValueError as ex:
+                parser.error(str(ex))
+
+    def input_(topics: tuple[str, ...] | None = None) -> bytes:
+        selector = enc
+        if selector is None and command in ("pub", "req") and topics is not None:
+            selector = match_input_io(io_defaults, topics)
+        data = source()
+        if selector == "raw":
+            return data
+        if selector is not None:
+            schema = resolve_schema(selector, False)
+            return bytes(pydsdl.serialize(schema, yaml.safe_load(data), relaxed=True))
+        return unescape(data.decode("latin-1")) if expand_escapes else data
+
+    @cache
+    def configured_output(topic: str) -> Output:
+        selector = match_topic_io(io_defaults, topic)
+        if selector == "raw":
+            return raw_output
+        if selector is not None:
+            return partial(dsdl_output, resolve_schema(selector, True))
+        return json_output
+
+    def dynamic_output(value: Arrival | Response | Exception, topic: Topic | None) -> None:
+        wire_topic = (
+            value.breadcrumb.topic.name if isinstance(value, Arrival) else topic.name if topic is not None else None
+        )
+        (configured_output(wire_topic) if wire_topic is not None else json_output)(value, topic)
 
-    output: Output = json_output
+    output: Output
     if dec == "raw":
         output = raw_output
     elif dec is not None:
         output = partial(dsdl_output, resolve_schema(dec, True))
+    elif command in ("sub", "req") and io_defaults:
+        output = dynamic_output
+    else:
+        output = json_output
+
     common = dict(
         can=can,
         bitrate=1_000_000 if bitrate is None else bitrate,