Issue OpenCyphal#99 proposal

This is a draft PR to review a proposed change to the public APIs for pydsdl.

This is non-breaking change that adds a new public method and datatype

read_files - a file-oriented entry point to the front end. This takes a
list of target DSDL files allowing the user to maintain an explicit list
instead of depending on globular filesystem discovery.

DsdlFile – We define and publish a new abstract data type that encapsulates
all the information collected about and from a given dsdl file. This allows
a backend to associate datatypes with files for the purposes of managing
dependencies (e.g. this would allow the creation of .d files by a back end).

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "test/public_regulated_data_types"]
+	path = test/public_regulated_data_types
+	url = https://github.com/OpenCyphal/public_regulated_data_types.git

conftest.py

@@ -0,0 +1,71 @@
+#
+# Copyright (C) OpenCyphal Development Team  <opencyphal.org>
+# Copyright Amazon.com Inc. or its affiliates.
+# SPDX-License-Identifier: MIT
+#
+"""
+Configuration for pytest tests including fixtures and hooks.
+"""
+
+import tempfile
+from pathlib import Path
+from typing import Any, Optional
+
+import pytest
+
+
+# +-------------------------------------------------------------------------------------------------------------------+
+# | TEST FIXTURES
+# +-------------------------------------------------------------------------------------------------------------------+
+class TemporaryDsdlContext:
+    """
+    Powers the temp_dsdl_factory test fixture.
+    """
+    def __init__(self) -> None:
+        self._base_dir: Optional[Any] = None
+
+    def new_file(self, file_path: Path, text: str | None = None) -> Path:
+        if file_path.is_absolute():
+            raise ValueError(f"{file_path} is an absolute path. The test fixture requires relative paths to work.")
+        file = self.base_dir / file_path
+        file.parent.mkdir(parents=True, exist_ok=True)
+        if text is not None:
+            file.write_text(text)
+        return file
+
+    @property
+    def base_dir(self) -> Path:
+        if self._base_dir is None:
+            self._base_dir = tempfile.TemporaryDirectory()
+        return Path(self._base_dir.name).resolve()
+
+    def _test_path_finalizer(self) -> None:
+        """
+        Finalizer to clean up any temporary directories created during the test.
+        """
+        if self._base_dir is not None:
+            self._base_dir.cleanup()
+            del self._base_dir
+            self._base_dir = None
+
+@pytest.fixture(scope="function")
+def temp_dsdl_factory(request: pytest.FixtureRequest) -> Any:  # pylint: disable=unused-argument
+    """
+    Fixture for pydsdl tests that have to create files as part of the test. This object stays in-scope for a given
+    test method and does not requires a context manager in the test itself.
+
+    Call `new_file(path)` to create a new file path in the fixture's temporary directory. This will create all
+    uncreated parent directories but will _not_ create the file unless text is provided: `new_file(path, "hello")`
+    """
+    f = TemporaryDsdlContext()
+    request.addfinalizer(f._test_path_finalizer)  # pylint: disable=protected-access
+    return f
+
+
+
+@pytest.fixture
+def public_types() -> Path:
+    """
+    Path to the public regulated data types directory used for tests.
+    """
+    return Path("test") / "public_regulated_data_types" / "uavcan"

pydsdl/__init__.py

@@ -25,8 +25,9 @@
 _sys.path = [str(_Path(__file__).parent / "third_party")] + _sys.path
 
 # Never import anything that is not available here - API stability guarantees are only provided for the exposed items.
+from ._dsdl import PrintOutputHandler as PrintOutputHandler
 from ._namespace import read_namespace as read_namespace
-from ._namespace import PrintOutputHandler as PrintOutputHandler
+from ._namespace import read_files as read_files
 
 # Error model.
 from ._error import FrontendError as FrontendError

pydsdl/_data_type_builder.py

@@ -2,16 +2,12 @@
 # This software is distributed under the terms of the MIT License.
 # Author: Pavel Kirienko <[email protected]>
 
-from typing import Optional, Callable, Iterable
 import logging
 from pathlib import Path
-from . import _serializable
-from . import _expression
-from . import _error
-from . import _dsdl_definition
-from . import _parser
-from . import _data_schema_builder
-from . import _port_id_ranges
+from typing import Callable, Iterable, Optional
+
+from . import _data_schema_builder, _error, _expression, _parser, _port_id_ranges, _serializable
+from ._dsdl import DefinitionVisitor, DsdlFileBuildable
 
 
 class AssertionCheckFailureError(_error.InvalidDefinitionError):
@@ -42,21 +38,25 @@ class MissingSerializationModeError(_error.InvalidDefinitionError):
 
 
 class DataTypeBuilder(_parser.StatementStreamProcessor):
+
+    # pylint: disable=too-many-arguments
     def __init__(
         self,
-        definition: _dsdl_definition.DSDLDefinition,
-        lookup_definitions: Iterable[_dsdl_definition.DSDLDefinition],
+        definition: DsdlFileBuildable,
+        lookup_definitions: Iterable[DsdlFileBuildable],
+        definition_visitors: Iterable[DefinitionVisitor],
         print_output_handler: Callable[[int, str], None],
         allow_unregulated_fixed_port_id: bool,
     ):
         self._definition = definition
         self._lookup_definitions = list(lookup_definitions)
+        self._definition_visitors = definition_visitors
         self._print_output_handler = print_output_handler
         self._allow_unregulated_fixed_port_id = allow_unregulated_fixed_port_id
         self._element_callback = None  # type: Optional[Callable[[str], None]]
 
-        assert isinstance(self._definition, _dsdl_definition.DSDLDefinition)
-        assert all(map(lambda x: isinstance(x, _dsdl_definition.DSDLDefinition), lookup_definitions))
+        assert isinstance(self._definition, DsdlFileBuildable)
+        assert all(map(lambda x: isinstance(x, DsdlFileBuildable), lookup_definitions))
         assert callable(self._print_output_handler)
         assert isinstance(self._allow_unregulated_fixed_port_id, bool)
 
@@ -198,6 +198,7 @@ def resolve_versioned_data_type(self, name: str, version: _serializable.Version)
         del name
         found = list(filter(lambda d: d.full_name == full_name and d.version == version, self._lookup_definitions))
         if not found:
+
             # Play Sherlock to help the user with mistakes like https://forum.opencyphal.org/t/904/2
             requested_ns = full_name.split(_serializable.CompositeType.NAME_COMPONENT_SEPARATOR)[0]
             lookup_nss = set(x.root_namespace for x in self._lookup_definitions)
@@ -221,15 +222,20 @@ def resolve_versioned_data_type(self, name: str, version: _serializable.Version)
             raise _error.InternalError("Conflicting definitions: %r" % found)
 
         target_definition = found[0]
-        assert isinstance(target_definition, _dsdl_definition.DSDLDefinition)
+        for visitor in self._definition_visitors:
+            visitor(self._definition, target_definition)
+
+        assert isinstance(target_definition, DsdlFileBuildable)
         assert target_definition.full_name == full_name
         assert target_definition.version == version
         # Recursion is cool.
-        return target_definition.read(
+        dt = target_definition.read(
             lookup_definitions=self._lookup_definitions,
+            definition_visitors=self._definition_visitors,
             print_output_handler=self._print_output_handler,
             allow_unregulated_fixed_port_id=self._allow_unregulated_fixed_port_id,
         )
+        return dt
 
     def _queue_attribute(self, element_callback: Callable[[str], None]) -> None:
         self._flush_attribute("")
@@ -266,7 +272,9 @@ def _on_assert_directive(self, line_number: int, value: Optional[_expression.Any
         elif value is None:
             raise InvalidDirectiveError("Assert directive requires an expression")
         else:
-            raise InvalidDirectiveError("The assertion check expression must yield a boolean, not %s" % value.TYPE_NAME)
+            raise InvalidDirectiveError(
+                "The assertion check expression must yield a boolean, not %s" % value.TYPE_NAME
+            )
 
     def _on_extent_directive(self, line_number: int, value: Optional[_expression.Any]) -> None:
         if self._structs[-1].serialization_mode is not None:
@@ -300,7 +308,9 @@ def _on_union_directive(self, _ln: int, value: Optional[_expression.Any]) -> Non
         if self._structs[-1].union:
             raise InvalidDirectiveError("Duplicated union directive")
         if self._structs[-1].attributes:
-            raise InvalidDirectiveError("The union directive must be placed before the first " "attribute definition")
+            raise InvalidDirectiveError(
+                "The union directive must be placed before the first " "attribute definition"
+            )
         self._structs[-1].make_union()
 
     def _on_deprecated_directive(self, _ln: int, value: Optional[_expression.Any]) -> None: