From cbb7603aebedf517d7b7c61ff6de02d3e7193409 Mon Sep 17 00:00:00 2001 From: Eugene Crosser Date: Tue, 19 Jul 2022 00:00:40 +0200 Subject: [PATCH] protocols: make "interface" module `protomodule.py` contains metaclass for the protocol classes, class definition for a protocol module, and a "Protocol" (PEP-544) class for a protocol definition class. --- loctrkd/__main__.py | 11 +-- loctrkd/beesure.py | 56 +------------- loctrkd/collector.py | 38 +--------- loctrkd/mkgpx.py | 11 +-- loctrkd/protomodule.py | 166 +++++++++++++++++++++++++++++++++++++++++ loctrkd/qry.py | 11 +-- loctrkd/watch.py | 11 +-- loctrkd/wsgateway.py | 15 +--- loctrkd/zx303proto.py | 56 +------------- 9 files changed, 180 insertions(+), 195 deletions(-) create mode 100644 loctrkd/protomodule.py diff --git a/loctrkd/__main__.py b/loctrkd/__main__.py index 14c33f3..10a2bf8 100644 --- a/loctrkd/__main__.py +++ b/loctrkd/__main__.py @@ -11,21 +11,12 @@ from typing import Any, cast, List, Tuple, Type, Union import zmq from . import common +from .protomodule import ProtoModule from .zmsg import Bcast, Resp log = getLogger("loctrkd") -class ProtoModule: - @staticmethod - def proto_handled(proto: str) -> bool: - ... - - @staticmethod - def class_by_prefix(prefix: str) -> Any: - ... - - pmods: List[ProtoModule] = [] diff --git a/loctrkd/beesure.py b/loctrkd/beesure.py index 3a71f5d..7f689c5 100755 --- a/loctrkd/beesure.py +++ b/loctrkd/beesure.py @@ -22,6 +22,8 @@ from typing import ( ) from types import SimpleNamespace +from .protomodule import ProtoClass + __all__ = ( "Stream", "class_by_prefix", @@ -186,63 +188,13 @@ def pblist(x: Union[str, List[Tuple[str, str]]]) -> List[Tuple[str, str]]: return lx -class MetaPkt(type): - """ - For each class corresponding to a message, automatically create - two nested classes `In` and `Out` that also inherit from their - "nest". Class attribute `IN_KWARGS` defined in the "nest" is - copied to the `In` nested class under the name `KWARGS`, and - likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS` - to the nested class `Out`. In addition, method `encode` is - defined in both classes equal to `in_encode()` and `out_encode()` - respectively. - """ - - if TYPE_CHECKING: - - def __getattr__(self, name: str) -> Any: - pass - - def __setattr__(self, name: str, value: Any) -> None: - pass - - def __new__( - cls: Type["MetaPkt"], - name: str, - bases: Tuple[type, ...], - attrs: Dict[str, Any], - ) -> "MetaPkt": - newcls = super().__new__(cls, name, bases, attrs) - newcls.In = super().__new__( - cls, - name + ".In", - (newcls,) + bases, - { - "KWARGS": newcls.IN_KWARGS, - "decode": newcls.in_decode, - "encode": newcls.in_encode, - }, - ) - newcls.Out = super().__new__( - cls, - name + ".Out", - (newcls,) + bases, - { - "KWARGS": newcls.OUT_KWARGS, - "decode": newcls.out_decode, - "encode": newcls.out_encode, - }, - ) - return newcls - - class Respond(Enum): NON = 0 # Incoming, no response needed INL = 1 # Birirectional, use `inline_response()` EXT = 2 # Birirectional, use external responder -class BeeSurePkt(metaclass=MetaPkt): +class BeeSurePkt(ProtoClass): RESPOND = Respond.NON # Do not send anything back by default IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () @@ -574,7 +526,7 @@ def proto_handled(proto: str) -> bool: return proto.startswith(PROTO_PREFIX) -def proto_name(obj: Union[MetaPkt, BeeSurePkt]) -> str: +def proto_name(obj: Union[Type[BeeSurePkt], BeeSurePkt]) -> str: return PROTO_PREFIX + ( obj.__class__.__name__ if isinstance(obj, BeeSurePkt) else obj.__name__ ) diff --git a/loctrkd/collector.py b/loctrkd/collector.py index 6fc606a..75cd33b 100644 --- a/loctrkd/collector.py +++ b/loctrkd/collector.py @@ -18,6 +18,7 @@ from typing import Any, cast, Dict, List, Optional, Set, Tuple, Union import zmq from . import common +from .protomodule import ProtoModule from .zmsg import Bcast, Resp log = getLogger("loctrkd/collector") @@ -25,43 +26,6 @@ log = getLogger("loctrkd/collector") MAXBUFFER: int = 4096 -class ProtoModule: - class Stream: - def recv(self, segment: bytes) -> List[Union[bytes, str]]: - ... - - def close(self) -> bytes: - ... - - @staticmethod - def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes: - ... - - @staticmethod - def probe_buffer(buffer: bytes) -> bool: - ... - - @staticmethod - def parse_message(packet: bytes, is_incoming: bool = True) -> Any: - ... - - @staticmethod - def inline_response(packet: bytes) -> Optional[bytes]: - ... - - @staticmethod - def is_goodbye_packet(packet: bytes) -> bool: - ... - - @staticmethod - def imei_from_packet(packet: bytes) -> Optional[str]: - ... - - @staticmethod - def proto_of_message(packet: bytes) -> str: - ... - - pmods: List[ProtoModule] = [] diff --git a/loctrkd/mkgpx.py b/loctrkd/mkgpx.py index 4ea7435..6c1ce00 100644 --- a/loctrkd/mkgpx.py +++ b/loctrkd/mkgpx.py @@ -14,20 +14,11 @@ from sys import argv from typing import Any, cast, List, Tuple from . import common +from .protomodule import ProtoModule log = getLogger("loctrkd/mkgpx") -class ProtoModule: - @staticmethod - def proto_handled(proto: str) -> bool: - ... - - @staticmethod - def parse_message(packet: bytes, is_incoming: bool = True) -> Any: - ... - - pmods: List[ProtoModule] = [] diff --git a/loctrkd/protomodule.py b/loctrkd/protomodule.py new file mode 100644 index 0000000..1a6f0fc --- /dev/null +++ b/loctrkd/protomodule.py @@ -0,0 +1,166 @@ +""" Things the module implementing a protocol exports """ + +from typing import ( + Any, + Callable, + Dict, + List, + Optional, + Protocol, + _ProtocolMeta, # How not to cheat here?! + Tuple, + Type, + TYPE_CHECKING, + Union, +) + + +class MetaPkt(type): + """ + For each class corresponding to a message, automatically create + two nested classes `In` and `Out` that also inherit from their + "nest". Class attribute `IN_KWARGS` defined in the "nest" is + copied to the `In` nested class under the name `KWARGS`, and + likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS` + to the nested class `Out`. In addition, methods `encode` and + `decode` are defined in both classes equal to `in_{en|de}code()` + and `out_{en|de}code()` respectively. + """ + + if TYPE_CHECKING: + + def __getattr__(self, name: str) -> Any: + pass + + def __setattr__(self, name: str, value: Any) -> None: + pass + + def in_decode(self, *args: Any) -> None: + ... + + def out_decode(self, *args: Any) -> None: + ... + + def in_encode(self, *args: Any) -> Any: + ... + + def out_encode(self, *args: Any) -> Any: + ... + + def __new__( + cls: Type["MetaPkt"], + name: str, + bases: Tuple[type, ...], + attrs: Dict[str, Any], + ) -> "MetaPkt": + newcls = super().__new__(cls, name, bases, attrs) + newcls.In = super().__new__( + cls, + name + ".In", + (newcls,) + bases, + { + "KWARGS": newcls.IN_KWARGS, + "decode": newcls.in_decode, + "encode": newcls.in_encode, + }, + ) + newcls.Out = super().__new__( + cls, + name + ".Out", + (newcls,) + bases, + { + "KWARGS": newcls.OUT_KWARGS, + "decode": newcls.out_decode, + "encode": newcls.out_encode, + }, + ) + return newcls + + +# Have to do this to prevent incomprehensible error message: +# TypeError: metaclass conflict: the metaclass of a derived class \ +# must be a (non-strict) subclass of the metaclasses of all its bases +class _MetaProto(_ProtocolMeta, MetaPkt): + pass + + +class ProtoClass(Protocol, metaclass=_MetaProto): + IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () + OUT_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () + + class In: + def __init__(self, *args: Any, **kwargs: Any) -> None: + ... + + def encode(self) -> bytes: + ... + + def decode(self, *args: Any, **kwargs: Any) -> None: + ... + + @property + def packed(self) -> bytes: + ... + + class Out: + def __init__(self, *args: Any, **kwargs: Any) -> None: + ... + + def encode(self) -> bytes: + ... + + def decode(self, *args: Any, **kwargs: Any) -> None: + ... + + @property + def packed(self) -> bytes: + ... + + +class ProtoModule: + class Stream: + def recv(self, segment: bytes) -> List[Union[bytes, str]]: + ... + + def close(self) -> bytes: + ... + + @staticmethod + def enframe(buffer: bytes, imei: Optional[str] = None) -> bytes: + ... + + @staticmethod + def exposed_protos() -> List[Tuple[str, bool]]: + ... + + @staticmethod + def probe_buffer(buffer: bytes) -> bool: + ... + + @staticmethod + def parse_message(packet: bytes, is_incoming: bool = True) -> Any: + ... + + @staticmethod + def inline_response(packet: bytes) -> Optional[bytes]: + ... + + @staticmethod + def is_goodbye_packet(packet: bytes) -> bool: + ... + + @staticmethod + def imei_from_packet(packet: bytes) -> Optional[str]: + ... + + @staticmethod + def proto_of_message(packet: bytes) -> str: + ... + + @staticmethod + def proto_handled(proto: str) -> bool: + ... + + @staticmethod + def class_by_prefix(prefix: str) -> Union[Type[ProtoClass], List[str]]: + ... diff --git a/loctrkd/qry.py b/loctrkd/qry.py index 650830e..6745cf2 100644 --- a/loctrkd/qry.py +++ b/loctrkd/qry.py @@ -10,20 +10,11 @@ from sys import argv from typing import Any, cast, List, Tuple from . import common +from .protomodule import ProtoModule log = getLogger("loctrkd/qry") -class ProtoModule: - @staticmethod - def proto_handled(proto: str) -> bool: - ... - - @staticmethod - def parse_message(packet: bytes, is_incoming: bool = True) -> Any: - ... - - pmods: List[ProtoModule] = [] diff --git a/loctrkd/watch.py b/loctrkd/watch.py index b2b5c06..006ec06 100644 --- a/loctrkd/watch.py +++ b/loctrkd/watch.py @@ -8,21 +8,12 @@ from typing import Any, cast, List import zmq from . import common +from .protomodule import ProtoModule from .zmsg import Bcast log = getLogger("loctrkd/watch") -class ProtoModule: - @staticmethod - def proto_handled(proto: str) -> bool: - ... - - @staticmethod - def parse_message(packet: bytes, is_incoming: bool = True) -> Any: - ... - - pmods: List[ProtoModule] = [] diff --git a/loctrkd/wsgateway.py b/loctrkd/wsgateway.py index 8f2c648..01a8042 100644 --- a/loctrkd/wsgateway.py +++ b/loctrkd/wsgateway.py @@ -23,25 +23,12 @@ import zmq from . import common from .evstore import initdb, fetch +from .protomodule import ProtoModule from .zmsg import Bcast, topic log = getLogger("loctrkd/wsgateway") -class ProtoModule: - @staticmethod - def parse_message(packet: bytes, is_incoming: bool = True) -> Any: - ... - - @staticmethod - def exposed_protos() -> List[Tuple[str, bool]]: - ... - - @staticmethod - def proto_handled(proto: str) -> bool: - ... - - htmlfile = None pmods: List[ProtoModule] = [] selector: List[Tuple[bool, str]] = [] diff --git a/loctrkd/zx303proto.py b/loctrkd/zx303proto.py index bd19e10..454214e 100755 --- a/loctrkd/zx303proto.py +++ b/loctrkd/zx303proto.py @@ -31,6 +31,8 @@ from typing import ( Union, ) +from .protomodule import ProtoClass + __all__ = ( "Stream", "class_by_prefix", @@ -247,63 +249,13 @@ def l3int(x: Union[str, List[int]]) -> List[int]: return lx -class MetaPkt(type): - """ - For each class corresponding to a message, automatically create - two nested classes `In` and `Out` that also inherit from their - "nest". Class attribute `IN_KWARGS` defined in the "nest" is - copied to the `In` nested class under the name `KWARGS`, and - likewise, `OUT_KWARGS` of the nest class is copied as `KWARGS` - to the nested class `Out`. In addition, method `encode` is - defined in both classes equal to `in_encode()` and `out_encode()` - respectively. - """ - - if TYPE_CHECKING: - - def __getattr__(self, name: str) -> Any: - pass - - def __setattr__(self, name: str, value: Any) -> None: - pass - - def __new__( - cls: Type["MetaPkt"], - name: str, - bases: Tuple[type, ...], - attrs: Dict[str, Any], - ) -> "MetaPkt": - newcls = super().__new__(cls, name, bases, attrs) - newcls.In = super().__new__( - cls, - name + ".In", - (newcls,) + bases, - { - "KWARGS": newcls.IN_KWARGS, - "decode": newcls.in_decode, - "encode": newcls.in_encode, - }, - ) - newcls.Out = super().__new__( - cls, - name + ".Out", - (newcls,) + bases, - { - "KWARGS": newcls.OUT_KWARGS, - "decode": newcls.out_decode, - "encode": newcls.out_encode, - }, - ) - return newcls - - class Respond(Enum): NON = 0 # Incoming, no response needed INL = 1 # Birirectional, use `inline_response()` EXT = 2 # Birirectional, use external responder -class GPS303Pkt(metaclass=MetaPkt): +class GPS303Pkt(ProtoClass): RESPOND = Respond.NON # Do not send anything back by default PROTO: int IN_KWARGS: Tuple[Tuple[str, Callable[[Any], Any], Any], ...] = () @@ -885,7 +837,7 @@ def proto_handled(proto: str) -> bool: return proto.startswith(PROTO_PREFIX) -def proto_name(obj: Union[MetaPkt, GPS303Pkt]) -> str: +def proto_name(obj: Union[Type[GPS303Pkt], GPS303Pkt]) -> str: return PROTO_PREFIX + ( obj.__class__.__name__ if isinstance(obj, GPS303Pkt) else obj.__name__ ) -- 2.43.0