From 00e185b4e7ca5abed9f56282924e183b1201ea33 Mon Sep 17 00:00:00 2001 From: Hemna Date: Wed, 20 Mar 2024 15:36:26 -0400 Subject: [PATCH] Fixed some tox errors related to mypy --- aprsd/plugin.py | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/aprsd/plugin.py b/aprsd/plugin.py index 905d583..c760d56 100644 --- a/aprsd/plugin.py +++ b/aprsd/plugin.py @@ -1,4 +1,5 @@ -# The base plugin class +from __future__ import annotations + import abc import importlib import inspect @@ -42,7 +43,7 @@ class APRSDPluginSpec: """A hook specification namespace.""" @hookspec - def filter(self, packet: packets.core.Packet): + def filter(self, packet: type[packets.Packet]): """My special little hook that you can customize.""" @@ -118,11 +119,11 @@ class APRSDPluginBase(metaclass=abc.ABCMeta): thread.stop() @abc.abstractmethod - def filter(self, packet: type[packets.core.Packet]) -> str | packets.core.MessagePacket: + def filter(self, packet: type[packets.Packet]) -> str | packets.MessagePacket: pass @abc.abstractmethod - def process(self, packet: type[packets.core.Packet]): + def process(self, packet: type[packets.Packet]): """This is called when the filter passes.""" @@ -154,7 +155,7 @@ class APRSDWatchListPluginBase(APRSDPluginBase, metaclass=abc.ABCMeta): LOG.warning("Watch list enabled, but no callsigns set.") @hookimpl - def filter(self, packet: type[packets.core.Packet]) -> str | packets.core.MessagePacket: + def filter(self, packet: type[packets.Packet]) -> str | packets.MessagePacket: result = packets.NULL_MESSAGE if self.enabled: wl = watch_list.WatchList() @@ -206,14 +207,14 @@ class APRSDRegexCommandPluginBase(APRSDPluginBase, metaclass=abc.ABCMeta): self.enabled = True @hookimpl - def filter(self, packet: packets.core.MessagePacket) -> str | packets.core.MessagePacket: + def filter(self, packet: packets.MessagePacket) -> str | packets.MessagePacket: LOG.info(f"{self.__class__.__name__} called") if not self.enabled: result = f"{self.__class__.__name__} isn't enabled" LOG.warning(result) return result - if not isinstance(packet, packets.core.MessagePacket): + if not isinstance(packet, packets.MessagePacket): LOG.warning(f"{self.__class__.__name__} Got a {packet.__class__.__name__} ignoring") return packets.NULL_MESSAGE @@ -226,7 +227,7 @@ class APRSDRegexCommandPluginBase(APRSDPluginBase, metaclass=abc.ABCMeta): # and is an APRS message format and has a message. if ( tocall == CONF.callsign - and isinstance(packet, packets.core.MessagePacket) + and isinstance(packet, packets.MessagePacket) and message ): if re.search(self.command_regex, message, re.IGNORECASE): @@ -269,7 +270,7 @@ class HelpPlugin(APRSDRegexCommandPluginBase): def help(self): return "Help: send APRS help or help " - def process(self, packet: packets.core.MessagePacket): + def process(self, packet: packets.MessagePacket): LOG.info("HelpPlugin") # fromcall = packet.get("from") message = packet.message_text @@ -469,12 +470,12 @@ class PluginManager: LOG.info("Completed Plugin Loading.") - def run(self, packet: packets.core.MessagePacket): + def run(self, packet: packets.MessagePacket): """Execute all the plugins run method.""" with self.lock: return self._pluggy_pm.hook.filter(packet=packet) - def run_watchlist(self, packet: packets.core.Packet): + def run_watchlist(self, packet: packets.Packet): with self.lock: return self._watchlist_pm.hook.filter(packet=packet)