Skip to content
Snippets Groups Projects
Commit c9d44192 authored by jkerdreu's avatar jkerdreu
Browse files

Use c_type instead of description.. 



git-svn-id: https://redmine.imt-atlantique.fr/svn/xaal/code/Python/branches/0.7@2587 b32b6428-25c9-4566-ad07-03861ab6144f
parent 0fa9e298
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
devices/protocols/HomeKit/xaal/homekit/accessories.py
+ 60
− 45
View file @ c9d44192
......@@ -2,7 +2,7 @@ from xaal.schemas import devices
from xaal.lib import tools,Device
from xaal.schemas import devices
from aiohomekit.model import ServicesTypes
from aiohomekit.model import ServicesTypes,CharacteristicsTypes
import json
import logging
......@@ -12,6 +12,7 @@ logger = logging.getLogger(__name__)
def find_service_class(name):
#return Service
if name in ['accessory-information']:
return Information
if name in ['light']:
......@@ -21,7 +22,7 @@ def find_service_class(name):
if name in ['occupancy']:
# right now occupancy seems to he used for switch without state
# w/ no state, i'm unable to monitor the change state.
return None
return Service
if name in ['motion']:
return Motion
if name in ['contact']:
......@@ -32,7 +33,7 @@ def find_service_class(name):
return Temperature
if name in ['battery']:
return Battery
return None
return Service
class Accessory(object):
......@@ -56,15 +57,14 @@ class Accessory(object):
s_iid = k['iid']
s_name = ServicesTypes.get_short(s_type)
klass = find_service_class(s_name)
if klass:
logger.debug(f"Service: {s_iid} {s_name}")
serv = klass(k,self)
if isinstance(serv,Information):
self.information = serv
for k in serv.subscribed:
self.subscribed.update({(self.aid,k):serv})
else:
logger.debug(f"Service: {s_iid} {s_name}")
serv = klass(k,self)
if serv.__class__ == Service:
logger.warn(f"Unknow service: {s_name}")
if isinstance(serv,Information):
self.information = serv
for k in serv.subscribed:
self.subscribed.update({(self.aid,k):serv})
def add_device(self,dev):
dev.address = self.last_addr
......@@ -85,39 +85,46 @@ class Service(object):
self.accessory = accessory
self.characteristics = service.get('characteristics',[])
self.subscribed = []
self.dump()
self.setup()
def get_data(self,characteristic):
c_iid = characteristic.get('iid',None)
c_type = characteristic.get('type',None)
c_type = CharacteristicsTypes.get_short(c_type)
desc = characteristic.get('description',None)
value = characteristic.get('value',None)
unit = characteristic.get('unit',None)
return (c_iid,desc,value,unit)
return (c_iid,c_type,desc,value,unit)
def search(self,description):
def search(self,c_type):
#import pdb;pdb.set_trace()
for obj in self.characteristics:
desc = obj.get('description',None)
if description == desc:
(c_iid,desc_,value,unit) = self.get_data(obj)
return (c_iid,value,unit)
return None
def search_value(self,description):
(c_iid,value,unit) = self.search(description)
cur_c_type = obj.get("type",None)
cur_c_type = CharacteristicsTypes.get_short(cur_c_type)
if c_type == cur_c_type:
return self.get_data(obj)
return (None,None,None,None,None)
def search_value(self,c_type):
(c_iid,c_type,desc,value,unit) = self.search(c_type)
return value
def monitor(self,desciption):
(c_iid,value,unit) = self.search(desciption)
def monitor(self,c_type):
(c_iid,c_type,desc,value,unit) = self.search(c_type)
if c_iid:
self.subscribe(c_iid)
return (c_iid,value,unit)
return (c_iid,c_type,desc,value,unit)
def dump(self):
for obj in self.characteristics:
(c_iid,desc,value,unit) = self.get_data(obj)
print(f" =>{c_iid} {desc} : {value} {unit}")
(c_iid,c_type,desc,value,unit) = self.get_data(obj)
if unit:
print(f" =>{c_iid} {c_type} {desc} : {value} {unit}")
else:
print(f" =>{c_iid} {c_type} {desc} : {value}")
def setup(self):
self.dump()
......@@ -142,17 +149,18 @@ class Service(object):
class Information(Service):
def setup(self):
#import pdb;pdb.set_trace()
self.vendor_id = 'HomeKit:' + self.search_value('Manufacturer')
self.product_id = self.search_value('Model')
self.version = self.search_value('FirmwareRevision')
self.info = self.search_value('Name')
self.hw_id = self.search_value('SerialNumber')
self.vendor_id = 'HomeKit:' + self.search_value('manufacturer')
self.product_id = self.search_value('model')
self.version = self.search_value('firmware.revision')
self.info = self.search_value('name')
self.hw_id = self.search_value('serial-number')
class Light(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('CurrentAmbientLightLevel')
(c_iid,c_type,desc,value,unit) = self.monitor('light-level.current')
self.dev = self.add_device(devices.luxmeter())
self.dev.attributes['illuminance'] = int(value)
if value!=None:
self.dev.attributes['illuminance'] = int(value)
def handler(self,src,data):
self.dev.attributes['illuminance'] = int(data['value'])
......@@ -160,9 +168,10 @@ class Light(Service):
class Lamp(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('On')
(c_iid,c_type,desc,value,unit) = self.monitor('On')
self.dev = self.add_device(devices.lamp())
self.dev.attributes['light'] = bool(value)
if value!=None:
self.dev.attributes['light'] = bool(value)
def handler(self,src,data):
self.dev.attributes['light'] = bool(data['value'])
......@@ -170,9 +179,10 @@ class Lamp(Service):
class Motion(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('MotionDetected')
(c_iid,c_type,desc,value,unit) = self.monitor('motion-detected')
self.dev = self.add_device(devices.motion())
self.dev.attributes['presence'] = bool(value)
if value!=None:
self.dev.attributes['presence'] = bool(value)
def handler(self,src,data):
self.dev.attributes['presence'] = bool(data['value'])
......@@ -180,27 +190,31 @@ class Motion(Service):
class Contact(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('ContactSensorState')
(c_iid,c_type,desc,value,unit) = self.monitor('contact-state')
self.dev = self.add_device(devices.contact())
self.dev.attributes['detected'] = bool(value)
if value!=None:
self.dev.attributes['detected'] = bool(value)
def handler(self,src,data):
self.dev.attributes['detected'] = bool(data['value'])
class Humidity(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('CurrentRelativeHumidity')
#import pdb;pdb.set_trace()
(c_iid,c_type,desc,value,unit) = self.monitor('relative-humidity.current')
self.dev = self.add_device(devices.hygrometer())
self.dev.attributes['humidity'] = round(value,2)
if value!=None:
self.dev.attributes['humidity'] = round(value,2)
def handler(self,src,data):
self.dev.attributes['humidity'] = round(data['value'],2)
class Temperature(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('CurrentTemperature')
(c_iid,c_type,desc,value,unit) = self.monitor('temperature.current')
self.dev = self.add_device(devices.thermometer())
self.dev.attributes['temperature'] = round(value,2)
if value!=None:
self.dev.attributes['temperature'] = round(value,2)
def handler(self,src,data):
self.dev.attributes['temperature'] = round(data['value'],2)
......@@ -208,9 +222,10 @@ class Temperature(Service):
class Battery(Service):
def setup(self):
(c_iid,value,unit) = self.monitor('BatteryLevel')
(c_iid,c_type,desc,value,unit) = self.monitor('battery-level')
self.dev = self.add_device(devices.battery())
self.dev.attributes['level'] = int(value)
if value!=None:
self.dev.attributes['level'] = int(value)
def handler(self,src,data):
self.dev.attributes['level']=int(data['value'])
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment