- Now support Python typing

git-svn-id: https://redmine.imt-atlantique.fr/svn/xaal/code/Python/branches/0.7@2997 b32b6428-25c9-4566-ad07-03861ab6144f

apps/homeassistant/xaal/binary_sensor.py

+import imp
 import logging
 
+from typing import Literal
+
 from homeassistant.components.binary_sensor import BinarySensorEntity, BinarySensorDeviceClass
 from homeassistant.const import STATE_ON, STATE_OFF
 
-from .core import XAALEntity, EntityFactory, async_setup_factory
+from .core import XAALEntity, EntityFactory, MonitorDevice, async_setup_factory
+from xaal.lib import Message
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -13,7 +17,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
 
 class Factory(EntityFactory):
 
-    def new_entity(self, device):
+    def new_entity(self, device: MonitorDevice) -> bool:
         entity = None
 
         if device.dev_type.startswith('motion.'):
@@ -37,7 +41,7 @@ class Factory(EntityFactory):
 class XAALBinarySensorEntity(XAALEntity, BinarySensorEntity):
 
     @property
-    def state(self):
+    def state(self) -> Literal["on", "off"] | None:
         try:
             attr = getattr(self,'_xaal_attribute')
             value = self.get_attribute(attr)
@@ -68,7 +72,7 @@ class Button(XAALBinarySensorEntity):
         _LOGGER.warning(f"Button event: {self.entity_id}")
         self.hass.bus.fire("xaal_event", {'entity_id': self.entity_id, "click_type": click_type})
 
-    def handle_notification(self, msg):
+    def handle_notification(self, msg: Message):
         if msg.action == 'click':
             self.click_event('single')
         if msg.action == 'double_click':

apps/homeassistant/xaal/bridge.py

 
 import asyncio
+from typing import Dict, List, Any, Type
 
 from homeassistant.core import HomeAssistant
-from xaal.lib import AsyncEngine, tools, MessageType
+from xaal.lib import AsyncEngine, tools, Device, Message, bindings
 from xaal.schemas import devices as schemas
+
+from .core import EntityFactory, XAALEntity, MonitorDevice
 from xaal.monitor import Monitor, Notification
 
+
 import logging
 _LOGGER = logging.getLogger(__name__)
 
@@ -15,7 +19,7 @@ DB_SERVER = tools.get_uuid('9064ccbc-84ea-11e8-80cc-82ed25e6aaaa')
 UNSUPPORTED_TYPES = ['cli','hmi','gateway','windgauge','barometer','soundmeter']
 
 
-def filter_msg(msg):
+def filter_msg(msg: Message) -> bool:
     m_type = msg.dev_type.split('.')[0]
     if m_type in UNSUPPORTED_TYPES:
         return False
@@ -37,36 +41,36 @@ class Bridge(object):
         self._factories = []
 
     @property
-    def engine(self):
+    def engine(self) -> AsyncEngine:
         return self._eng
 
-    async def on_start(self):
+    async def on_start(self) -> None:
         _LOGGER.info(f"{self._eng} started")
         #await self.wait_is_ready()
         print("Subscribing..")
         self._mon.subscribe(self.monitor_event)
         self._eng.subscribe(self.monitor_notification)
 
-    def on_stop(self):
+    def on_stop(self) -> None:
         _LOGGER.info(f"{self._eng} stopped")
 
-    def add_entity(self, addr, entity):
+    def add_entity(self, addr: bindings.UUID, entity: XAALEntity) -> None:
         _LOGGER.debug(f"new Entity {addr} {entity}")
         self._entities.update({addr: entity})
 
-    def remove_entity(self, addr):
+    def remove_entity(self, addr: bindings.UUID) -> None:
         self._entities.pop(addr)
 
-    def get_entity(self, addr):
+    def get_entity(self, addr: bindings.UUID) -> XAALEntity:
         return self._entities.get(addr, None)
 
-    def add_factory(self, klass):
+    def add_factory(self, klass: Type[EntityFactory]):
         self._factories.append(klass)
 
-    def remove_factory(self, klass):
+    def remove_factory(self, klass: Type[EntityFactory]):
         self._factories.remove(klass)
 
-    def setup_device(self):
+    def setup_device(self) -> Device:
         dev = schemas.hmi()
         dev.dev_type = 'hmi.hass'
         dev.vendor_id = 'IMT Atlantique'
