Skip to content

Commit

Permalink
Merge branch 'staging' of https://github.com/bastoscorp/pvAlert into …
Browse files Browse the repository at this point in the history
…staging
  • Loading branch information
bastoscorp committed Jul 17, 2024
2 parents 6dbdbf8 + e1e7d87 commit eb2f9e8
Show file tree
Hide file tree
Showing 10 changed files with 987 additions and 44 deletions.
1 change: 1 addition & 0 deletions .github/workflows/python-app.yml
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ on:
pull_request:
branches: [ "main", "staging" ]


permissions:
contents: read

Expand Down
23 changes: 11 additions & 12 deletions business/actionDeviceManager.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,12 +2,12 @@
from business.actionDevicePhilipsHue import ActionDevicePhilipsHue
import logging
from random import randrange
class ActionDeviceManager:


class ActionDeviceManager:
config: Config = None
testing_status = {}


def __init__(self, conf: Config):
self.config = conf

Expand All @@ -21,18 +21,18 @@ def get_device_status(self, dev_type: str, dev_name: str):
if self.testing_status.get(dev_name) == None:
if randrange(10) % 2 == 0:
status = {'status': True,
'message': "powered on"
}
'message': "powered on"
}
else:
status = {'status': False,
'message': "powered off"
}
'message': "powered off"
}
self.testing_status[dev_name] = status
return self.testing_status.get(dev_name)
else:
logging.warning(dev_type + " device type not implemented")

def enable_device(self,dev_type: str, dev_name: str):
def enable_device(self, dev_type: str, dev_name: str):
status = None
if dev_type == "philips_hue":
dev_manager: ActionDevicePhilipsHue = ActionDevicePhilipsHue(self.config)
Expand All @@ -47,18 +47,17 @@ def enable_device(self,dev_type: str, dev_name: str):
else:
logging.warning(dev_type + " device type not implemented")


def disable_device(self,dev_type: str, dev_name: str):
def disable_device(self, dev_type: str, dev_name: str):
status = None
if dev_type == "philips_hue":
dev_manager: ActionDevicePhilipsHue = ActionDevicePhilipsHue(self.config)
status = dev_manager.disable_plug(dev_name)
return status
if dev_type == "testing":
status = {'status': False,
'message': "powered off"
}
'message': "powered off"
}
self.testing_status[dev_name] = status
return True
else:
logging.warning(dev_type + " device type not implemented")
logging.warning(dev_type + " device type not implemented")
72 changes: 40 additions & 32 deletions business/actionDevicePhilipsHue.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
from config.config import Config
from os.path import exists as file_exists

import logging
import requests
Expand All @@ -8,10 +9,8 @@
import json

import urllib3
urllib3.disable_warnings()


from os.path import exists as file_exists
urllib3.disable_warnings()


class ActionDevicePhilipsHue:
Expand Down Expand Up @@ -44,7 +43,7 @@ def load_cache_data(self):
return ret

