Skip to content

Fixing the pyit600 errors on connection and improving the gateway json response #42

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.

Sign up for GitHub

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

Open
samuellazea wants to merge 5 commits into
base: master
Choose a base branch
from
Open

Fixing the pyit600 errors on connection and improving the gateway json response #42

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
a00f7e9
Update gateway.py
samuellazea Feb 26, 2025
2983711
Update gateway.py to print the Gateway response json pretty
samuellazea Feb 26, 2025
6e42932
Changes related to smoke sensor list
samuellazea Feb 26, 2025
83b2d37
Add logic for Smoke Sensor
samuellazea Feb 28, 2025
9a89179
Small debugging
samuellazea Mar 7, 2025
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
26 changes: 21 additions & 5 deletions pyit600/gateway.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ async def connect(self) -> str:
return gateway["sGateway"]["NetworkLANMAC"]
except IT600ConnectionError as ae:
try:
with async_timeout.timeout(self._request_timeout):
async with async_timeout.timeout(self._request_timeout):
await self._session.get(f"http://{self._host}:{self._port}/")
except Exception:
raise IT600ConnectionError(
Expand Down Expand Up @@ -165,7 +165,7 @@ async def poll_status(self, send_callback=False) -> None:
filter(lambda x: "sIASZS" in x or
("sBasicS" in x and
"ModelIdentifier" in x["sBasicS"] and
x["sBasicS"]["ModelIdentifier"] in ["it600MINITRV", "it600Receiver"]), all_devices["id"])
x["sBasicS"]["ModelIdentifier"] in ["it600MINITRV", "it600Receiver", "SmokeSensor-EM"]), all_devices["id"])
)

await self._refresh_binary_sensor_devices(binary_sensors, send_callback)
Expand Down Expand Up @@ -402,6 +402,7 @@ async def _refresh_sensor_devices(self, devices: List[Any], send_callback=False)

async def _refresh_binary_sensor_devices(self, devices: List[Any], send_callback=False):
local_devices = {}
_LOGGER.debug(f"Logging _refresh_binary_sensor_devices: {devices}")

if devices:
status = await self._make_encrypted_request(
Expand All @@ -422,6 +423,13 @@ async def _refresh_binary_sensor_devices(self, devices: List[Any], send_callback
model: Optional[str] = device_status.get("DeviceL", {}).get("ModelIdentifier_i", None)
if model in ["it600MINITRV", "it600Receiver"]:
is_on: Optional[bool] = device_status.get("sIT600I", {}).get("RelayStatus", None)
elif model == "SmokeSensor-EM":
# First try to get the standard alarm attribute
is_on: Optional[bool] = device_status.get("sIASZS", {}).get("ErrorIASZSAlarmed1", None)
# If it doesn't exist, default to 0 (not alarmed)
if is_on is None:
is_on = 0
_LOGGER.debug(f"Smoke sensor is_on: {is_on}")
else:
is_on: Optional[bool] = device_status.get("sIASZS", {}).get("ErrorIASZSAlarmed1", None)

Expand Down Expand Up @@ -449,6 +457,8 @@ async def _refresh_binary_sensor_devices(self, devices: List[Any], send_callback
)

local_devices[device.unique_id] = device
_LOGGER.debug(f"Detected device: {model}, Device Class: {device}, Unique ID: {device_status['data']['UniID']}")


if send_callback:
self._binary_sensor_devices[device.unique_id] = device
Expand Down Expand Up @@ -919,7 +929,7 @@ async def _make_encrypted_request(self, command: str, request_body: dict) -> Any
if self._debug:
_LOGGER.debug("Gateway request: POST %s\n%s\n", request_url, request_body_json)

with async_timeout.timeout(self._request_timeout):
async with async_timeout.timeout(self._request_timeout):
resp = await self._session.post(
request_url,
data=self._encryptor.encrypt(request_body_json),
Expand All @@ -929,7 +939,12 @@ async def _make_encrypted_request(self, command: str, request_body: dict) -> Any
response_json_string = self._encryptor.decrypt(response_bytes)

if self._debug:
_LOGGER.debug("Gateway response:\n%s\n", response_json_string)
try:
response_json = json.loads(response_json_string) # Parse the string into JSON
_LOGGER.debug("Gateway response:\n%s\n", json.dumps(response_json, indent=4)) # Pretty print JSON
except json.JSONDecodeError as e:
_LOGGER.error("Failed to decode JSON response: %s\n", e)
_LOGGER.debug("Raw Gateway response:\n%s\n", response_json_string)

response_json = json.loads(response_json_string)

Expand All @@ -953,7 +968,8 @@ async def _make_encrypted_request(self, command: str, request_body: dict) -> Any
"check if you have specified host/IP address correctly"
) from e
except Exception as e:
_LOGGER.error("Exception. %s / %s", type(e), repr(e.args), e)
_LOGGER.error("Exception: %s", repr(e))
# _LOGGER.error("Exception. %s / %s", type(e), repr(e.args), e)
raise IT600CommandError(
"Unknown error occurred while communicating with iT600 gateway"
) from e
Expand Down