@@ -76,7 +80,7 @@ class Bridge(object):
         self._eng.add_device(dev)
         return dev
 
-    def send_request(self, targets, action, body=None):
+    def send_request(self, targets: List[bindings.UUID], action: str, body: Dict[str, Any] | None =None):
         self._mon.engine.send_request(self._dev, targets, action, body)
 
     async def wait_is_ready(self) -> bool:
@@ -86,7 +90,7 @@ class Bridge(object):
             await asyncio.sleep(0.5)
         return False
 
-    def monitor_event(self, event, dev):
+    def monitor_event(self, notif: Notification, dev: MonitorDevice):
         # DB_SERVER keep spamming us
         if dev.address == DB_SERVER: return
 
@@ -95,8 +99,8 @@ class Bridge(object):
 
         # WARNING, never call ha_state update on a newly created entity, 
         # hass will block on this task (tricks: asyncio.sleep can fix that)
-        if entity and (event == Notification.attribute_change):
-            _LOGGER.debug(f"{event} {dev.address} => {entity}")
+        if entity and (notif == Notification.attribute_change):
+            _LOGGER.debug(f"{notif} {dev.address} => {entity}")
             entity.schedule_update_ha_state()
             return
 
@@ -111,7 +115,7 @@ class Bridge(object):
                 _LOGGER.warning(f"Unable to find entity for {dev.address} {dev.dev_type} ")
 
 
-    def monitor_notification(self, msg):
+    def monitor_notification(self, msg: Message):
         # right now the monitor doesn't send event on notification, so the bridge deals w/
         # both monitor events & messages.
         if (not msg.is_notify()) or msg.is_alive() or msg.is_attributes_change():

apps/homeassistant/xaal/core.py

 
+from typing import Any, Callable, Dict
 from .const import DOMAIN
 
-from homeassistant.helpers.entity import Entity
-
+from homeassistant.helpers.entity import Entity, DeviceInfo
+from xaal.lib import bindings
+from xaal.monitor.monitor import Device as MonitorDevice
 
+from .bridge import Bridge
 import logging
 _LOGGER = logging.getLogger(__name__)
 
@@ -11,12 +14,12 @@ _LOGGER = logging.getLogger(__name__)
 class XAALEntity(Entity):
     # _attr_has_entity_name = True
 
-    def __init__(self, dev, bridge):
+    def __init__(self, dev: MonitorDevice, bridge: Bridge) -> None:
         self._dev = dev
         self._bridge = bridge
 
     @property
-    def device_info(self):
+    def device_info(self) -> DeviceInfo | None :
         dev = self._dev
         ident = "dev:" + str(dev.address)
 
@@ -36,14 +39,14 @@ class XAALEntity(Entity):
             "manufacturer": dev.description.get("vendor_id", ""),
         }
 
-    def send_request(self, action, body=None):
+    def send_request(self, action: str, body: Dict[str, Any] | None =None) -> None:
         _LOGGER.debug(f"{self} {action} {body}")
         self._bridge.send_request([self._dev.address, ], action, body)
 
-    def get_attribute(self, name, default=None):
+    def get_attribute(self, name: str, default: Dict[str, Any] =None) -> Any:
         return self._dev.attributes.get(name, default)
 
-    def short_type(self):
+    def short_type(self) -> str:
         """ return a fake device class for entity that doesn't have one """
         # this apply for light, button
         # NOTE: I don't know why some entities don't have a device class
@@ -54,7 +57,7 @@ class XAALEntity(Entity):
         return True
 
     @property
-    def should_poll(self):
+    def should_poll(self) -> bool:
         """No polling needed."""
         return False
 
@@ -71,13 +74,13 @@ class XAALEntity(Entity):
 
 class EntityFactory(object):
 
-    def __init__(self, bridge, async_add_entitites):
+    def __init__(self, bridge: Bridge, async_add_entitites: Callable) -> None:
         self._bridge = bridge
         self._async_add_entitites = async_add_entitites
         self._bridge.add_factory(self)
 
 
-    def add_entity(self, entity, address):
+    def add_entity(self, entity: Entity, address: bindings.UUID) -> None:
         self._async_add_entitites([entity])
         self._bridge.add_entity(address, entity)
 

apps/homeassistant/xaal/cover.py

