Type hints for messages

libs/lib/xaal/lib/config.py

@@ -27,7 +27,7 @@ else:
     self.conf_dir = os.path.expanduser("~") + '/.xaal'
 
 
-def load_config(name='xaal.ini'):
+def load_config(name: str ='xaal.ini'):
     filename = os.path.join(self.conf_dir, name)
     if not os.path.isfile(filename):
         print("Unable to load xAAL config file [%s]" % filename)

libs/lib/xaal/lib/helpers.py

@@ -7,6 +7,7 @@ import logging
 import logging.handlers
 import os
 import time
+from typing import Optional, Any
 import coloredlogs
 from decorator import decorator
 
@@ -30,16 +31,16 @@ def timeit(method, *args, **kwargs):
     logger.debug('%r (%r, %r) %2.6f sec' % (method.__name__, args, kwargs, te-ts))
     return result
 
-def set_console_title(value):
+def set_console_title(value: str):
     # set xterm title
     print("\x1B]0;xAAL => %s\x07" % value, end='\r')
 
-def setup_console_logger(level=config.log_level):
+def setup_console_logger(level: str = config.log_level):
     fmt = '%(asctime)s %(name)-25s %(funcName)-18s %(levelname)-8s %(message)s'
     #fmt = '[%(name)s] %(funcName)s %(levelname)s: %(message)s'
     coloredlogs.install(level=level,fmt=fmt)
 
-def setup_file_logger(name,level=config.log_level, filename = None):
+def setup_file_logger(name: str,level: str =config.log_level, filename: Optional[str] = None):
     filename = filename or os.path.join(config.log_path,'%s.log' % name)
     formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(name)s - %(message)s')
     handler = logging.handlers.RotatingFileHandler(filename, 'a', 10000, 1, 'utf8')
@@ -56,7 +57,7 @@ def setup_file_logger(name,level=config.log_level, filename = None):
 # 
 # Default arguments console_log and file_log are (and should) never be used.
 # ---------------------------------------------------------------------------
-def run_package(pkg_name, pkg_setup, console_log = True, file_log=False):
+def run_package(pkg_name: str, pkg_setup: Any, console_log: bool = True, file_log: bool =False):
     if console_log:
         set_console_title(pkg_name)
         setup_console_logger()

libs/lib/xaal/lib/messages.py

@@ -18,6 +18,7 @@
 #  along with xAAL. If not, see <http://www.gnu.org/licenses/>.
 #
 
+from xaal.lib.types import DeviceT
 from . import tools
 from . import config
 from .bindings import UUID
@@ -32,12 +33,32 @@ import pysodium
 import struct
 import sys
 
+from typing import Any, Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .devices import Device
+
+
 import logging
 logger = logging.getLogger(__name__)
 
 ALIVE_ADDR = UUID("00000000-0000-0000-0000-000000000000")
 
 
+class MessageType(Enum):
+    NOTIFY  = 0
+    REQUEST = 1
+    REPLY   = 2
+
+
+class MessageAction(Enum):
+    ALIVE             = "alive"
+    IS_ALIVE          = "is_alive"
+    ATTRIBUTES_CHANGE = "attributes_change"
+    GET_ATTRIBUTES    = "get_attributes"
+    GET_DESCRIPTION   = "get_description"
+
+
 class MessageFactory(object):
     """Message Factory:
     - Build xAAL message
@@ -47,7 +68,7 @@ class MessageFactory(object):
     def __init__(self, cipher_key):
         self.cipher_key = cipher_key    # key encode / decode message built from passphrase
 
-    def encode_msg(self, msg):
+    def encode_msg(self, msg: 'Message'):
         """Apply security layer and return encode MSG in CBOR
         :param msg: xAAL msg instance
         :type msg: Message
@@ -82,7 +103,7 @@ class MessageFactory(object):
         pkt = cbor.dumps(result)
         return pkt
 
-    def decode_msg(self, data, filter_func=None):
+    def decode_msg(self, data: bytes, filter_func: Any =None) -> Optional['Message']:
         """Decode incoming CBOR data and De-Ciphering
         :param data: data received from the multicast bus
         :type data: cbor
@@ -160,7 +181,14 @@ class MessageFactory(object):
     #####################################################
     # MSG builder
     #####################################################
-    def build_msg(self, dev=None, targets=[], msg_type=None, action=None, body=None):
+    def build_msg(
+        self,
+        dev: Optional[Device] = None,
+        targets: list = [],
+        msg_type: Optional[MessageType] = None,
+        action: Optional[str] = None,
+        body: Optional[dict] = None,
+    ):
         """the build method takes in parameters :
         -A device
         -The list of targets of the message