def save_cache_data(self):
if self.bridge_ip == None:
if self.bridge_ip is None:
self.discover_ip()
cache_file = self.config.hue_cache_file
data = {'bridge_ip': self.bridge_ip
Expand All @@ -61,76 +60,83 @@ def discover_ip(self):
data = rep_data[0]
self.bridge_ip = data['internalipaddress']
else:
logging.error("issue to get Philips Hue Bridge ip address, got http error : " + response.status_code + " reason : " + response.reason)
except requests.exceptions as C:
logging.error(
"issue to get Philips Hue Bridge ip address, got http error : " + str(
response.status_code) + " reason : " + response.reason)
except requests.exceptions:
logging.error("issue to get Philips Hue Bridge ip address")

def get_device_id(self,device_name):
def get_device_id(self, device_name):
url_raw = self.config.hue_url_all_dev
url = url_raw.replace(self.bridge_ip_token, self.bridge_ip)
try:
headers = {"hue-application-key": self.config.hue_username}
response = requests.get(url, headers=headers, verify=False)
target_id = ""
if response.status_code == 200:
rep_data = response.json()
data = rep_data['data']
data_size= len(data)
data_size = len(data)
i = 0
target_id = ""
while i < data_size:
dev = data[i]
metadata = dev['metadata']
name = metadata['name']
if name == device_name:
target_id =str(dev['id'])
target_id = str(dev['id'])
i = data_size
i += 1
else:
logging.error("issue to reach Philips Hue Bridge, got http error : " + response.status_code + " reason : " + response.reason)

logging.error("issue to reach Philips Hue Bridge, got http error : " + str(
response.status_code) + " reason : " + response.reason)

if target_id != "":
return target_id
else:
logging.error("cannot find device named :" + device_name)
return None
except requests.exceptions as C:
except requests.exceptions:
logging.error("issue to reach Philips Hue bridge")
return None

def get_device_status(self, device_name):
hue_id = self.get_device_id(device_name)
status = None
if hue_id != None:
if hue_id is not None:
url_raw = self.config.hue_url_all_dev
url = url_raw.replace(self.bridge_ip_token, self.bridge_ip)
url = url + '/' + hue_id
headers = {"hue-application-key": self.config.hue_username}
try:
response = requests.get(url , headers=headers , verify=False)
response = requests.get(url, headers=headers, verify=False)
if response.status_code == 200:
rep_data = response.json()
data = rep_data['data']
status = data[0]["on"]["on"]
else:
logging.error("issue to reach Philips Hue Bridge, got http error : " + response.status_code + " reason : " + response.reason)
except requests.exceptions as C:
logging.error(
"issue to reach Philips Hue Bridge, got http error : " + str(
response.status_code) + " reason : " + response.reason)
except requests.exceptions:
logging.error("issue to reach Philips Hue bridge")
else:
logging.error("error to get this device id")
raise Exception("issue to get " + device_name + " id")
if status == True:
if status is True:
data_to_return = {'status': True,
'message': "powered on"
}
'message': "powered on"
}
return data_to_return
if status == False:
if status is False:
data_to_return = {'status': False,
'message': "powered off"
}
'message': "powered off"
}
return data_to_return
if status == None:
if status is None:
return None


def send_command(self,action,device_name):
def send_command(self, action, device_name):
data = None
ret = False
if action == "enable":
Expand All @@ -145,9 +151,9 @@ def send_command(self,action,device_name):
"on": False
}
}
if data != None:
if data is not None:
hue_id = self.get_device_id(device_name)
if hue_id != None:
if hue_id is not None:
url_raw = self.config.hue_url_all_dev
url = url_raw.replace(self.bridge_ip_token, self.bridge_ip)
url = url + '/' + hue_id
Expand All @@ -165,16 +171,18 @@ def send_command(self,action,device_name):
ret = True
else:
logging.error(
"issue to reach Philips Hue Bridge, got http error : " + response.status_code + " reason : " + response.reason)
"issue to reach Philips Hue Bridge, got http error : " + str(
response.status_code) + " reason : " + response.reason)

except requests.exceptions as C:
except requests.exceptions:
logging.error("issue to reach Philips Hue bridge")
else:
logging.error("error to get this device id")
raise Exception("issue to get " + device_name + " id")
return ret
def enable_plug(self,device_name):

def enable_plug(self, device_name):
return self.send_command("enable", device_name)

def disable_plug(self,device_name):
return self.send_command("disable", device_name)
def disable_plug(self, device_name):
return self.send_command("disable", device_name)
Binary file added data/sessionFile.txt
Binary file not shown.
45 changes: 45 additions & 0 deletions e2e_tests/test_actiondevicephilipshue.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import ipaddress
import os
from os.path import exists as file_exists

import sys
sys.path.insert(0, '../pvAlert')

from config.config import Config
from business.actionDevicePhilipsHue import ActionDevicePhilipsHue

# should have a configured philips hue tor testing

dirname = os.path.dirname(__file__)
#get parrent config file
home = os.path.dirname(dirname)
conffile = os.path.join(home, 'config.ini')
conf = Config(conffile)
adm_hue = ActionDevicePhilipsHue(conf)

def test_save_cache():
if adm_hue.bridge_ip != None:
adm_hue.save_cache_data()
file = conf.hue_cache_file
assert file_exists(file)
else:
assert False

def test_load_cache():
adm_hue.bridge_ip = None
adm_hue.load_cache_data()
# test if it is an v4 ip address
assert ipaddress.ip_address(adm_hue.bridge_ip)

def test_bridge_ip_discovery():
adm_hue.bridge_ip = None
adm_hue.discover_ip()
# test if it is an v4 ip address
assert ipaddress.ip_address(adm_hue.bridge_ip)


def test_get_device_id_KO():
hue_id = adm_hue.get_device_id("azerzrtzretzertezrt")
assert hue_id == None


Loading

0 comments on commit eb2f9e8

Please sign in to comment.