Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

refactor!: add query.ALERTS to query the system #118

Merged
merged 5 commits into from
Oct 12, 2023
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
85 changes: 55 additions & 30 deletions src/elmo/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -490,6 +490,8 @@ def _get_descriptions(self):

@require_session
def get_status(self):
# This definition is mantained to allow backward compatibility
xtimmy86x marked this conversation as resolved.
Show resolved Hide resolved
# and will be removed in the next major release
"""Retrieve the status and convert its keys to snake_case format.

This method sends a POST request to retrieve the status and then converts
Expand Down Expand Up @@ -580,6 +582,9 @@ def query(self, query):
key_group = "inputs"
endpoint = self._router.inputs
_LOGGER.debug("Client | Querying inputs")
elif query == q.ALERTS:
endpoint = self._router.status
_LOGGER.debug("Client | Querying alerts")
else:
# Bail-out if the query is not recognized
raise QueryNotValid()
Expand All @@ -594,34 +599,54 @@ def query(self, query):
# `excluded` field is available only on inputs, but to return the same `dict`
# structure, we default "excluded" as False for sectors. In fact, sectors
# are never excluded.
entries = response.json()
_LOGGER.debug(f"Client | Query response: {entries}")
items = {}
result = {
"last_id": entries[-1]["Id"],
key_group: items,
}
try:
for entry in entries:
if entry["InUse"]:
# Address potential data inconsistency between cloud data and main unit.
# In some installations, they may be out of sync, resulting in the cloud
# providing a sector/input that doesn't actually exist in the main unit.
# To handle this, we default the name to "Unknown" if its description
# isn't found in the cloud data to prevent KeyError.
name = descriptions[query].get(entry["Index"], "Unknown")
item = {
"id": entry.get("Id"),
"index": entry.get("Index"),
"element": entry.get("Element"),
"excluded": entry.get("Excluded", False),
"status": entry.get(status, False),
"name": name,
}

items[entry.get("Index")] = item
except KeyError as err:
raise ParseError(f"Client | Unable to parse query response: {err}") from err

_LOGGER.debug(f"Client | Query parsed successfully: {result}")
return result
if query == q.SECTORS or query == q.INPUTS:
xtimmy86x marked this conversation as resolved.
Show resolved Hide resolved
entries = response.json()
_LOGGER.debug(f"Client | Query response: {entries}")
items = {}
result = {
"last_id": entries[-1]["Id"],
key_group: items,
}
try:
for entry in entries:
if entry["InUse"]:
# Address potential data inconsistency between cloud data and main unit.
# In some installations, they may be out of sync, resulting in the cloud
# providing a sector/input that doesn't actually exist in the main unit.
# To handle this, we default the name to "Unknown" if its description
# isn't found in the cloud data to prevent KeyError.
name = descriptions[query].get(entry["Index"], "Unknown")
item = {
"id": entry.get("Id"),
"index": entry.get("Index"),
"element": entry.get("Element"),
"excluded": entry.get("Excluded", False),
"status": entry.get(status, False),
"name": name,
}

items[entry.get("Index")] = item
except KeyError as err:
raise ParseError(f"Client | Unable to parse query response: {err}") from err

_LOGGER.debug(f"Client | Query parsed successfully: {result}")
return result

if query == q.ALERTS:
try:
# Check if the response has the expected format
msg = response.json()
status = msg["PanelLeds"]
anomalies = msg["PanelAnomalies"]
except (KeyError, ValueError):
raise ParseError("Unexpected response format from the server.")

# Merge the 'status' and 'anomalies' dictionaries
merged_dict = {**status, **anomalies}

# Convert the dict to a snake_case one to simplify the usage in other modules
snake_case_dict = {_camel_to_snake_case(k): v for k, v in merged_dict.items()}
_LOGGER.debug(f"Client | Status retrieved: {snake_case_dict}")

return snake_case_dict
1 change: 1 addition & 0 deletions src/elmo/query.py
Original file line number Diff line number Diff line change
@@ -1,2 +1,3 @@
SECTORS = 9
INPUTS = 10
ALERTS = 11
83 changes: 83 additions & 0 deletions tests/test_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -1438,6 +1438,89 @@ def test_client_get_inputs_status(server, mocker):
}


def test_client_get_alerts_status(server, mocker):
xtimmy86x marked this conversation as resolved.
Show resolved Hide resolved
"""Should query a Elmo system to retrieve alerts status."""
html = """
{
"StatusUid": 1,
"PanelLeds": {
"InputsLed": 2,
"AnomaliesLed": 1,
"AlarmLed": 0,
"TamperLed": 0
},
"PanelAnomalies": {
"HasAnomaly": false,
"PanelTamper": 0,
"PanelNoPower": 0,
"PanelLowBattery": 0,
"GsmAnomaly": 0,
"GsmLowBalance": 0,
"PstnAnomaly": 0,
"SystemTest": 0,
"ModuleRegistration": 0,
"RfInterference": 0,
"InputFailure": 0,
"InputAlarm": 0,
"InputBypass": 0,
"InputLowBattery": 0,
"InputNoSupervision": 0,
"DeviceTamper": 0,
"DeviceFailure": 0,
"DeviceNoPower": 0,
"DeviceLowBattery": 0,
"DeviceNoSupervision": 0,
"DeviceSystemBlock": 0
},
"PanelAlignmentAdv": {
"ManualFwUpAvailable": false,
"Id": 1,
"Index": -1,
"Element": 0
}
}
"""

# query() depends on _get_descriptions()
server.add(responses.POST, "https://example.com/api/statusadv", body=html, status=200)
client = ElmoClient(base_url="https://example.com", domain="domain")
client._session_id = "test"
mocker.patch.object(client, "_get_descriptions")
client._get_descriptions.return_value = {
10: {0: "Alarm", 1: "Window kitchen", 2: "Door entryway", 3: "Window bathroom"},
}
# Test
alerts = client.query(query.ALERTS)
# Expected output
assert alerts == {
"inputs_led": 2,
"anomalies_led": 1,
"alarm_led": 0,
"tamper_led": 0,
"has_anomaly": False,
"panel_tamper": 0,
"panel_no_power": 0,
"panel_low_battery": 0,
"gsm_anomaly": 0,
"gsm_low_balance": 0,
"pstn_anomaly": 0,
"system_test": 0,
"module_registration": 0,
"rf_interference": 0,
"input_failure": 0,
"input_alarm": 0,
"input_bypass": 0,
"input_low_battery": 0,
"input_no_supervision": 0,
"device_tamper": 0,
"device_failure": 0,
"device_no_power": 0,
"device_low_battery": 0,
"device_no_supervision": 0,
"device_system_block": 0,
}


def test_client_get_sectors_missing_area(server, mocker):
"""Should set an Unknown `sector` name if the description is missing.
Regression test for: https://github.com/palazzem/econnect-python/issues/91"""
Expand Down