diff --git a/.gitignore b/.gitignore index 6edbb74..2318402 100644 --- a/.gitignore +++ b/.gitignore @@ -105,3 +105,7 @@ tox.ini HowToUploadToPyPi.txt MANIFEST.in .pytest_cache/v/cache/lastfailed + +# vscode +launch.json +settings.json diff --git a/example/playwith.py b/example/playwith.py index 77994a9..61c43aa 100644 --- a/example/playwith.py +++ b/example/playwith.py @@ -1,16 +1,22 @@ +"""This is the test to play around.""" + import requests import xml.etree.ElementTree as ET import time current_position = None + def get_state(): + """Get the state.""" full_url = "http://192.168.86.114/zrap/chscan/ch1" device_info_response = requests.get(full_url) root = ET.fromstring(device_info_response.text) return str(root[0][0].text) + def count_time_till_stop(): + """Count the time till stop.""" start = time.time() time.sleep(float(0.5)) while(get_state() != '0'): @@ -18,10 +24,12 @@ def count_time_till_stop(): end = time.time() print(end - start) + def go_to_position(postion): + """Go to position.""" sleep_time = None global current_position - if current_position is None: + if current_position is None: payload = "cmd=close" full_url = "http://192.168.86.114/zrap/chctrl/ch1" requests.post(full_url, data=payload) @@ -42,19 +50,19 @@ def go_to_position(postion): full_url = "http://192.168.86.114/zrap/chctrl/ch1" requests.post(full_url, data=payload) sleep_time = float(55.3/100*(postion-current_position)) - time.sleep(sleep_time) payload = "cmd=stop" full_url = "http://192.168.86.114/zrap/chctrl/ch1" requests.post(full_url, data=payload) current_position = postion + device_info_response = requests.get("http://192.168.86.114/zrap/chdes") -#payload = "cmd=open" -#full_url = "http://192.168.86.114/zrap/chctrl/ch1" -#device_info_response = requests.post(full_url, data=payload) -#print(device_info_response) +# payload = "cmd=open" +# full_url = "http://192.168.86.114/zrap/chctrl/ch1" +# device_info_response = requests.post(full_url, data=payload) +# print(device_info_response) go_to_position(50) time.sleep(float(2)) @@ -67,4 +75,3 @@ def go_to_position(postion): go_to_position(10) time.sleep(float(2)) go_to_position(0) - diff --git a/example/sample.py b/example/sample.py index 37bf60b..b37ff20 100644 --- a/example/sample.py +++ b/example/sample.py @@ -1,16 +1,29 @@ +"""This is a Sample how to use the zeptrion Air Api.""" + from zeptrionAirApi import ZeptrionAirHub -hub = ZeptrionAirHub(3) #Initialize the ZeptrionAIrHub and browse for 5sec -allZeptrionPanels = hub.get_all_panels() #returns all found panels -allZeptrionChannels = hub.get_all_channels() #returns all found channels -allLightChannels = hub.get_all_light_channels() #returns all found channels that are Light switches -allGroups = hub.get_all_group() #returns all groups, that have been defined on the zeptrion app +# Initialize the ZeptrionAIrHub and browse for 5sec +hub = ZeptrionAirHub(3) +# returns all found panels +allZeptrionPanels = hub.get_all_panels() +# returns all found channels +allZeptrionChannels = hub.get_all_channels() +# returns all found channels that are Light switches +allLightChannels = hub.get_all_light_channels() +# returns all groups, that have been defined on the zeptrion app +allGroups = hub.get_all_group() for group in allGroups: output_to_return = "Channels in Group: " + str(group) - hub.get_all_channels_by_group(group) #returns all channles that are defined for that group + # returns all channles that are defined for that group + hub.get_all_channels_by_group(group) -channels = hub.get_all_channels_by_cat(1) #returns all found channels that are Light switches --> Category 1 -channel = channels[0] #pick the first light -channel.toggle_light() #toggle the light -channel.turn_on_light() #turn the light on -channel.turn_off_light() #turn the light off \ No newline at end of file +# returns all found channels that are Light switches --> Category 1 +channels = hub.get_all_channels_by_cat(1) +# pick the first light +channel = channels[0] +# toggle the light +channel.toggle_light() +# turn the light on +channel.turn_on_light() +# turn the light off +channel.turn_off_light() diff --git a/sample_smart_button.py b/sample_smart_button.py new file mode 100644 index 0000000..9441982 --- /dev/null +++ b/sample_smart_button.py @@ -0,0 +1,28 @@ +"""This is the test for Smart Button.""" + +from zeptrionAirApi import ZeptrionAirHub +import json + +hub = ZeptrionAirHub(3) # Initialize the ZeptrionAIrHub and browse for 5sec +allZeptrionPanels = hub.get_all_panels() # returns all found panels +# smartBtns = hub.get_all_smart_buttons() # returns all found smart buttons +# hub.set_all_smart_buttons_to_program_mode() + +prog_elements = {} +prog_elements['prog_req_type'] = 'POST' +prog_elements['prog_url'] = '192.168.86.167' +prog_elements['prog_path'] = '/test' +prog_elements['prog_typ'] = 'application/json' +prog_elements['prog_header_field'] = '' +prog_elements['prog_port'] = '1880' +body = {} +body['state'] = 20 +body['attributes'] = {} +body['attributes']['unit_of_measurement'] = '°C' +body['attributes']['friendly_name'] = 'Bathroom Temperature' +prog_elements['prog_body'] = json.dumps(body) + +# result = hub.programm_smart_btn(prog_elements) +# print(result) +result = hub.get_panels_with_smart_button() +print(result) diff --git a/setup.py b/setup.py index 30c7dc3..af069ad 100644 --- a/setup.py +++ b/setup.py @@ -1,4 +1,6 @@ -"""A setuptools based setup module. +""" +A setuptools based setup module. + See: https://packaging.python.org/en/latest/distributing.html https://github.com/pypa/sampleproject @@ -17,17 +19,17 @@ long_description = f.read() setup( - name = 'zeptrionAirApi', + name='zeptrionAirApi', packages=find_packages(exclude=['contrib', 'docs', 'tests']), - version = '1.0.4', - description = "This are the Classes to use the Zeptrion Air Lights etc.", + version='1.0.5', + description="This are the Classes to use the Zeptrion Air Lights etc.", long_description=long_description, - author = "Swissglider", - author_email = "swissglider@mailschweiz.com", - url = "https://github.com/swissglider/zeptrionAirApi", - keywords = "zeptrion zeptrion-air Light-Switch Switch Blind-Switch", + author="Swissglider", + author_email="swissglider@mailschweiz.com", + url="https://github.com/swissglider/zeptrionAirApi", + keywords="zeptrion zeptrion-air Light-Switch Switch Blind-Switch", install_requires=['requests', 'zeroconf', ], - classifiers = [ + classifiers=[ "Development Status :: 5 - Production/Stable", "Intended Audience :: Developers", "Topic :: Software Development :: Libraries :: Python Modules", @@ -35,5 +37,5 @@ "Programming Language :: Python :: 3", ], setup_requires=['pytest-runner'], - tests_require=['pytest'], -) \ No newline at end of file + tests_require=['pytest', 'pytest-asyncio'], +) diff --git a/test_script.sh b/test_script.sh old mode 100644 new mode 100755 index ee4203a..2a2ff1d --- a/test_script.sh +++ b/test_script.sh @@ -13,10 +13,12 @@ echo " ### pycodestyle --first test - "+$1 pycodestyle --first $1 echo " ### pylint 3.6 test - "+$1 -/Users/diener/Library/Python/3.6/bin/pylint $1 +/Users/diener/Library/Python/3.6/bin/pylint ./zeptrionAirApi -#echo " ### python3 setup.py test" -#python3 setup.py test +# echo " ### python3 setup.py test" +# python3 setup.py test echo " ### pytest and coverage" -py.test --cov ./zeptrionAirApi \ No newline at end of file +py.test --cov ./zeptrionAirApi + +# py.test -s --cov-report html --cov ./zeptrionAirApi \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py index 1b147b9..41d22b1 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1,4 +1,6 @@ +"""This is the __init_.py file.""" + # __init__.py -#from zeptrionAirApi.zeptrion_air_hub import ZeptrionAirHub -#from zeptrionAirApi.zeptrion_air_panel import zeptrion_air_panel -#from zeptrionAirApi.zeptrion_air_channel import zeptrion_air_channel +# from zeptrionAirApi.zeptrion_air_hub import ZeptrionAirHub +# from zeptrionAirApi.zeptrion_air_panel import zeptrion_air_panel +# from zeptrionAirApi.zeptrion_air_channel import zeptrion_air_channel diff --git a/tests/test_zeptrionAirApi.py b/tests/test_zeptrionAirApi.py index 7825da0..64ebc13 100644 --- a/tests/test_zeptrionAirApi.py +++ b/tests/test_zeptrionAirApi.py @@ -1,15 +1,35 @@ import unittest import time +import json from zeptrionAirApi import ZeptrionAirHub + class TestZeptrionAir(unittest.TestCase): def setup_class(self): self.hub = ZeptrionAirHub(3) - + +# ============================================ +# Hub tests +# ============================================ + def test_if_Hub_is_not_None(self): ''' Test if hub is initialized. ''' self.assertTrue(self.hub is not None) + def test_if_service_type_is_correct(self): + ''' Test if service type is correct. ''' + service_type = self.hub.service_type + self.assertTrue(service_type == '_zapp._tcp.local.') + + def test_get_all_smart_buttons(self): + ''' Test if service smart buttons are found. ''' + panel_with_smart_buttons = self.hub.get_panels_with_smart_button() + self.assertTrue(panel_with_smart_buttons) + +# ============================================ +# Panel tests +# ============================================ + def test_if_panels_are_found(self): ''' Test if panels are found. ''' panels = self.hub.all_found_panels @@ -20,42 +40,11 @@ def test_if_new_panels_found(self): latest_panels = self.hub.latest_found_panels self.assertTrue(latest_panels) - def test_if_service_type_is_correct(self): - ''' Test if service type is correct. ''' - service_type = self.hub.service_type - self.assertTrue(service_type == '_zapp._tcp.local.') - def test_get_all_panels(self): ''' Test if get_all_pannels returns some panels. ''' panels = self.hub.get_all_panels() self.assertTrue(panels) - def test_get_all_channels(self): - ''' Test if get_all_channels returns some channesl. ''' - channels = self.hub.get_all_channels() - self.assertTrue(channels) - - def test_get_all_light_channels(self): - ''' Test if get_all_light_channels returns some lights. ''' - channels = self.hub.get_all_light_channels() - self.assertTrue(channels) - - def test_get_all_channels_by_cat5(self): - ''' Test if get_all_channels_by_cat returns some channels with cat 5. ''' - channels = self.hub.get_all_channels_by_cat(5) - self.assertTrue(channels) - - def test_get_all_channels_by_cat35(self): - ''' Test if get_all_channels_by_cat returns some channels with cat 35. ''' - channels = self.hub.get_all_channels_by_cat(35) - self.assertTrue(not channels) - - def test_get_all_channels_by_group(self): - ''' Test if get_all_channels_by_group not None. ''' - groups = self.hub.get_all_group() - channels = self.hub.get_all_channels_by_group(groups[1]) - self.assertTrue(channels) - def test_panel_name(self): ''' Test if panel_name not None. ''' panels = self.hub.get_all_panels() @@ -80,6 +69,40 @@ def test_panel_to_string(self): for panel in panels: self.assertTrue(str(panel)) +# ============================================ +# Channel tests +# ============================================ + + def test_get_all_channels(self): + ''' Test if get_all_channels returns some channesl. ''' + channels = self.hub.get_all_channels() + self.assertTrue(channels) + + def test_get_all_light_channels(self): + ''' Test if get_all_light_channels returns some lights. ''' + channels = self.hub.get_all_light_channels() + self.assertTrue(channels) + + def test_get_all_channels_by_cat5(self): + ''' + Test if get_all_channels_by_cat returns some channels with cat 5. + ''' + channels = self.hub.get_all_channels_by_cat(5) + self.assertTrue(channels) + + def test_get_all_channels_by_cat35(self): + ''' + Test if get_all_channels_by_cat returns some channels with cat 35. + ''' + channels = self.hub.get_all_channels_by_cat(35) + self.assertTrue(not channels) + + def test_get_all_channels_by_group(self): + ''' Test if get_all_channels_by_group not None. ''' + groups = self.hub.get_all_group() + channels = self.hub.get_all_channels_by_group(groups[1]) + self.assertTrue(channels) + def test_channel_to_string(self): ''' Test if to String not None. ''' channels = self.hub.get_all_channels() @@ -90,13 +113,13 @@ def test_channel_panel_name(self): ''' Test if panel_name returns not None. ''' channels = self.hub.get_all_channels() channel = channels[0] - self.assertTrue(channel.panel_name) + self.assertTrue(channel.panel.panel_name) def test_channel_panel_type(self): ''' Test if panel_type returns not None. ''' channels = self.hub.get_all_channels() channel = channels[0] - channel.panel_type + channel.panel.panel_type self.assertTrue(True) def test_channel_uniq_id(self): @@ -110,19 +133,19 @@ def test_channel_panel_ip(self): ''' Test if panel_ip returns not None. ''' channels = self.hub.get_all_channels() channel = channels[0] - self.assertTrue(channel.panel_ip) + self.assertTrue(channel.panel.panel_ip) def test_channel_panel_port(self): ''' Test if panel_port returns not None. ''' channels = self.hub.get_all_channels() channel = channels[0] - self.assertTrue(channel.panel_port) + self.assertTrue(channel.panel.panel_port) def test_channel_panel_url(self): ''' Test if panel_url returns not None. ''' channels = self.hub.get_all_channels() channel = channels[0] - self.assertTrue(channel.panel_url) + self.assertTrue(channel.panel.panel_url) def test_channel_turn_on_light(self): ''' Test if turn_on_light returns not None. ''' @@ -152,7 +175,7 @@ def test_control_light_with_cat5(self): channel = channels[0] channel.light_controller.toggle_light() self.assertTrue(True) - + def test_blind_controll_down_up(self): ''' Test blind controller move down/tilt/up. ''' channels = self.hub.get_all_channels_by_cat(5) @@ -178,3 +201,20 @@ def test_blind_controll_down_up(self): channel.blind_controller.move_up_blind() self.assertTrue(True) + def test_programm_smart_buttons(self): + prog_elements = {} + prog_elements['prog_req_type'] = 'POST' + prog_elements['prog_url'] = '192.168.86.167' + prog_elements['prog_path'] = '/test' + prog_elements['prog_typ'] = 'application/json' + prog_elements['prog_header_field'] = '' + prog_elements['prog_port'] = '1880' + body = {} + body['state'] = 20 + body['attributes'] = {} + body['attributes']['unit_of_measurement'] = '°C' + body['attributes']['friendly_name'] = 'Bathroom Temperature' + prog_elements['prog_body'] = json.dumps(body) + + result = self.hub.programm_smart_btn(prog_elements) + self.assertTrue(result) diff --git a/zeptrionAirApi/__init__.py b/zeptrionAirApi/__init__.py index ebcd49b..ded0108 100644 --- a/zeptrionAirApi/__init__.py +++ b/zeptrionAirApi/__init__.py @@ -1,2 +1,4 @@ +"""This is the __init_.py file.""" + # __init__.py -from zeptrionAirApi.zeptrion_air_hub import ZeptrionAirHub +from zeptrionAirApi.zeptrion_air_hub import ZeptrionAirHub # noqa: F401 diff --git a/zeptrionAirApi/zeptrion_air_channel.py b/zeptrionAirApi/zeptrion_air_channel.py index fa336f2..97badcd 100644 --- a/zeptrionAirApi/zeptrion_air_channel.py +++ b/zeptrionAirApi/zeptrion_air_channel.py @@ -5,8 +5,13 @@ https://github.com/swissglider/zeptrionAirApi """ -from .zeptrion_air_channel_blind_controller import ZeptrionAirChannelBlindController # noqa: E501 -from .zeptrion_air_channel_light_controller import ZeptrionAirChannelLightController # noqa: E501 +from .zeptrion_air_channel_blind_controller \ + import ZeptrionAirChannelBlindController +from .zeptrion_air_channel_light_controller \ + import ZeptrionAirChannelLightController +from .zeptrion_air_channel_update_helper \ + import ZeptrionAirChannelUpdatHelper \ + as Helper class ZeptrionAirChannel: @@ -33,14 +38,17 @@ def __init__( """ self._channel_info = channel_info self._panel = panel - self._blind_controller = ZeptrionAirChannelBlindController( + helper = Helper( self._channel_info['channel_cat'], self._channel_info['channel_id'], - self._panel.panel_url) + self._panel.panel_url + ) + self._blind_controller = ZeptrionAirChannelBlindController( + helper + ) self._light_controller = ZeptrionAirChannelLightController( - self._channel_info['channel_cat'], - self._channel_info['channel_id'], - self._panel.panel_url) + helper + ) ''' Cat: -1: Not configured @@ -51,7 +59,7 @@ def __init__( @property def channel_uniq_id(self): """Return the uniq ID from the channel.""" - return str(self.panel_name)+str(self.channel_id) + return str(self._panel.panel_name)+str(self.channel_id) @property def channel_id(self): @@ -84,29 +92,9 @@ def channel_cat(self): return self._channel_info['channel_cat'] @property - def panel_name(self): - """Return the Panel-Name from the channel.""" - return self._panel.panel_name - - @property - def panel_type(self): - """Return the Panel-Type from the channel.""" - return self._panel.panel_type - - @property - def panel_ip(self): - """Return the IP from the channel/panel.""" - return self._panel.panel_ip - - @property - def panel_port(self): - """Return the Port from the channel/panel.""" - return self._panel.panel_port - - @property - def panel_url(self): - """Return the URL from the channel/panel.""" - return self._panel.panel_url + def panel(self): + """Return the Panel from the button.""" + return self._panel @property def channel_blind_state(self): @@ -136,6 +124,6 @@ def __repr__(self): return_str += "\tIcon: " + str(self.channel_icon) + '\n' return_str += "\tType: " + str(self.channel_type) + '\n' return_str += "\tCat: " + str(self.channel_cat) + '\n' - return_str += "\tIP: " + str(self.panel_ip) + '\n' + return_str += "\tIP: " + str(self._panel.panel_ip) + '\n' return_str += '\n' return return_str diff --git a/zeptrionAirApi/zeptrion_air_channel_blind_controller.py b/zeptrionAirApi/zeptrion_air_channel_blind_controller.py index e1a53a6..bcf03e0 100644 --- a/zeptrionAirApi/zeptrion_air_channel_blind_controller.py +++ b/zeptrionAirApi/zeptrion_air_channel_blind_controller.py @@ -7,7 +7,6 @@ import time import requests -import xml.etree.ElementTree as ET class ZeptrionAirChannelBlindController: @@ -18,12 +17,9 @@ class ZeptrionAirChannelBlindController: https://github.com/swissglider/zeptrionAirApi """ - def __init__( - self, channel_cat, channel_id, panel_url): + def __init__(self, helper): """Init the ZeptrionAirChannelBlindController.""" - self._panel_url = panel_url - self._channel_cat = channel_cat - self._channel_id = channel_id + self.helper = helper self._blind_position = None self._state = None @@ -92,21 +88,12 @@ def go_to_position(self, postion): self._blind_position = postion def _control_blind(self, payload): - if self._channel_cat == '5': - import requests - full_url = self._panel_url - full_url += "/zrap/chctrl/" + self._channel_id + if self.helper.channel_cat == '5': + full_url = self.helper.panel_url + full_url += "/zrap/chctrl/" + self.helper.channel_id requests.post(full_url, data=payload) self._state = self.update() def update(self): """Update the real status of the channel switch.""" - full_url = self._panel_url - full_url += "/zrap/chscan/" + self._channel_id - device_info_response = requests.get(full_url) - if device_info_response.status_code == 200: - root = ET.fromstring(device_info_response.text) - if root[0][0].text == '100': - return True - return False - return False # pragma: no cover + return self.helper.update() diff --git a/zeptrionAirApi/zeptrion_air_channel_light_controller.py b/zeptrionAirApi/zeptrion_air_channel_light_controller.py index 63c317c..0ceb33a 100644 --- a/zeptrionAirApi/zeptrion_air_channel_light_controller.py +++ b/zeptrionAirApi/zeptrion_air_channel_light_controller.py @@ -5,8 +5,6 @@ https://github.com/swissglider/zeptrionAirApi """ -import time - class ZeptrionAirChannelLightController: """ @@ -16,27 +14,10 @@ class ZeptrionAirChannelLightController: https://github.com/swissglider/zeptrionAirApi """ - def __init__( - self, channel_cat, channel_id, panel_url): - """Init the ZeptrionAirChannelLightController. """ - self._panel_url = panel_url - self._channel_cat = channel_cat - self._channel_id = channel_id - self._blind_position = None - - def update(self): - """Update the real status of the channel switch.""" - import requests - import xml.etree.ElementTree as ET - full_url = self._panel_url - full_url += "/zrap/chscan/" + self._channel_id - device_info_response = requests.get(full_url) - if device_info_response.status_code == 200: - root = ET.fromstring(device_info_response.text) - if root[0][0].text == '100': - return True - return False - return False # pragma: no cover + def __init__(self, helper): + """Init the ZeptrionAirChannelLightController.""" + self.helper = helper + self._led_state = None def turn_on_light(self): """Turn the real light switch off.""" @@ -51,9 +32,13 @@ def toggle_light(self): self._control_light("cmd=toggle") def _control_light(self, payload): - if self._channel_cat == '1': + if self.helper.channel_cat == '1': import requests - full_url = self._panel_url - full_url += "/zrap/chctrl/" + self._channel_id + full_url = self.helper.panel_url + full_url += "/zrap/chctrl/" + self.helper.channel_id requests.post(full_url, data=payload) - self._state = self.update() \ No newline at end of file + self._led_state = self.update() + + def update(self): + """Update the real status of the channel switch.""" + return self.helper.update() diff --git a/zeptrionAirApi/zeptrion_air_channel_update_helper.py b/zeptrionAirApi/zeptrion_air_channel_update_helper.py new file mode 100644 index 0000000..ceeab9a --- /dev/null +++ b/zeptrionAirApi/zeptrion_air_channel_update_helper.py @@ -0,0 +1,52 @@ +""" +Support for ZeptrionAirChannel. + +For more details about this Class, please refer to the documentation at +https://github.com/swissglider/zeptrionAirApi +""" + +import xml.etree.ElementTree as ET +import requests + + +class ZeptrionAirChannelUpdatHelper: + """ + Support for ZeptrionAirChannel. + + For more details about this Class, please refer to the documentation at + https://github.com/swissglider/zeptrionAirApi + """ + + def __init__( + self, channel_cat, channel_id, panel_url): + """Init the ZeptrionAirChannelBlindController.""" + self._channel_cat = channel_cat + self._panel_url = panel_url + self._channel_id = channel_id + + @property + def panel_url(self): + """Return panel url.""" + return self._panel_url + + @property + def channel_cat(self): + """Return channel categorie.""" + return self._channel_cat + + @property + def channel_id(self): + """Return channel uidrl.""" + return self._channel_id + + def update(self): + """Update the real status of the channel switch.""" + full_url = self.panel_url + full_url += "/zrap/chscan/" + self.channel_cat + device_info_response = requests.get(full_url) + if device_info_response.status_code == 200: + root = ET.fromstring(device_info_response.text) + if root[0][0].text == '100': + return True + return False + return False # pragma: no cover diff --git a/zeptrionAirApi/zeptrion_air_hub.py b/zeptrionAirApi/zeptrion_air_hub.py index 822bc05..c9d8a0d 100644 --- a/zeptrionAirApi/zeptrion_air_hub.py +++ b/zeptrionAirApi/zeptrion_air_hub.py @@ -129,3 +129,43 @@ def get_all_group(self): for temp_channel in temp_channels: groups.append(temp_channel.channel_group) return list(set(groups)) + + def get_panels_with_smart_button(self): + """Return all Panels that have smart buttons.""" + panel_with_smt_btn = [] + for panel in self.get_all_panels(): + if panel.has_smart_buttons: + panel_with_smt_btn.append(panel) + return panel_with_smt_btn + + def programm_smart_btn(self, prog_elements): # noqa: D301 + """ + Set the pressed Smart Buttons with the given program. + + :param prog_elements: -- dictornary with the following parameter + :prog_elements['prog_req_type'] + --> Type for the Webservice to be called + --> ex. POST/GET/PUT/DELETE + :prog_elements['prog_url'] + --> Url of the Webservice to be called + :prog_elements['prog_path'] + --> Path + :prog_elements['prog_typ'] + --> Type of the Webservice to be called + --> ex. application/json + :prog_elements['prog_header_field'] + --> individual HTTP header field + --> ex. 'SOAPACTION:http://test/foo#MyMessage\r\n' + :prog_elements['prog_port'] + --> Port of the Webservice to be called + :prog_elements['prog_body'] + --> Body of the Webservice to be called + """ + import asyncio + program_mode = [ + panel.programm_btn(prog_elements) + for panel in self.get_panels_with_smart_button() + ] + loop = asyncio.get_event_loop() + result = loop.run_until_complete(asyncio.gather(*program_mode)) + return result diff --git a/zeptrionAirApi/zeptrion_air_panel.py b/zeptrionAirApi/zeptrion_air_panel.py index bd3f865..f419669 100644 --- a/zeptrionAirApi/zeptrion_air_panel.py +++ b/zeptrionAirApi/zeptrion_air_panel.py @@ -27,6 +27,8 @@ def __init__(self, panel_info, panel_name, panel_type): self._panel_url = url self._panel_channels = [] self.__read_panel_channels__() + self._has_smart_buttons = False + self.__read_smart_buttons__() @property def panel_name(self): @@ -58,6 +60,11 @@ def panel_channels(self): """Return the Channels from the panel as List.""" return self._panel_channels + @property + def has_smart_buttons(self): + """Return the Smart Buttons from the panel as List.""" + return self._has_smart_buttons + def __repr__(self): """Return a String representing the ZeptrionAirPanel.""" return_string = "Panel Name: " + self.panel_name + '\n' @@ -105,3 +112,99 @@ def get_device_rssi(self): root = ET.fromstring(device_info_response.text) return root[0].text return '' # pragma: no cover + + def __read_smart_buttons__(self): + """Return the Smart Buttons that panel has.""" + import requests + import json + device_info_response = requests.get( + self.panel_url + "/zapi/smartfront/id") + + if device_info_response.status_code == 200: + # button_info = json.loads(device_info_response.text) + + device_led_response = requests.get( + self.panel_url + "/zapi/smartfront/led") + if device_led_response.status_code == 200: + buttons = json.loads(device_led_response.text) + for button in buttons: + if(button): + self._has_smart_buttons = True + return + + async def programm_btn(self, prog_elements): # noqa: D301 + """ + Set the pressed Smart Buttons with the given program. + + :param prog_elements: -- dictornary with the following parameter + :prog_elements['prog_req_type'] + --> Type for the Webservice to be called + --> ex. POST/GET/PUT/DELETE + :prog_elements['prog_url'] + --> Url of the Webservice to be called + :prog_elements['prog_path'] + --> Path + :prog_elements['prog_typ'] + --> Type of the Webservice to be called + --> ex. application/json + :prog_elements['prog_header_field'] + --> individual HTTP header field + --> ex. "SOAPACTION:http://test/foo#MyMessage\r\n" + :prog_elements['prog_port'] + --> Port of the Webservice to be called + :prog_elements['prog_body'] + --> Body of the Webservice to be called + """ + import aiohttp + import requests + import json + url = self.panel_url + "/zapi/smartbt/prgm" + payload = {'on': True} + headers = {'content-type': 'application/json'} + requests.post(url, data=json.dumps(payload), headers=headers) + url = self.panel_url + "/zapi/smartbt/prgn" + prgn_response = requests.get(url) + if prgn_response.text and prgn_response.status_code == 200: + result = json.loads(prgn_response.text) + if result['prg']: + async with aiohttp.ClientSession() as session: + url = self.panel_url + "/zapi/smartbt/prgs" + payload = [{ + 'req': prog_elements['prog_req_type'], + 'loc': prog_elements['prog_url'], + 'pth': prog_elements['prog_path'], + 'typ': prog_elements['prog_typ'], + 'hdr': prog_elements['prog_header_field'], + 'prt': prog_elements['prog_port'], + 'bdy': prog_elements['prog_body'] + }] + headers = {'content-type': 'application/json'} + async with session.post( + url, + data=json.dumps(payload), + headers=headers + ) as response: + await response.text() + if response.status == 200: + return { + 'changed': True, + 'tried to change': False, + 'status_code': response.status, + 'text': response.text(), + 'response': response + } + return { + 'changed': False, + 'tried to change': True, + 'status_code': response.status, + 'text': response.text(), + 'response': response + } # pragma: no cover + else: + return { + 'changed': False, + 'tried to change': True, + 'status_code': prgn_response.status_code, + 'text': prgn_response.text, + 'response': prgn_response + } # pragma: no cover