Skip to content
Snippets Groups Projects
Commit 0f9e33c2 authored by KERDREUX Jerome's avatar KERDREUX Jerome
Browse files

Format

parent cf0d0e0f
No related branches found
No related tags found
1 merge request!1First try of type hints
Hide whitespace changes
Inline Side-by-side
libs/lib/xaal/lib/devices.py
+ 69
− 67
View file @ 0f9e33c2
......@@ -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)
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment