diff --git a/aprsd/packets/filter.py b/aprsd/packets/filter.py index 152366b..93eeb32 100644 --- a/aprsd/packets/filter.py +++ b/aprsd/packets/filter.py @@ -1,16 +1,16 @@ import logging -from typing import Callable, Protocol, runtime_checkable, Union, Dict +from typing import Callable, Dict, Protocol, Union, runtime_checkable from aprsd.packets import core from aprsd.utils import singleton -LOG = logging.getLogger("APRSD") +LOG = logging.getLogger('APRSD') @runtime_checkable class PacketFilterProtocol(Protocol): - """Protocol API for a packet filter class. - """ + """Protocol API for a packet filter class.""" + def filter(self, packet: type[core.Packet]) -> Union[type[core.Packet], None]: """When we get a packet from the network. @@ -22,20 +22,23 @@ class PacketFilterProtocol(Protocol): @singleton class PacketFilter: - def __init__(self): self.filters: Dict[str, Callable] = {} def register(self, packet_filter: Callable) -> None: if not isinstance(packet_filter, PacketFilterProtocol): - raise TypeError(f"class {packet_filter} is not a PacketFilterProtocol object") + raise TypeError( + f'class {packet_filter} is not a PacketFilterProtocol object' + ) if packet_filter not in self.filters: self.filters[packet_filter] = packet_filter() def unregister(self, packet_filter: Callable) -> None: if not isinstance(packet_filter, PacketFilterProtocol): - raise TypeError(f"class {packet_filter} is not a PacketFilterProtocol object") + raise TypeError( + f'class {packet_filter} is not a PacketFilterProtocol object' + ) if packet_filter in self.filters: del self.filters[packet_filter] @@ -51,8 +54,12 @@ class PacketFilter: for packet_filter in self.filters: try: if not self.filters[packet_filter].filter(packet): - LOG.debug(f"{self.filters[packet_filter].__class__.__name__} dropped {packet.__class__.__name__}:{packet.human_info}") + LOG.debug( + f'{self.filters[packet_filter].__class__.__name__} dropped {packet.__class__.__name__}:{packet.human_info}' + ) return None except Exception as ex: - LOG.error(f"{packet_filter.__clas__.__name__} failed filtering packet {packet.__class__.__name__} : {ex}") + LOG.error( + f'{packet_filter.__class__.__name__} failed filtering packet {packet.__class__.__name__} : {ex}' + ) return packet