Extend shelly support. (#989)

BridgeEmulator/HueEmulator3.py

@@ -34,7 +34,6 @@
 # Tell users what view to go to when they need to login.
 login_manager.login_view = "core.login"
 
-
 @login_manager.user_loader
 def user_loader(email):
  if email not in bridgeConfig["config"]["users"]:
@@ -44,7 +43,6 @@ def user_loader(email):
  user.id = email
  return user
 
-
 @login_manager.request_loader
 def request_loader(request):
  email = request.form.get('email')

BridgeEmulator/HueObjects/__init__.py

@@ -15,7 +15,6 @@
 
 eventstream = []
 
-
 def v1StateToV2(v1State):
  v2State = {}
  if "on" in v1State:
@@ -48,11 +47,9 @@ def v2StateToV1(v2State):
  v1State["transitiontime"] = v2State["transitiontime"]
  return v1State
 
-
 def genV2Uuid():
  return str(uuid.uuid4())
 
-
 def generate_unique_id():
  rand_bytes = [random.randrange(0, 256) for _ in range(3)]
  return "00:17:88:01:00:%02x:%02x:%02x-0b" % (rand_bytes[0], rand_bytes[1], rand_bytes[2])
@@ -236,7 +233,6 @@ def save(self):
 
  return result
 
-
 class ApiUser():
  def __init__(self, username, name, client_key, create_date=datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S"), last_use_date=datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")):
  self.username = username
@@ -253,7 +249,6 @@ def save(self):
 
 
 class Light():
-
  def __init__(self, data):
  self.name = data["name"]
  self.modelid = data["modelid"]
@@ -271,6 +266,7 @@ def __init__(self, data):
  self.streaming = False
  self.dynamics = deepcopy(lightTypes[self.modelid]["dynamics"])
  self.effect = "no_effect"
