Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# If the Vera disconnects before writing a full response (as lu_sdata
# will do when interrupted by a Luup reload), the requests module will
# happily return 200 with an empty string. So, test for empty response,
# so we don't rely on the JSON parser to throw an exception.
if response.text == "":
raise PyveraError("Empty response from Vera")
# Catch a wide swath of what the JSON parser might throw, within
# reason. Unfortunately, some parsers don't specifically return
# json.decode.JSONDecodeError, but so far most seem to derive what
# they do throw from ValueError, so that's helpful.
try:
result = response.json()
except ValueError as ex:
raise PyveraError("JSON decode error: " + str(ex))
if not (
isinstance(result, dict)
and "loadtime" in result
and "dataversion" in result
):
raise PyveraError("Unexpected/garbled response from Vera")
# At this point, all good. Update timestamp and return change data.
device_data = result.get("devices", [])
timestamp = {
"loadtime": result.get("loadtime", 0),
"dataversion": result.get("dataversion", 1),
}
return device_data, timestamp
# Catch a wide swath of what the JSON parser might throw, within
# reason. Unfortunately, some parsers don't specifically return
# json.decode.JSONDecodeError, but so far most seem to derive what
# they do throw from ValueError, so that's helpful.
try:
result = response.json()
except ValueError as ex:
raise PyveraError("JSON decode error: " + str(ex))
if not (
isinstance(result, dict)
and "loadtime" in result
and "dataversion" in result
):
raise PyveraError("Unexpected/garbled response from Vera")
# At this point, all good. Update timestamp and return change data.
device_data = result.get("devices", [])
timestamp = {
"loadtime": result.get("loadtime", 0),
"dataversion": result.get("dataversion", 1),
}
return device_data, timestamp
data_changed = False
try:
LOG.debug("Polling for Vera changes")
device_data, new_timestamp = self.get_device_data(self._last_updated)
if (
new_timestamp["dataversion"] != self._last_updated["dataversion"]
or self.always_update()
):
alert_data = self.get_alert_data(self._last_updated)
data_changed = True
else:
data_changed = False
self._last_updated = new_timestamp
except requests.RequestException as ex:
LOG.debug("Caught RequestException: %s", str(ex))
except PyveraError as ex:
LOG.debug("Non-fatal error in poll: %s", str(ex))
except Exception as ex:
LOG.exception("Vera poll thread general exception: %s", str(ex))
raise
else:
LOG.debug("Poll returned")
if data_changed or self.always_update():
self._event(device_data, alert_data)
else:
LOG.debug("No changes in poll interval")
return True
# After error, discard timestamp for fresh update. pyvera issue #89
self._last_updated = {"dataversion": 1, "loadtime": 0}
LOG.info("Could not poll Vera")
invocation of get_changed_devices. Use a timestamp of TIMESTAMP_NONE
for the first invocation.
"""
payload = {
"LoadTime": timestamp["loadtime"],
"DataVersion": timestamp["dataversion"],
"id": "status",
}
LOG.debug("get_alerts() requesting payload %s", str(payload))
response = self.data_request(payload)
response.raise_for_status()
if response.text == "":
raise PyveraError("Empty response from Vera")
try:
result = response.json()
except ValueError as ex:
raise PyveraError("JSON decode error: " + str(ex))
if not (
isinstance(result, dict)
and "LoadTime" in result
and "DataVersion" in result
):
raise PyveraError("Unexpected/garbled response from Vera")
return result.get("alerts", [])
response.raise_for_status()
if response.text == "":
raise PyveraError("Empty response from Vera")
try:
result = response.json()
except ValueError as ex:
raise PyveraError("JSON decode error: " + str(ex))
if not (
isinstance(result, dict)
and "LoadTime" in result
and "DataVersion" in result
):
raise PyveraError("Unexpected/garbled response from Vera")
return result.get("alerts", [])
"LoadTime": timestamp["loadtime"],
"DataVersion": timestamp["dataversion"],
"id": "status",
}
LOG.debug("get_alerts() requesting payload %s", str(payload))
response = self.data_request(payload)
response.raise_for_status()
if response.text == "":
raise PyveraError("Empty response from Vera")
try:
result = response.json()
except ValueError as ex:
raise PyveraError("JSON decode error: " + str(ex))
if not (
isinstance(result, dict)
and "LoadTime" in result
and "DataVersion" in result
):
raise PyveraError("Unexpected/garbled response from Vera")
return result.get("alerts", [])
"minimumdelay": SUBSCRIPTION_MIN_WAIT,
"id": "lu_sdata",
}
payload.update(timestamp)
# double the timeout here so requests doesn't timeout before vera
LOG.debug("get_changed_devices() requesting payload %s", str(payload))
response = self.data_request(payload, TIMEOUT * 2)
response.raise_for_status()
# If the Vera disconnects before writing a full response (as lu_sdata
# will do when interrupted by a Luup reload), the requests module will
# happily return 200 with an empty string. So, test for empty response,
# so we don't rely on the JSON parser to throw an exception.
if response.text == "":
raise PyveraError("Empty response from Vera")
# Catch a wide swath of what the JSON parser might throw, within
# reason. Unfortunately, some parsers don't specifically return
# json.decode.JSONDecodeError, but so far most seem to derive what
# they do throw from ValueError, so that's helpful.
try:
result = response.json()
except ValueError as ex:
raise PyveraError("JSON decode error: " + str(ex))
if not (
isinstance(result, dict)
and "loadtime" in result
and "dataversion" in result
):
raise PyveraError("Unexpected/garbled response from Vera")