@@ -3,8 +3,7 @@ from typing import Any
 
 from homeassistant.components.cover import CoverEntity, CoverDeviceClass, CoverEntityFeature, ATTR_POSITION
 
-
-from .core import XAALEntity, EntityFactory, async_setup_factory
+from .core import XAALEntity, EntityFactory, MonitorDevice, async_setup_factory
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -15,7 +14,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
 
 class Factory(EntityFactory):
 
-    def new_entity(self, device):
+    def new_entity(self, device: MonitorDevice) -> bool:
         entity = None
         
         if device.dev_type == 'shutter.basic':
@@ -80,5 +79,5 @@ class ShutterPosition(Shutter):
 
     def set_cover_position(self, **kwargs: Any) -> None:
         position = kwargs.get(ATTR_POSITION, None)
-        if position:
+        if position != None:
             self.send_request('set_position',{'position':position})

apps/homeassistant/xaal/light.py

 import logging
 
-from homeassistant.components.light import ATTR_BRIGHTNESS, ATTR_HS_COLOR, ATTR_COLOR_TEMP, LightEntity
+from homeassistant.components.light import ATTR_BRIGHTNESS, ATTR_HS_COLOR, ATTR_COLOR_TEMP, LightEntity, ColorMode
 from homeassistant.util import color as color_util
 
-from .core import XAALEntity, EntityFactory, async_setup_factory
+from .core import XAALEntity, EntityFactory, MonitorDevice, async_setup_factory
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -14,7 +14,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
 
 class Factory(EntityFactory):
 
-    def new_entity(self, device):
+    def new_entity(self, device: MonitorDevice) -> bool:
         if device.dev_type.startswith('lamp.'):
             entity = Lamp(device, self._bridge)
             self.add_entity(entity,device.address)
@@ -33,7 +33,7 @@ class Lamp(XAALEntity, LightEntity):
             return {"brightness"}
 
     @property
-    def color_mode(self):
+    def color_mode(self) -> ColorMode | str | None:
         mode = self.get_attribute('mode')
         if mode == 'white':
             return 'color_temp'
@@ -43,12 +43,13 @@ class Lamp(XAALEntity, LightEntity):
         return 'brightness'
 
     @property
-    def brightness(self):
+    def brightness(self) -> int | None:
         brightness = self.get_attribute('brightness',0)
         return round(255 * (int(brightness) / 100))
 
+
     @property
-    def hs_color(self):
+    def hs_color(self) -> tuple[float, float] | None:
         hsv = self.get_attribute('hsv')
         if hsv:
             return (hsv[0], hsv[1]*100)

apps/homeassistant/xaal/sensor.py

 import logging
+from typing import Any
 
 from homeassistant.components.sensor import SensorDeviceClass, SensorEntity
-
 from homeassistant import const
 
-from .core import XAALEntity, EntityFactory, async_setup_factory
-
+from .core import XAALEntity, EntityFactory, MonitorDevice, async_setup_factory
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -16,7 +15,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
 
 class Factory(EntityFactory):
 
-    def new_entity(self, device):
+    def new_entity(self, device: MonitorDevice) -> bool:
         entity = None
         if device.dev_type.startswith('thermometer.'):
             entity = Thermometer(device, self._bridge)
@@ -48,7 +47,7 @@ class Factory(EntityFactory):
 class XAALSensorEntity(XAALEntity, SensorEntity):
 
     @property
-    def native_value(self):
+    def native_value(self) -> Any:
         target = getattr(self,'_xaal_attribute')
         return self.get_attribute(target)
 

apps/homeassistant/xaal/switch.py

@@ -2,7 +2,7 @@ import logging
 from homeassistant.components.switch import SwitchEntity, DEVICE_CLASS_OUTLET
 
 
-from .core import EntityFactory, XAALEntity, async_setup_factory
+from .core import EntityFactory, XAALEntity, MonitorDevice, async_setup_factory
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -13,7 +13,7 @@ async def async_setup_entry(hass, config_entry, async_add_entities):
 
 class Factory(EntityFactory):
 
-    def new_entity(self, device):
+    def new_entity(self, device: MonitorDevice) -> bool:
         if device.dev_type.startswith('powerrelay.'):
             entity = PowerRelay(device, self._bridge)
             self.add_entity(entity,device.address)