create lockfile class

Author: songmeo

pycyphal/dsdl/__init__.py

@@ -38,6 +38,8 @@
 from ._support_wrappers import to_builtin as to_builtin
 from ._support_wrappers import update_from_builtin as update_from_builtin
 
+from ._lockfile import Lockfile
+
 
 def generate_package(*args, **kwargs):  # type: ignore  # pragma: no cover
     """Deprecated alias of :func:`compile`."""

pycyphal/dsdl/_compiler.py

@@ -15,7 +15,7 @@
 import nunavut
 import nunavut.lang
 import nunavut.jinja
-
+from ._lockfile import Lockfile
 
 _AnyPath = Union[str, pathlib.Path]
 
@@ -149,101 +149,83 @@ def compile(  # pylint: disable=redefined-builtin
     composite_types: list[pydsdl.CompositeType] = []
     output_directory = pathlib.Path(pathlib.Path.cwd() if output_directory is None else output_directory).resolve()
 
-    if root_namespace_directory is not None:
-        root_namespace_directory = pathlib.Path(root_namespace_directory).resolve()
-        if root_namespace_directory.parent == output_directory:
-            # https://github.com/OpenCyphal/pycyphal/issues/133 and https://github.com/OpenCyphal/pycyphal/issues/127
-            raise ValueError(
-                "The specified destination may overwrite the DSDL root namespace directory. "
-                "Consider specifying a different output directory instead."
+    lockfile = Lockfile(root_namespace_directory, root_namespace_name, output_directory)
+    lockfile_created = lockfile.create()
+
+    if lockfile_created:
+        if root_namespace_directory is not None:
+            root_namespace_directory = pathlib.Path(root_namespace_directory).resolve()
+            if root_namespace_directory.parent == output_directory:
+                # https://github.com/OpenCyphal/pycyphal/issues/133 and https://github.com/OpenCyphal/pycyphal/issues/127
+                raise ValueError(
+                    "The specified destination may overwrite the DSDL root namespace directory. "
+                    "Consider specifying a different output directory instead."
+                )
+
+            # Read the DSDL definitions
+            composite_types = pydsdl.read_namespace(
+                root_namespace_directory=str(root_namespace_directory),
+                lookup_directories=list(map(str, lookup_directories or [])),
+                allow_unregulated_fixed_port_id=allow_unregulated_fixed_port_id,
+            )
+            if not composite_types:
+                _logger.info("Root namespace directory %r does not contain DSDL definitions", root_namespace_directory)
+                return None
+            (root_namespace_name,) = set(map(lambda x: x.root_namespace, composite_types))  # type: ignore
+            _logger.info("Read %d definitions from root namespace %r", len(composite_types), root_namespace_name)
+
+            # Generate code
+            assert isinstance(output_directory, pathlib.Path)
+            root_ns = nunavut.build_namespace_tree(
+                types=composite_types,
+                root_namespace_dir=str(root_namespace_directory),
+                output_dir=str(output_directory),
+                language_context=language_context,
+            )
+            code_generator = nunavut.jinja.DSDLCodeGenerator(
+                namespace=root_ns,
+                generate_namespace_types=nunavut.YesNoDefault.YES,
+                followlinks=True,
+            )
+            code_generator.generate_all()
+            _logger.info(
+                "Generated %d types from the root namespace %r in %.1f seconds",
+                len(composite_types),
+                root_namespace_name,
+                time.monotonic() - started_at,
+            )
+        else:
+            root_ns = nunavut.build_namespace_tree(
+                types=[],
+                root_namespace_dir=str(""),
+                output_dir=str(output_directory),
+                language_context=language_context,
             )
 
-        # Read the DSDL definitions
-        composite_types = pydsdl.read_namespace(
-            root_namespace_directory=str(root_namespace_directory),
-            lookup_directories=list(map(str, lookup_directories or [])),
-            allow_unregulated_fixed_port_id=allow_unregulated_fixed_port_id,
-        )
-        if not composite_types:
-            _logger.info("Root namespace directory %r does not contain DSDL definitions", root_namespace_directory)
-            return None
-        (root_namespace_name,) = set(map(lambda x: x.root_namespace, composite_types))  # type: ignore
-        _logger.info("Read %d definitions from root namespace %r", len(composite_types), root_namespace_name)
-
-        # Generate code
-        assert isinstance(output_directory, pathlib.Path)
-        root_ns = nunavut.build_namespace_tree(
-            types=composite_types,
-            root_namespace_dir=str(root_namespace_directory),
-            output_dir=str(output_directory),
-            language_context=language_context,
-        )
-        code_generator = nunavut.jinja.DSDLCodeGenerator(
+        support_generator = nunavut.jinja.SupportGenerator(
             namespace=root_ns,
-            generate_namespace_types=nunavut.YesNoDefault.YES,
-            followlinks=True,
-        )
-        code_generator.generate_all()
-        _logger.info(
-            "Generated %d types from the root namespace %r in %.1f seconds",
-            len(composite_types),
-            root_namespace_name,
-            time.monotonic() - started_at,
-        )
-    else:
-        root_ns = nunavut.build_namespace_tree(
-            types=[],
-            root_namespace_dir=str(""),
-            output_dir=str(output_directory),
-            language_context=language_context,
         )
+        support_generator.generate_all()
 
-    lockfile = f"{root_namespace_name}.lock"
-    if not root_namespace_name:
-        lockfile = "support.lock"
-    lockfile_path = f"{output_directory}/{lockfile}"
-
-    while True:
-        if not os.path.exists(lockfile_path):
-            try:
-                pathlib.Path(output_directory).mkdir(parents=True, exist_ok=True)
-                pathlib.Path(lockfile_path).touch()
+        # A minor UX improvement; see https://github.com/OpenCyphal/pycyphal/issues/115
+        for p in sys.path:
+            if pathlib.Path(p).resolve() == pathlib.Path(output_directory):
                 break
-            except PermissionError:
-                _logger.warning("Unable to create lockfile at %s", lockfile_path)
-        else:
-            time.sleep(1)
-            if pathlib.Path.exists(output_directory / pathlib.Path(root_namespace_name)):
-                return GeneratedPackageInfo(
-                    path=pathlib.Path(output_directory) / pathlib.Path(root_namespace_name),
-                    models=composite_types,
-                    name=root_namespace_name,
-                )
-
-    support_generator = nunavut.jinja.SupportGenerator(
-        namespace=root_ns,
-    )
-    support_generator.generate_all()
-
-    # A minor UX improvement; see https://github.com/OpenCyphal/pycyphal/issues/115
-    for p in sys.path:
-        if pathlib.Path(p).resolve() == pathlib.Path(output_directory):
-            break
-    else:
-        if os.name == "nt":
-            quick_fix = f'Quick fix: `$env:PYTHONPATH += ";{output_directory.resolve()}"`'
-        elif os.name == "posix":
-            quick_fix = f'Quick fix: `export PYTHONPATH="{output_directory.resolve()}"`'
         else:
-            quick_fix = "Quick fix is not available for this OS."
-        _logger.info(
-            "Generated package is stored in %r, which is not in Python module search path list. "
-            "The package will fail to import unless you add the destination directory to sys.path or PYTHONPATH. %s",
-            str(output_directory),
-            quick_fix,
-        )
+            if os.name == "nt":
+                quick_fix = f'Quick fix: `$env:PYTHONPATH += ";{output_directory.resolve()}"`'
+            elif os.name == "posix":
+                quick_fix = f'Quick fix: `export PYTHONPATH="{output_directory.resolve()}"`'
+            else:
+                quick_fix = "Quick fix is not available for this OS."
+            _logger.info(
+                "Generated package is stored in %r, which is not in Python module search path list. "
+                "The package will fail to import unless you add the destination directory to sys.path or PYTHONPATH. %s",
+                str(output_directory),
+                quick_fix,
+            )
 
-    pathlib.Path(lockfile_path).unlink()
+        lockfile.remove()
 
     return GeneratedPackageInfo(
         path=pathlib.Path(output_directory) / pathlib.Path(root_namespace_name),

pycyphal/dsdl/_lockfile.py

@@ -0,0 +1,36 @@
+import logging
+import pathlib
+import time
+
+_logger = logging.getLogger(__name__)
+
+
+class Lockfile:
+
+    def __init__(self, root_namespace_directory, root_namespace_name, output_directory) -> None:
+        self.output_directory = output_directory
+        self.root_namespace_name = root_namespace_name
+        self.root_namespace_directory = root_namespace_directory
+
+    @property
+    def lockfile_path(self) -> str:
+        return f"{self.output_directory}/{self.root_namespace_name if self.root_namespace_name else 'support'}.lock"
+
+    def create(self):
+        root_namespace_name = self.root_namespace_name
+        output_directory = self.output_directory
+        lockfile_path = self.lockfile_path
+
+        while True:
+            try:
+                pathlib.Path(output_directory).mkdir(parents=True, exist_ok=True)
+                fp = open(lockfile_path, "x")
+                fp.close()
+                return True
+            except FileExistsError:
+                time.sleep(1)
+                if pathlib.Path.exists(output_directory / pathlib.Path(root_namespace_name)):
+                    return False
+
+    def remove(self):
+        pathlib.Path(self.lockfile_path).unlink()