mirror of
https://github.com/craigerl/aprsd.git
synced 2026-01-19 03:55:42 -05:00
fixed some formatting issues with filter.py
This commit is contained in:
parent
9c06957943
commit
e821fbf53e
@ -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
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user