[Port] add more logging to aid in debug

This commit is contained in:
jan 2026-02-16 12:23:40 -08:00
commit cf822c7dcf
3 changed files with 36 additions and 0 deletions

View file

@ -17,6 +17,7 @@ from .error import PortError, format_stacktrace
logger = logging.getLogger(__name__)
port_logger = logging.getLogger('masque.ports')
@functools.total_ordering
@ -207,6 +208,19 @@ class PortList(metaclass=ABCMeta):
def ports(self, value: dict[str, Port]) -> None:
pass
def _log_port_update(self, name: str) -> None:
""" Log the current state of the named port """
port_logger.info("Port %s: %s", name, self.ports[name])
def _log_port_removal(self, name: str) -> None:
""" Log that the named port has been removed """
port_logger.info("Port %s: removed", name)
def _log_bulk_update(self, label: str) -> None:
""" Log all current ports at DEBUG level """
for name, port in self.ports.items():
port_logger.debug("%s: Port %s: %s", label, name, port)
@overload
def __getitem__(self, key: str) -> Port:
pass
@ -260,6 +274,7 @@ class PortList(metaclass=ABCMeta):
raise PortError(f'Port {name} already exists.')
assert name not in self.ports
self.ports[name] = value
self._log_port_update(name)
return self
def rename_ports(
@ -286,11 +301,22 @@ class PortList(metaclass=ABCMeta):
if duplicates:
raise PortError(f'Unrenamed ports would be overwritten: {duplicates}')
for kk, vv in mapping.items():
if vv is None:
self._log_port_removal(kk)
elif vv != kk:
self._log_port_removal(kk)
renamed = {vv: self.ports.pop(kk) for kk, vv in mapping.items()}
if None in renamed:
del renamed[None]
self.ports.update(renamed) # type: ignore
for vv in mapping.values():
if vv is not None:
self._log_port_update(vv)
return self
def add_port_pair(
@ -319,6 +345,8 @@ class PortList(metaclass=ABCMeta):
}
self.check_ports(names)
self.ports.update(new_ports)
self._log_port_update(names[0])
self._log_port_update(names[1])
return self
def plugged(
@ -388,6 +416,7 @@ class PortList(metaclass=ABCMeta):
for pp in chain(a_names, b_names):
del self.ports[pp]
self._log_port_removal(pp)
return self
def check_ports(