add plugged() for manually-aligned ports

masque/file/utils.py

@@ -8,14 +8,70 @@ import logging
 import tempfile
 import shutil
 from contextlib import contextmanager
+from pprint import pformat
+from itertools import chain
 
-from .. import Pattern, PatternError
+from .. import Pattern, PatternError, Library, LibraryError
 from ..shapes import Polygon, Path
 
 
 logger = logging.getLogger(__name__)
 
 
+def preflight(
+        lib: Library,
+        sort: bool = True,
+        allow_dangling_refs: bool | None = None,
+        allow_named_layers: bool = True,
+        prune_empty_patterns: bool = False,
+        ) -> Library:
+    """
+    Run a standard set of useful operations and checks, usually done immediately prior
+    to writing to a file (or immediately after reading).
+
+    Args:
+        sort: Whether to sort the patterns based on their names, and sort the pattern contents.
+            Default True. Useful for reproducible builds.
+        allow_dangling_refs: If `None` (default), warns about any refs to patterns that are not
+            in the provided library. If `True`, no check is performed; if `False`, a `LibraryError`
+            is raised instead.
+        allow_named_layers: If `False`, raises a `PatternError` if any layer is referred to by
+            a string instead of a number (or tuple).
+        prune_empty_patterns: Runs `Library.prune_empty()`, recursively deleting any empty patterns.
+
+    Returns:
+        `lib` or an equivalent sorted library
+    """
+    if sort:
+        lib = Library(dict(sorted(
+            (nn: pp.sort()) for nn, pp in lib.items()
+            )))
+
+    if not allow_dangling_refs:
+        refs = lib.referenced_patterns()
+        dangling = refs - set(lib.keys())
+        msg = 'Dangling refs in found: ' + pformat(dangling)
+        if allow_dangling_refs is None:
+            logger.warning(msg)
+        else:
+            raise LibraryError(msg)
+
+    if not allow_named_layers:
+        named_layers = defaultdict(set)
+        for name, pat in lib.items():
+            for layer in chain(pat.shapes.keys(), pat.labels.keys()):
+                if isinstance(layer, str):
+                    named_layers[name].add(layer)
+        raise PatternError('Non-numeric layers found:' + pformat(named_layers))
+
+    if prune_empty_patterns:
+        pruned = lib.prune_empty()
+        logger.info(f'Preflight pruned {len(pruned)} empty patterns')
+        logger.debug('Pruned: ' + pformat(pruned))
+
+    return lib
+
+
 def mangle_name(name: str) -> str:
     """
     Sanitize a name.

masque/pattern.py

@@ -297,6 +297,31 @@ class Pattern(PortList, AnnotatableImpl, Mirrorable):
 
         return True
 
+    def sort(self) -> Self:
+        """
+        Sort the element dicts (shapes, labels, refs) and their contained lists.
+        This is primarily useful for making builds more reproducible.
+
+        Returns:
+            self
+        """
+        self.refs = dict(sorted(
+            (tgt, sorted(rrs)) for tgt, rrs in self.refs.items()
+            ))
+        self.labels = dict(sorted(
+            (layer, sorted(lls)) for layer, lls in self.labels.items(),
+            key=lambda kk, vv: layer2key(ll),
+            ))
+        self.shapes = dict(sorted(
+            (layer, sorted(sss)) for layer, sss in self.shapes.items(),
+            key=lambda kk, vv: layer2key(ll),
+            ))
+
+        self.ports = dict(sorted(self.ports.items()))
+        self.annotations = dict(sorted(self.annotations.items()))
+
+        return self
+
     def append(self, other_pattern: 'Pattern') -> Self:
         """
         Appends all shapes, labels and refs from other_pattern to self's shapes,

masque/ports.py

@@ -5,6 +5,7 @@ import logging
 import functools
 from collections import Counter
 from abc import ABCMeta, abstractmethod
+from itertools import chain
 
 import numpy
 from numpy import pi
@@ -269,6 +270,75 @@ class PortList(metaclass=ABCMeta):
         self.ports.update(new_ports)
         return self
 
+    def plugged(
+            self,
+            connections: dict[str, str],
+            ) -> Self:
+        """
+        Verify that the ports specified by `connections` are coincident and have opposing
+        rotations, then remove the ports.
+
+        This is used when ports have been "manually" aligned as part of some other routing,
+        but for whatever reason were not eliminated via `plug()`.
+
+        Args:
+            connections: Pairs of ports which "plug" each other (same offset, opposing directions)
+
+        Returns:
+            self
+
+        Raises:
+            `PortError` if the ports are not properly aligned.
+        """
+        a_names, b_names = list(zip(*connections.items()))
+        a_ports = [self.ports[pp] for pp in a_names]
+        b_ports = [self.ports[pp] for pp in b_names]
+
+        a_types = [pp.ptype for pp in a_ports]
+        b_types = [pp.ptype for pp in b_ports]
+        type_conflicts = numpy.array([at != bt and at != 'unk' and bt != 'unk'
+                                      for at, bt in zip(a_types, b_types)])
+
+        if type_conflicts.any():
+            msg = 'Ports have conflicting types:\n'
+            for nn, (k, v) in enumerate(connections.items()):
+                if type_conflicts[nn]:
+                    msg += f'{k} | {a_types[nn]}:{b_types[nn]} | {v}\n'
+            msg = ''.join(traceback.format_stack()) + '\n' + msg
+            warnings.warn(msg, stacklevel=2)
+
+        a_offsets = numpy.array([pp.offset for pp in a_ports])
+        b_offsets = numpy.array([pp.offset for pp in b_ports])
+        a_rotations = numpy.array([pp.rotation if pp.rotation is not None else 0 for pp in a_ports])
+        b_rotations = numpy.array([pp.rotation if pp.rotation is not None else 0 for pp in b_ports])
+        a_has_rot = numpy.array([pp.rotation is not None for pp in a_ports], dtype=bool)
+        b_has_rot = numpy.array([pp.rotation is not None for pp in b_ports], dtype=bool)
+        has_rot = a_has_rot & b_has_rot
+
+        if has_rot.any():
+            rotations = numpy.mod(a_rotations - b_rotations - pi, 2 * pi)
+            rotations[~has_rot] = rotations[has_rot][0]
+
+            if not numpy.allclose(rotations, 0):
+                rot_deg = numpy.rad2deg(rotations)
+                msg = 'Port orientations do not match:\n'
+                for nn, (k, v) in enumerate(connections.items()):
+                    if not numpy.isclose(rot_deg[nn], 0):
+                        msg += f'{k} | {rot_deg[nn]:g} | {v}\n'
+                raise PortError(msg)
+
+        translations = a_offsets - b_offsets
+        if not numpy.allclose(translations, 0):
+            msg = 'Port translations do not match:\n'
+            for nn, (k, v) in enumerate(connections.items()):
+                if not numpy.allclose(translations[nn], 0):
+                    msg += f'{k} | {translations[nn]} | {v}\n'
+            raise PortError(msg)
+
+        for pp in chain(a_names, b_names):
+            del self.ports[pp]
+        return self
+
     def check_ports(
             self,
             other_names: Iterable[str],