diff --git a/libs/lib/xaal/lib/config.py b/libs/lib/xaal/lib/config.py
index 9a4cf6f396746f49866314b71ae49806a7319276..0989bc5b396f9244e394e22b10e264ac20345fe1 100644
--- a/libs/lib/xaal/lib/config.py
+++ b/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)
diff --git a/libs/lib/xaal/lib/helpers.py b/libs/lib/xaal/lib/helpers.py
index d2b4849604cb3f93bc2c381c991363930f47c70a..a03f904a8596a65398a06bb73559b4ead284a099 100644
--- a/libs/lib/xaal/lib/helpers.py
+++ b/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()
diff --git a/libs/lib/xaal/lib/messages.py b/libs/lib/xaal/lib/messages.py
index 343e0a0c15d6dc422eaa59dd55c7f9349847f44d..6280144f566429b82a1a45e90aba5763ee3cfb83 100644
--- a/libs/lib/xaal/lib/messages.py
+++ b/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