+
  # entertainment
  streamMessage = {"creationtime": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [{"id": str(uuid.uuid5(
@@ -281,20 +277,23 @@ def __init__(self, data):
  streamMessage["id_v1"] = "/lights/" + self.id_v1
  streamMessage["data"][0].update(self.getV2Entertainment())
  eventstream.append(streamMessage)
+
  # zigbee_connectivity
  streamMessage = {"creationtime": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [self.getZigBee()],
  "id": str(uuid.uuid4()),
  "type": "add"
  }
  eventstream.append(streamMessage)
+
  # light
  streamMessage = {"creationtime": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [self.getV2Api()],
  "id": str(uuid.uuid4()),
  "type": "add"
  }
  eventstream.append(streamMessage)
+
  # device
  streamMessage = {"creationtime": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [self.getDevice()],
@@ -313,6 +312,7 @@ def __del__(self):
  }
  streamMessage["id_v1"] = "/lights/" + self.id_v1
  eventstream.append(streamMessage)
+
  ## device ##
  streamMessage = {"creationtime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [{"id": self.getDevice()["id"], "type": "device"}],
@@ -321,6 +321,7 @@ def __del__(self):
  }
  streamMessage["id_v1"] = "/lights/" + self.id_v1
  eventstream.append(streamMessage)
+
  # Zigbee Connectivity
  streamMessage = {"creationtime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [{"id": self.getZigBee()["id"], "type": "zigbee_connectivity"}],
@@ -329,6 +330,7 @@ def __del__(self):
  }
  streamMessage["id_v1"] = "/lights/" + self.id_v1
  eventstream.append(streamMessage)
+
  # Entertainment
  streamMessage = {"creationtime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
  "data": [{"id": self.getV2Entertainment()["id"], "type": "entertainment"}],
@@ -337,6 +339,7 @@ def __del__(self):
  }
  streamMessage["id_v1"] = "/lights/" + self.id_v1
  eventstream.append(streamMessage)
+
  logging.info(self.name + " light was destroyed.")
 
  def update_attr(self, newdata):

BridgeEmulator/flaskUI/restful.py

@@ -74,7 +74,7 @@ def post(self):
  logging.info(postDict)
  if "devicetype" in postDict:
  last_button_press = bridgeConfig["config"]["linkbutton"]["lastlinkbuttonpushed"]
- if last_button_press + 30 >= datetime.now().timestamp():
+ if last_button_press + 30 >= datetime.now().timestamp(): # 30 sec offset
  username = str(uuid.uuid1()).replace('-', '')
  if postDict["devicetype"].startswith("Hue Essentials"):
  username = "hueess" + username[-26:]
@@ -85,8 +85,7 @@ def post(self):
  # client_key = "321c0c2ebfa7361e55491095b2f5f9db"
 
  response[0]["success"]["clientkey"] = client_key
- bridgeConfig["apiUsers"][username] = HueObjects.ApiUser(
- username, postDict["devicetype"], client_key)
+ bridgeConfig["apiUsers"][username] = HueObjects.ApiUser(username, postDict["devicetype"], client_key)
  logging.debug(response)
  configManager.bridgeConfig.save_config()
  return response
@@ -113,8 +112,7 @@ def get(self, username):
  result[resource] = {}
  for resource_id in bridgeConfig[resource]:
  if resource_id != "0":
- result[resource][resource_id] = bridgeConfig[resource][resource_id].getV1Api(
- ).copy()
+ result[resource][resource_id] = bridgeConfig[resource][resource_id].getV1Api().copy()
  return result
 
 
@@ -128,8 +126,7 @@ def get(self, username, resource):
  response = {}
  if resource in ["lights", "groups", "scenes", "rules", "resourcelinks", "schedules", "sensors"]:
  for object in bridgeConfig[resource]:
- response[object] = bridgeConfig[resource][object].getV1Api(
- ).copy()
+ response[object] = bridgeConfig[resource][object].getV1Api().copy()
  elif resource == "config":
  response = buildConfig()
  return response
@@ -152,8 +149,7 @@ def post(self, username, resource):
  postDict = request.get_json(force=True)
  logging.info(postDict)
  if resource == "lights": # add light manually from the web interface
- Thread(target=manualAddLight, args=[
- postDict["ip"], postDict["protocol"], postDict["config"]]).start()
+ Thread(target=manualAddLight, args=[postDict["ip"], postDict["protocol"], postDict["config"]]).start()
  return [{"success": {"/" + resource: "Searching for new devices"}}]
  v2Resource = None
  # find the first unused id for new object
@@ -224,14 +220,12 @@ def post(self, username, resource):
  elif resource == "rules":
  bridgeConfig[resource][new_object_id] = HueObjects.Rule(postDict)
  elif resource == "resourcelinks":
- bridgeConfig[resource][new_object_id] = HueObjects.ResourceLink(
- postDict)
+ bridgeConfig[resource][new_object_id] = HueObjects.ResourceLink(postDict)
  elif resource == "sensors":
  v2Resource = "device"
  bridgeConfig[resource][new_object_id] = HueObjects.Sensor(postDict)
  elif resource == "schedules":
- bridgeConfig[resource][new_object_id] = HueObjects.Schedule(
- postDict)
+ bridgeConfig[resource][new_object_id] = HueObjects.Schedule(postDict)
  newObject = bridgeConfig[resource][new_object_id]
  if v2Resource != "none":
  streamMessage = {"creationtime": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
@@ -279,8 +273,7 @@ def put(self, username, resource):
  responseList = []
  response_location = "/" + resource + "/"
  for key, value in putDict.items():
- responseList.append(
- {"success": {response_location + key: value}})
+ responseList.append({"success": {response_location + key: value}})
  logging.debug(responseList)
  configManager.bridgeConfig.save_config(backup=False, resource=resource)
  return responseList
@@ -453,9 +446,8 @@ def delete(self, username, resource, resourceid, param):
  return [{"error": {"type": 4, "address": "/" + resource + "/" + resourceid, "description": "method, DELETE, not available for resource, " + resource + "/" + resourceid}}]
 
  del bridgeConfig[resource][resourceid][param]
- return [{"success": "/" + resource + "/" + resourceid + "/" + param + " deleted."}]
  configManager.bridgeConfig.save_config()
-
+ return [{"success": "/" + resource + "/" + resourceid + "/" + param + " deleted."}]
 
 class ElementParamId(Resource):
  def get(self, username, resource, resourceid, param, paramid):

BridgeEmulator/lights/protocols/shelly.py

@@ -7,50 +7,86 @@
 #bridgeConfig = configManager.bridgeConfig.yaml_config
 #newLights = configManager.runtimeConfig.newLights
 
-def sendRequest(url, timeout=3):
- head = {"Content-type": "application/json"}
- response = requests.get(url, timeout=timeout, headers=head)
- return response.text
-
-
 def discover(detectedLights, device_ips):
- logging.debug("shelly: <discover> invoked!")
+ logging.debug('shelly: <discover> invoked!')
  for ip in device_ips:
  try:
- logging.debug("shelly: probing ip " + ip)
- response = requests.get("http://" + ip + "/shelly", timeout=3)
+ logging.debug('shelly: probing ip ' + ip)
+ response = requests.get('http://' + ip + '/shelly', timeout = 5)
  if response.status_code == 200:
+ logging.debug('Shelly: ' + ip + ' is a shelly device ')
  device_data = json.loads(response.text)
- if device_data["type"] == "SHSW-1":
 
- logging.debug("shelly: " + ip + " is a shelly device ")
- shelly_response = requests.get("http://" + ip + "/status", timeout=5)
- shelly_data = json.loads(shelly_response.text)
- logging.debug("shelly: ip: " + shelly_data["wifi_sta"]["ip"])
- logging.debug("shelly: Mac: " + shelly_data["mac"])
- detectedLights.append({"protocol": "shelly", "name": ip, "modelid": "LOM001", "protocol_cfg": {"ip": ip, "mac": shelly_data["mac"]}})
+ device_model = ''
+ if (not 'gen' in device_data) and ('type' in device_data):
+ device_model = device_data['type']
+ elif ('gen' in device_data) and ('model' in device_data):
+ device_model = device_data['model']
+ else:
+ logging.info('Shelly: <discover> not implemented api version!')
+
+ if (device_model == 'SHSW-1') or (device_model == 'SHSW-PM'):
+ shelly_data = request_api_v1(ip, 'status')
+ logging.debug('Shelly: IP: ' + shelly_data['wifi_sta']['ip'])
+ logging.debug('Shelly: MAC: ' + shelly_data['mac'])
+
+ config = {'ip': ip, 'mac': shelly_data['mac'], 'gen': 1}
+
+ shelly_data = request_api_v1(ip, 'settings')
+ name = shelly_data['name'] if 'name' in shelly_data else ip
+ name = name.strip() if name.strip() != '' else ip
+ detectedLights.append({'protocol': 'shelly', 'name': name, 'modelid': 'LOM001', 'protocol_cfg': config})
+ elif (device_model == 'SNSW-001P8EU'):
+ shelly_data = request_api_v2(ip, 'WiFi.GetStatus')
+ logging.debug('Shelly: IP: ' + shelly_data['sta_ip'])
+ shelly_data = request_api_v2(ip, 'Shelly.GetDeviceInfo')
+ logging.debug('Shelly: MAC: ' + shelly_data['mac'])
 
- except Exception as e:
- logging.debug("shelly: ip " + ip + " is unknow device, " + str(e))
+ config = {'ip': ip, 'mac': shelly_data['mac'], 'gen': device_data['gen'] }
 
+ name = shelly_data['name'] if 'name' in shelly_data else ip
+ name = name.strip() if name.strip() != '' else ip
+ detectedLights.append({'protocol': 'shelly', 'name': name, 'modelid': 'LOM001', 'protocol_cfg': config})
+ else:
+ logging.info('Shelly: ' + ip + ' is not supported ')
+ except Exception as exception:
+ logging.debug('Shelly: IP ' + ip + ' is unknown device, ' + str(exception))
 
 def set_light(light, data):
- logging.debug("shelly: <set_light> invoked! IP=" + light.protocol_cfg["ip"])
+ config = light.protocol_cfg
+ logging.debug('Shelly: <set_light> invoked! IP=' + config['ip'])
 
  for key, value in data.items():
- if key == "on":
- if value:
- sendRequest("http://" + light.protocol_cfg["ip"] + "/relay/0/?turn=on")
+ if key == 'on':
+ if (not 'gen' in config) or (config['gen'] == 1):
+ request_api_v1(config['ip'], 'relay/0?turn=' + ('on' if value else 'off'))
+ elif config['gen'] == 2:
+ request_api_v2(config['ip'], 'Switch.Set?id=0&on=' + str(value).lower())
  else:
- sendRequest("http://" + light.protocol_cfg["ip"] + "/relay/0/?turn=off")
-
+ logging.info('Shelly: <set_light> not implemented api version!')
 
 def get_light_state(light):
- logging.debug("shelly: <get_light_state> invoked!")
- data = sendRequest("http://" + light.protocol_cfg["ip"] + "/relay/0")
- light_data = json.loads(data)
+ config = light.protocol_cfg
+ logging.debug('Shelly: <get_light_state> invoked! IP=' + config['ip'])
+
  state = {}
+ if (not 'gen' in config) or (config['gen'] == 1):
+ data = request_api_v1(config['ip'], 'relay/0')
+ state['on'] = data['ison'] if 'ison' in data else False
+ elif config['gen'] == 2:
+ data = request_api_v2(config['ip'], 'Switch.GetStatus?id=0')
+ state['on'] = data['output'] if 'output' in data else False
+ else:
+ logging.info('Shelly: <get_light_state> not implemented api version!')
 
- if 'ison' in light_data:
- state['on'] = True if light_data["ison"] == "true" else False
  return state
+
+def request_api_v1(ip, request):
+ head = {'Content-type': 'application/json'}
+ response = requests.get('http://' + ip + '/' + request, timeout = 5, headers = head)
+ return json.loads(response.text) if response.status_code == 200 else {}
+
+def request_api_v2(ip, request):
+ head = {'Content-type': 'application/json'}
+ response = requests.get('http://' + ip + '/rpc/' + request, timeout = 5, headers = head)
+ return json.loads(response.text) if response.status_code == 200 else {}