@@ -187,7 +215,7 @@ class MessageFactory(object):
         data = self.encode_msg(message)
         return data
 
-    def build_alive_for(self, dev, timeout=0):
+    def build_alive_for(self, dev: Device, timeout: int =0) -> bytes:
         """Build Alive message for a given device
         timeout = 0 is the minimum value
         """
@@ -196,7 +224,7 @@ class MessageFactory(object):
         message = self.build_msg(dev=dev, targets=[], msg_type=MessageType.NOTIFY, action=MessageAction.ALIVE.value, body=body)
         return message
 
-    def build_error_msg(self, dev, errcode, description=None):
+    def build_error_msg(self, dev: Device, errcode:int , description: Optional[str] =None):
         """Build a Error message"""
         message = Message()
         body = {}
@@ -207,19 +235,6 @@ class MessageFactory(object):
         return message
 
 
-class MessageType(Enum):
-    NOTIFY  = 0
-    REQUEST = 1
-    REPLY   = 2
-
-
-class MessageAction(Enum):
-    ALIVE             = "alive"
-    IS_ALIVE          = "is_alive"
-    ATTRIBUTES_CHANGE = "attributes_change"
-    GET_ATTRIBUTES    = "get_attributes"
-    GET_DESCRIPTION   = "get_description"
-
 
 class Message(object):
     """Message object used for incomming & outgoint message"""
@@ -237,11 +252,11 @@ class Message(object):
         self.body = {}                       # message body
 
     @property
-    def targets(self):
+    def targets(self) -> list:
         return self.__targets
 
     @targets.setter
-    def targets(self, values):
+    def targets(self, values: list):
         if not isinstance(values, list):
             raise MessageError("Expected a list for targetsList, got %s" % (type(values),))
         for uid in values:
@@ -249,7 +264,7 @@ class Message(object):
                 raise MessageError("Bad target addr: %s" % uid)
         self.__targets = values
 
-    def targets_as_string(self):
+    def targets_as_string(self) -> list:
         return [str(k) for k in self.targets]
 
     def dump(self):
@@ -271,57 +286,57 @@ class Message(object):
             r.append(["body", tmp])
         print(tabulate(r, headers=["Fied", "Value"], tablefmt="psql"))
 
-    def __repr__(self):
+    def __repr__(self) -> str:
         return f"<xaal.Message {id(self):x} {self.source} {self.dev_type} {self.msg_type} {self.action}>"
 
-    def is_request(self):
+    def is_request(self) -> bool:
         if MessageType(self.msg_type) == MessageType.REQUEST:
             return True
         return False
 
-    def is_reply(self):
+    def is_reply(self) -> bool:
         if MessageType(self.msg_type) == MessageType.REPLY:
             return True
         return False
 
-    def is_notify(self):
+    def is_notify(self) -> bool:
         if MessageType(self.msg_type) == MessageType.NOTIFY:
             return True
         return False
 
-    def is_alive(self):
+    def is_alive(self) -> bool:
         if self.is_notify() and self.action == MessageAction.ALIVE.value:
             return True
         return False
 
-    def is_request_isalive(self):
+    def is_request_isalive(self) -> bool:
         if self.is_request() and self.action == MessageAction.IS_ALIVE.value:
             return True
         return False
 
-    def is_attributes_change(self):
+    def is_attributes_change(self) -> bool:
         if self.is_notify() and self.action == MessageAction.ATTRIBUTES_CHANGE.value:
             return True
         return False
 
-    def is_get_attribute_reply(self):
+    def is_get_attribute_reply(self) -> bool:
         if self.is_reply() and self.action == MessageAction.GET_ATTRIBUTES.value:
             return True
         return False
 
-    def is_get_description_reply(self):
+    def is_get_description_reply(self) -> bool:
         if self.is_reply() and self.action == MessageAction.GET_DESCRIPTION.value:
             return True
         return False
 
 
-def build_nonce(data):
+def build_nonce(data: tuple) -> bytes:
     """ Big-Endian, time in seconds and time in microseconds """
     nonce = struct.pack('>QL', data[0], data[1])
     return nonce
 
 
-def build_timestamp():
+def build_timestamp() -> list:
     """Return array [seconds since epoch, microseconds since last seconds] Time = UTC+0000"""
     epoch = datetime.datetime.fromtimestamp(0, datetime.UTC)
     timestamp = datetime.datetime.now(datetime.UTC) - epoch