diff --git a/libs/lib/xaal/lib/devices.py b/libs/lib/xaal/lib/devices.py
index 28cd320d8e291bf0df724e9fcabdece7e31d1dc7..bf494359c24a0c880202c49ed61861be620ff211 100644
--- a/libs/lib/xaal/lib/devices.py
+++ b/libs/lib/xaal/lib/devices.py
@@ -18,44 +18,45 @@
# along with xAAL. If not, see <http://www.gnu.org/licenses/>.
#
-import time
import logging
-from tabulate import tabulate
-from typing import Optional, Union, Any
+import time
+import typing
+from typing import Any, Optional, Union
+from tabulate import tabulate
-from . import config
-from . import tools
-from . import bindings
+from . import bindings, config, tools
from .exceptions import DeviceError
-from .types import DeviceT, EngineT, AsyncEngineT
+if typing.TYPE_CHECKING:
+ from .aioengine import AsyncEngine
+ from .devices import Device
+ from .engine import Engine
logger = logging.getLogger(__name__)
class Attribute(object):
-
- def __init__(self, name, dev: Optional[DeviceT] = None, default: Any = None):
+ def __init__(self, name, dev: Optional["Device"] = None, default: Any = None):
self.name = name
self.default = default
- self.device: Optional[Device] = dev
+ self.device: Optional["Device"] = dev
self.__value = default
@property
- def value(self):
+ def value(self) -> Any:
return self.__value
@value.setter
- def value(self, value):
+ def value(self, value: Any):
if value != self.__value and self.device:
eng = self.device.engine
if eng:
eng.add_attributes_change(self)
- logger.debug("Attr change %s %s=%s" % (self.device.address,self.name,value))
+ logger.debug("Attr change %s %s=%s" % (self.device.address, self.name, value))
self.__value = value
- def __repr__(self): # pragma: no cover
+ def __repr__(self) -> str: # pragma: no cover
return f"<{self.__module__}.Attribute {self.name} at 0x{id(self):x}>"
@@ -63,22 +64,23 @@ class Attributes(list):
"""Devices owns a attributes list. This list also have dict-like access"""
def __getitem__(self, value):
- if isinstance(value,int):
- return list.__getitem__(self,value)
+ if isinstance(value, int):
+ return list.__getitem__(self, value)
for k in self:
- if (value == k.name):
+ if value == k.name:
return k.value
raise KeyError(value)
def __setitem__(self, name, value):
- if isinstance(name,int):
- return list.__setitem__(self,name,value)
+ if isinstance(name, int):
+ return list.__setitem__(self, name, value)
for k in self:
- if (name == k.name):
+ if name == k.name:
k.value = value
return
raise KeyError(name)
+
class Device(object):
__slots__ = [
"__dev_type",
@@ -105,32 +107,33 @@ class Device(object):
self,
dev_type: str,
addr: Optional[bindings.UUID] = None,
- engine: Union[EngineT, AsyncEngineT, None] = None,
+ engine: Union["Engine", "AsyncEngine", None] = None,
):
-
# xAAL internal attributes for a device
- self.dev_type = dev_type # xaal dev_type
- self.address = addr # xaal addr
- self.group_id = None # group devices
- self.vendor_id = None # vendor ID ie : ACME
- self.product_id = None # product ID
- self.hw_id = None # hardware info
- self.version = None # product release
- self.url = None # product URL
- self.schema = None # schema URL
- self.info = None # additionnal info
+ self.dev_type = dev_type # xaal dev_type
+ self.address = addr # xaal addr
+ self.group_id = None # group devices
+ self.vendor_id = None # vendor ID ie : ACME
+ self.product_id = None # product ID
+ self.hw_id = None # hardware info
+ self.version = None # product release
+ self.url = None # product URL
+ self.schema = None # schema URL
+ self.info = None # additionnal info
# Unsupported stuffs
self.unsupported_attributes = []
self.unsupported_methods = []
self.unsupported_notifications = []
# Alive management
- self.alive_period = config.alive_timer # time in sec between two alive
+ self.alive_period = config.alive_timer # time in sec between two alive
self.next_alive = 0
# Default attributes & methods
self.__attributes = Attributes()
- self.methods = {'get_attributes' : self._get_attributes,
- 'get_description': self._get_description }
+ self.methods = {
+ "get_attributes": self._get_attributes,
+ "get_description": self._get_description,
+ }
self.engine = engine
@property
@@ -149,7 +152,7 @@ class Device(object):
@version.setter
def version(self, value: Any):
- # version must be a string
+ # version must be a string
if value:
self.__version = "%s" % value
else:
@@ -181,7 +184,7 @@ class Device(object):
# attributes
def new_attribute(self, name: str, default: Any = None):
- attr = Attribute(name,self,default)
+ attr = Attribute(name, self, default)
self.add_attribute(attr)
return attr
@@ -207,23 +210,23 @@ class Device(object):
@attributes.setter
def attributes(self, values: Attributes):
- if isinstance(values,Attributes):
+ if isinstance(values, Attributes):
self.__attributes = values
else:
raise DeviceError("Invalid attributes list, use class Attributes)")
def add_method(self, name: str, func: Any):
- self.methods.update({name:func})
+ self.methods.update({name: func})
def get_methods(self) -> dict:
return self.methods
def update_alive(self):
- """ update the alive timimg"""
+ """update the alive timimg"""
self.next_alive = time.time() + self.alive_period
def get_timeout(self) -> int:
- """ return Alive timeout used for isAlive msg"""
+ """return Alive timeout used for isAlive msg"""
return 2 * self.alive_period
#####################################################
@@ -233,26 +236,25 @@ class Device(object):
print("= Device: %s" % self)
# info & description
r = []
- r.append(['dev_type',self.dev_type])
- r.append(['address',self.address])
- for k,v in self._get_description().items():
- r.append([k,v])
- print(tabulate(r,tablefmt="fancy_grid"))
+ r.append(["dev_type", self.dev_type])
+ r.append(["address", self.address])
+ for k, v in self._get_description().items():
+ r.append([k, v])
+ print(tabulate(r, tablefmt="fancy_grid"))
# attributes
if len(self._get_attributes()) > 0:
r = []
- for k,v in self._get_attributes().items():
- r.append([k,str(v)])
- print(tabulate(r,tablefmt="fancy_grid"))
+ for k, v in self._get_attributes().items():
+ r.append([k, str(v)])
+ print(tabulate(r, tablefmt="fancy_grid"))
# methods
if len(self.methods) > 0:
r = []
- for k,v in self.methods.items():
- r.append([k,v.__name__])
- print(tabulate(r,tablefmt="fancy_grid"))
-
+ for k, v in self.methods.items():
+ r.append([k, v.__name__])
+ print(tabulate(r, tablefmt="fancy_grid"))
def __repr__(self) -> str:
return f"<xaal.Device {id(self):x} {self.address} {self.dev_type}>"
@@ -263,27 +265,27 @@ class Device(object):
def _get_description(self) -> dict:
result = {}
if self.vendor_id:
- result['vendor_id'] = self.vendor_id
+ result["vendor_id"] = self.vendor_id
if self.product_id:
- result['product_id'] = self.product_id
+ result["product_id"] = self.product_id
if self.version:
- result['version'] = self.version
+ result["version"] = self.version
if self.url:
- result['url'] = self.url
+ result["url"] = self.url
if self.schema:
- result['schema'] = self.schema
+ result["schema"] = self.schema
if self.info:
- result['info'] = self.info
+ result["info"] = self.info
if self.hw_id:
- result['hw_id'] = self.hw_id
+ result["hw_id"] = self.hw_id
if self.group_id:
- result['group_id'] = self.group_id
+ result["group_id"] = self.group_id
if self.unsupported_methods:
- result['unsupported_methods'] = self.unsupported_methods
+ result["unsupported_methods"] = self.unsupported_methods
if self.unsupported_notifications:
- result['unsupported_notifications'] = self.unsupported_notifications
+ result["unsupported_notifications"] = self.unsupported_notifications
if self.unsupported_attributes:
- result['unsupported_attributes'] = self.unsupported_attributes
+ result["unsupported_attributes"] = self.unsupported_attributes
return result
def _get_attributes(self, _attributes=None):
@@ -313,7 +315,7 @@ class Device(object):
result.update({attr.name: attr.value})
return result
- def send_notification(self, notification:str, body:dict={}):
- """ queue an notification, this is just a method helper """
+ def send_notification(self, notification: str, body: dict = {}):
+ """queue an notification, this is just a method helper"""
if self.engine:
- self.engine.send_notification(self,notification,body)
+ self.engine.send_notification(self, notification, body)