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] = []
)
from types import SimpleNamespace
+from .protomodule import ProtoClass
+
__all__ = (
"Stream",
"class_by_prefix",
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], ...] = ()
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__
)
import zmq
from . import common
+from .protomodule import ProtoModule
from .zmsg import Bcast, Resp
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] = []
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] = []
--- /dev/null
+""" 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]]:
+ ...
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] = []
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] = []
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]] = []
Union,
)
+from .protomodule import ProtoClass
+
__all__ = (
"Stream",
"class_by_prefix",
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], ...] = ()
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__
)
projects / loctrkd.git / commitdiff