diff --git a/.github/workflows/hacs_validate.yaml b/.github/workflows/hacs_validate.yaml new file mode 100644 index 0000000..d074026 --- /dev/null +++ b/.github/workflows/hacs_validate.yaml @@ -0,0 +1,18 @@ +name: HACS Validate + +on: + push: + pull_request: + schedule: + - cron: "0 0 * * *" + workflow_dispatch: + +jobs: + validate-hacs: + runs-on: "ubuntu-latest" + steps: + - uses: "actions/checkout@v3" + - name: HACS validation + uses: "hacs/action@main" + with: + category: "integration" \ No newline at end of file diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml new file mode 100644 index 0000000..802cfb9 --- /dev/null +++ b/.github/workflows/tests.yaml @@ -0,0 +1,32 @@ +name: All tests + +on: + push: + branches: [ "main" ] + pull_request: + branches: [ "main" ] + schedule: + - cron: '0 0 * * 0' + workflow_dispatch: + +jobs: + build: + + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + python-version: ["3.10", "3.11"] + + steps: + - uses: actions/checkout@v3 + - name: Set up Python ${{ matrix.python-version }} + uses: actions/setup-python@v3 + with: + python-version: ${{ matrix.python-version }} + - name: Install dependencies + run: | + pip install -r requirements_dev.txt + - name: Test with pytest + run: | + pytest -p no:sugar \ No newline at end of file diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..feae5c1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +*.pyc +__pycache__ \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..9e2bb5a --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,3 @@ +# 1.0.0 + +* Initial release \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..c142047 --- /dev/null +++ b/README.md @@ -0,0 +1,59 @@ +# HKC Alarm Integration for Home Assistant + +This repository contains an unofficial Home Assistant integration for [HKC Alarm](https://www.hkcsecurity.com/) systems, allowing you to control and monitor your HKC Alarm directly from Home Assistant. + +## Installation + +You will need [HACS](https://hacs.xyz) installed in your Home Assistant server. Install the integration by installing this repository as a Custom Repository. Then, navigate to Integrations, Add an Integration and select HKC Alarm. You will then be asked to enter: + +* **Panel ID**: Your HKC Alarm Panel ID. +* **Panel Password**: Your HKC Alarm Panel Password. +* **Alarm Code**: Your HKC Alarm Code. +* **Update Interval (seconds)**: (Optional) Custom update interval for fetching data from HKC Alarm. Default is 60 seconds. Recommend keeping this at 60s, as this is similar to the Mobile App's polling interval, and we want to respect HKC's API. + +[![Open your Home Assistant instance and add this integration](https://my.home-assistant.io/badges/config_flow_start.svg)](https://my.home-assistant.io/redirect/config_flow_start/?domain=hkc_alarm) + +## Entities + +The integration updates data every minute by default. It exposes the following entities: + +* An alarm control panel entity representing the HKC Alarm system. +* Sensor entities for each input on the HKC Alarm system. + +The *State* of the alarm control panel is either `armed_home`, `armed_away`, or `disarmed`. The sensor entities will have states `Open` or `Closed` based on the state of the corresponding input on the HKC Alarm system. + +## Sample Automation to notify about alarm state changes + +```yaml +alias: HKC Alarm State Notifications +description: "" +trigger: + - platform: state + entity_id: alarm_control_panel.hkc_alarm_system +condition: [] +action: + - service: notify.notify + data: + title: 🚨 HKC Alarm Notification 🚨 + message: > + Alarm System is now {{ states('alarm_control_panel.hkc_alarm_system') }} +mode: single +``` + + +## Troubleshooting + +If you encounter issues with this integration, you can enable debug logging for this component by adding the following lines to your `configuration.yaml` file: + +```yaml +logger: + logs: + custom_components.hkc_alarm: debug +``` + +This will produce detailed debug logs which can help in diagnosing the problem. + +## Links + +- [Github Repository](https://github.com/jasonmadigan/pyhkc) +- [HKC Alarm PyPi Package](https://pypi.org/project/pyhkc/) \ No newline at end of file diff --git a/custom_components/hkc_alarm/__init__.py b/custom_components/hkc_alarm/__init__.py new file mode 100644 index 0000000..f4be115 --- /dev/null +++ b/custom_components/hkc_alarm/__init__.py @@ -0,0 +1,41 @@ +from pyhkc.hkc_api import HKCAlarm +from datetime import timedelta +from homeassistant.core import callback +from .const import DOMAIN, DEFAULT_UPDATE_INTERVAL, CONF_UPDATE_INTERVAL + +async def async_setup_entry(hass, entry): + panel_id = entry.data["panel_id"] + panel_password = entry.data["panel_password"] + user_code = entry.data["user_code"] + + hkc_alarm = await hass.async_add_executor_job( + HKCAlarm, panel_id, panel_password, user_code + ) + + # Get update interval from options, or use default + update_interval = entry.options.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL) + SCAN_INTERVAL = timedelta(seconds=update_interval) + + # Create a dictionary to store both the HKCAlarm instance and SCAN_INTERVAL + entry_data = { + "hkc_alarm": hkc_alarm, + "scan_interval": SCAN_INTERVAL + } + + # Store the dictionary in hass.data + hass.data.setdefault(DOMAIN, {})[entry.entry_id] = entry_data + + @callback + def update_options(entry): + """Update options.""" + nonlocal entry_data + update_interval = entry.options.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL) + entry_data["scan_interval"] = timedelta(seconds=update_interval) + + entry.add_update_listener(update_options) + + # Load platforms + await hass.config_entries.async_forward_entry_setups( + entry, ["alarm_control_panel", "sensor"] + ) + return True diff --git a/custom_components/hkc_alarm/alarm_control_panel.py b/custom_components/hkc_alarm/alarm_control_panel.py new file mode 100644 index 0000000..b821826 --- /dev/null +++ b/custom_components/hkc_alarm/alarm_control_panel.py @@ -0,0 +1,148 @@ +from homeassistant.components.alarm_control_panel import AlarmControlPanelEntity +from .const import DOMAIN +from homeassistant.components.alarm_control_panel import ( + AlarmControlPanelEntity, + AlarmControlPanelEntityFeature, +) +from homeassistant.helpers.update_coordinator import ( + DataUpdateCoordinator, + CoordinatorEntity, + UpdateFailed, +) +from datetime import timedelta +from .const import DOMAIN, CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL + +import logging + +from datetime import timedelta +from homeassistant.core import callback + + +_logger = logging.getLogger(__name__) + + +class HKCAlarmControlPanel(AlarmControlPanelEntity, CoordinatorEntity): + _attr_supported_features = ( + AlarmControlPanelEntityFeature.ARM_HOME + | AlarmControlPanelEntityFeature.ARM_AWAY + ) + + def __init__(self, data, device_info, coordinator): + AlarmControlPanelEntity.__init__(self) + CoordinatorEntity.__init__(self, coordinator) + self._hkc_alarm = data.get('hkc_alarm') + self._scan_interval = data.get('scan_interval') + self._device_info = device_info + self._state = None + self._panel_data = None + + @property + def unique_id(self): + """Return the unique ID of the sensor.""" + return self._hkc_alarm.panel_id + "panel" + + @property + def extra_state_attributes(self): + """Return the state attributes of the alarm control panel.""" + if self._panel_data is None: + return None + + # Extract the desired attributes from self._panel_data + attributes = { + "Green LED": self._panel_data['greenLed'], + "Red LED": self._panel_data['redLed'], + "Amber LED": self._panel_data['amberLed'], + "Cursor On": self._panel_data['cursorOn'], + "Cursor Index": self._panel_data['cursorIndex'], + "Display": self._panel_data['display'], + "Blink": self._panel_data['blink'], + } + return attributes + + @property + def device_info(self): + return { + "identifiers": {(DOMAIN, self._hkc_alarm.panel_id)}, + "name": "HKC Alarm System", + "manufacturer": "HKC", + "model": "HKC Alarm", + "sw_version": "1.0.0", + } + + @property + def state(self): + return self._state + + @property + def name(self): + return "HKC Alarm System" + + @property + def available(self) -> bool: + """Return True if alarm is available.""" + return self._panel_data is not None and "display" in self._panel_data + + async def async_alarm_disarm(self, code: str | None = None) -> None: + """Send disarm command.""" + await self.hass.async_add_executor_job(self._hkc_alarm.disarm) + + async def async_alarm_arm_home(self, code: str | None = None) -> None: + """Send arm home command.""" + await self.hass.async_add_executor_job(self._hkc_alarm.arm_partset_a) + + async def async_alarm_arm_away(self, code: str | None = None) -> None: + """Send arm away command.""" + await self.hass.async_add_executor_job(self._hkc_alarm.arm_fullset) + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + _logger.error("self.coordinator.data") + _logger.error(self.coordinator.data) + status, panel_data = self.coordinator.data + self._panel_data = panel_data + blocks = status.get("blocks", []) + + if any(block["armState"] == 3 for block in blocks): + self._state = "armed_away" + elif any(block["armState"] == 1 for block in blocks): + self._state = "armed_home" + else: + self._state = "disarmed" + self.async_write_ha_state() # Update the state with the latest data + + +async def async_setup_entry(hass, entry, async_add_entities): + hkc_alarm = hass.data[DOMAIN][entry.entry_id] + update_interval = entry.data.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL) + + async def _async_fetch_data(): + try: + hkc_alarm = hass.data[DOMAIN][entry.entry_id]["hkc_alarm"] + status = await hass.async_add_executor_job(hkc_alarm.get_system_status) + panel_data = await hass.async_add_executor_job(hkc_alarm.get_panel) + return status, panel_data + except Exception as e: + _logger.error(f"Exception occurred while fetching data: {e}") + raise UpdateFailed(f"Failed to update: {e}") + + + coordinator = DataUpdateCoordinator( + hass, + _logger, + name="hkc_alarm_data", + update_method=_async_fetch_data, + update_interval=timedelta(seconds=update_interval), + ) + + await coordinator.async_config_entry_first_refresh() + + device_info = { + "identifiers": {(DOMAIN, entry.entry_id)}, + "name": "HKC Alarm System", + "manufacturer": "HKC", + } + async_add_entities( + [HKCAlarmControlPanel(hkc_alarm, device_info, coordinator)], + True, + ) diff --git a/custom_components/hkc_alarm/config_flow.py b/custom_components/hkc_alarm/config_flow.py new file mode 100644 index 0000000..9d6d44f --- /dev/null +++ b/custom_components/hkc_alarm/config_flow.py @@ -0,0 +1,89 @@ +"""Config flow for HKC Alarm.""" +import logging +from pyhkc.hkc_api import HKCAlarm +from homeassistant import config_entries, exceptions +import voluptuous as vol + +from .const import DOMAIN, DEFAULT_UPDATE_INTERVAL, CONF_UPDATE_INTERVAL + +from .const import DOMAIN + +_LOGGER = logging.getLogger(__name__) + + +class HKCAlarmConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): + """Handle a config flow for HKC Alarm.""" + + VERSION = 2 + + async def async_step_user(self, user_input=None): + """Handle the initial step.""" + errors = {} + + if user_input is not None: + alarm_code = user_input["alarm_code"] + panel_password = user_input["panel_password"] + panel_id = user_input["panel_id"] + + # Initialize the HKCAlarm class in the executor + api = await self.hass.async_add_executor_job( + HKCAlarm, panel_id, panel_password, alarm_code + ) + + # Using the new check_login method + is_authenticated = await self.hass.async_add_executor_job(api.check_login) + + if not is_authenticated: + errors["base"] = "invalid_auth" + return self.async_show_form( + step_id="user", + data_schema=vol.Schema( + { + vol.Required("alarm_code"): str, + vol.Required("panel_password"): str, + vol.Required("panel_id"): str, + } + ), + errors=errors, + ) + return self.async_create_entry( + title="HKC Alarm", + data={ + "panel_id": panel_id, + "panel_password": panel_password, + "user_code": alarm_code, + CONF_UPDATE_INTERVAL: user_input[CONF_UPDATE_INTERVAL], # include update_interval in the entry data + }, + ) + + return self.async_show_form( + step_id="user", + data_schema=vol.Schema( + { + vol.Required("alarm_code"): str, + vol.Required("panel_password"): str, + vol.Required("panel_id"): str, + vol.Optional(CONF_UPDATE_INTERVAL, default=DEFAULT_UPDATE_INTERVAL): int, # include update_interval in the schema + } + ), + errors=errors, + ) + + async def async_step_options(self, user_input=None): + """Handle the options step.""" + if user_input is not None: + return self.async_create_entry(title="", data=user_input) + + options_schema = vol.Schema( + { + vol.Optional( + CONF_UPDATE_INTERVAL, + default=self.entry.options.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL), + ): int, + } + ) + + return self.async_show_form( + step_id="options", + data_schema=options_schema, + ) diff --git a/custom_components/hkc_alarm/const.py b/custom_components/hkc_alarm/const.py new file mode 100644 index 0000000..31f3af5 --- /dev/null +++ b/custom_components/hkc_alarm/const.py @@ -0,0 +1,6 @@ +"""Constants for the HKC Alarm integration.""" + +# Integration domain identifier. This is the name used in the configuration.yaml file. +DOMAIN = "hkc_alarm" +DEFAULT_UPDATE_INTERVAL = 60 # Default update interval in seconds +CONF_UPDATE_INTERVAL = "update_interval" \ No newline at end of file diff --git a/custom_components/hkc_alarm/manifest.json b/custom_components/hkc_alarm/manifest.json new file mode 100644 index 0000000..2261d5e --- /dev/null +++ b/custom_components/hkc_alarm/manifest.json @@ -0,0 +1,15 @@ +{ + "domain": "hkc_alarm", + "name": "HKC Alarm", + "version": "1.0.0", + "documentation": "https://github.com/jasonmadigan/pyhkc", + "dependencies": [], + "codeowners": [ + "@jasonmadigan" + ], + "requirements": [ + "pyhkc==0.4.1" + ], + "config_flow": true, + "iot_class": "cloud_polling" +} \ No newline at end of file diff --git a/custom_components/hkc_alarm/sensor.py b/custom_components/hkc_alarm/sensor.py new file mode 100644 index 0000000..27edebb --- /dev/null +++ b/custom_components/hkc_alarm/sensor.py @@ -0,0 +1,199 @@ +import logging +import asyncio +import pytz +from homeassistant.core import callback +from homeassistant.helpers.entity import Entity +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed +from .const import DOMAIN +from datetime import datetime, timedelta +from homeassistant.helpers.service import async_register_admin_service +import voluptuous as vol +from homeassistant.helpers.event import async_track_time_interval +from .const import DOMAIN, DEFAULT_UPDATE_INTERVAL, CONF_UPDATE_INTERVAL +from homeassistant.helpers.update_coordinator import CoordinatorEntity + + + +_logger = logging.getLogger(__name__) + +class HKCSensor(CoordinatorEntity): + _panel_data = {} # Class variable shared among all instances + _update_lock = asyncio.Lock() # Lock to ensure only one update at a time + _last_update = datetime.min # Initialize with the earliest possible datetime + + def __init__(self, hkc_alarm, input_data, coordinator): + super().__init__(coordinator) # Ensure the coordinator is properly initialized + self._hkc_alarm = hkc_alarm + self._input_data = input_data + self.coordinator = coordinator + + @property + def unique_id(self): + """Return the unique ID of the sensor.""" + return self._hkc_alarm.panel_id + str(self._input_data['inputId']) + + @property + def device_info(self): + return { + "identifiers": {(DOMAIN, self._hkc_alarm.panel_id)}, + "name": "HKC Alarm System", + "manufacturer": "HKC", + "model": "HKC Alarm", + "sw_version": "1.0.0", + } + + @property + def state(self): + """Determine the state of the sensor.""" + + # Check for the default timestamp + if self._input_data["timestamp"] == "0001-01-01T00:00:00": + _logger.debug( + f"Sensor {self.name} state determined as 'Unused' due to default timestamp." + ) + return "Unused" + + # Parse panel time + panel_time_str = self._panel_data.get('display', '') + + try: + panel_time = datetime.strptime(panel_time_str, "%a %d %b %H:%M") + except ValueError: + _logger.debug( + f"Failed to parse panel time: {panel_time_str} for sensor {self.name}. Setting state to 'Unknown'." + ) + return "Unknown" # Return an unknown state if panel time parsing fails + + # Ensure Panel Time is in UTC + current_year = datetime.utcnow().year + panel_time = panel_time.replace(year=current_year, tzinfo=pytz.UTC) + + # Parse sensor timestamp + try: + sensor_timestamp = datetime.strptime( + self._input_data["timestamp"], "%Y-%m-%dT%H:%M:%SZ" + ).replace(tzinfo=pytz.UTC) # Ensure sensor_timestamp is treated as UTC + except ValueError: + try: + sensor_timestamp = datetime.strptime( + self._input_data["timestamp"], "%Y-%m-%dT%H:%M:%S" + ).replace(tzinfo=pytz.UTC) # Ensure sensor_timestamp is treated as UTC + except ValueError: + _logger.debug( + f"Failed to parse timestamp: {self._input_data['timestamp']} for sensor {self.name}. Setting state to 'Unknown'." + ) + return "Unknown" # Return an unknown state if timestamp parsing fails + + time_difference = panel_time - sensor_timestamp + + # Handle cases where the timestamp is very old or invalid + if time_difference > timedelta(days=365): + _logger.debug( + f"Sensor {self.name} has an old timestamp: {self._input_data['timestamp']}. Setting state to 'Closed'." + ) + return "Closed" # Or return "Unknown" if you prefer + + # Check if the time difference is within 60 seconds (maximum panel time resolution) to determine 'Open' state + if abs(time_difference) < timedelta(seconds=60): + _logger.debug( + f"Sensor {self.name} state determined as 'Open' due to timestamp within 30 seconds of panel time." + ) + return "Open" + elif self._input_data["inputState"] == 1: + _logger.debug( + f"Sensor {self.name} state determined as 'Open' due to inputState being 1." + ) + return "Open" + else: + _logger.debug(f"Sensor {self.name} state determined as 'Closed'.") + return "Closed" + + + @property + def name(self): + return self._input_data["description"] + + @property + def should_poll(self): + """Return False, entities are updated by the coordinator.""" + return False + + @callback + def _handle_coordinator_update(self) -> None: + """Handle updated data from the coordinator.""" + # Search for the matching sensor data based on inputId + matching_sensor_data = next( + (sensor_data for sensor_data in self.coordinator.data if sensor_data.get('inputId') == self._input_data.get('inputId')), + None # Default to None if no matching sensor data is found + ) + + if matching_sensor_data is not None: + # Update self._input_data with the matching sensor data + self._input_data = matching_sensor_data + else: + _logger.warning(f"No matching sensor data found for inputId {self._input_data.get('inputId')}") + + self.async_write_ha_state() # Update the state with the latest data + + + @classmethod + async def update_panel_data(cls, hkc_alarm, hass): + async with cls._update_lock: + now = datetime.utcnow() + if (now - cls._last_update) < timedelta(seconds=30): # 30 seconds cooldown + return cls._panel_data # Return existing data if updated recently + + cls._panel_data = await hass.async_add_executor_job(hkc_alarm.get_panel) + cls._last_update = now # Update the last update timestamp + return cls._panel_data # Return the updated data + + async def async_update(self): + """Update the sensor.""" + _logger.debug(f"Updating sensor {self.name}") + await self.coordinator.async_request_refresh() + + +async def async_setup_entry(hass, entry, async_add_entities): + hkc_alarm = hass.data[DOMAIN][entry.entry_id] + update_interval = entry.data.get(CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL) + + async def _async_fetch_data(): + try: + hkc_alarm = hass.data[DOMAIN][entry.entry_id]["hkc_alarm"] + await HKCSensor.update_panel_data(hkc_alarm, hass) + data = await hass.async_add_executor_job(hkc_alarm.get_all_inputs) + return data + except Exception as e: + _logger.error(f"Exception occurred while fetching HKC data: {e}") + raise UpdateFailed(f"Failed to update: {e}") + + coordinator = DataUpdateCoordinator( + hass, + _logger, + name="hkc_sensor_data", + update_method=_async_fetch_data, + update_interval=timedelta(seconds=update_interval), + always_update=True, + ) + + await coordinator.async_config_entry_first_refresh() + + # Register a custom service for testing + async def async_force_refresh(service_call): + """Force refresh data from HKC.""" + await coordinator.async_request_refresh() + + hass.services.async_register( + DOMAIN, + "force_refresh", + async_force_refresh, + schema=vol.Schema({}), + ) + + all_inputs = coordinator.data + # Filter out the inputs with empty description + filtered_inputs = [input_data for input_data in all_inputs if input_data['description']] + async_add_entities( + [HKCSensor(hass.data[DOMAIN][entry.entry_id]["hkc_alarm"], input_data, coordinator) for input_data in filtered_inputs], + True, + ) \ No newline at end of file diff --git a/custom_components/hkc_alarm/translations/en.json b/custom_components/hkc_alarm/translations/en.json new file mode 100644 index 0000000..8c7285b --- /dev/null +++ b/custom_components/hkc_alarm/translations/en.json @@ -0,0 +1,29 @@ +{ + "config": { + "step": { + "user": { + "description": "Integration for HKC Alarm systems.", + "data": { + "alarm_code": "Alarm Code", + "panel_password": "Panel Password", + "panel_id": "Panel ID", + "update_interval": "Update Interval (seconds)" + } + }, + "options": { + "title": "Options", + "data": { + "update_interval": "Update Interval (seconds)" + } + } + }, + "error": { + "cannot_connect": "Unable to connect to the HKC Alarm.", + "invalid_auth": "Invalid authentication details.", + "unknown": "An unknown error occurred." + }, + "abort": { + "already_configured": "HKC Alarm integration is already configured." + } + } +} \ No newline at end of file diff --git a/hacs.json b/hacs.json new file mode 100644 index 0000000..e69de29 diff --git a/requirements_dev.txt b/requirements_dev.txt new file mode 100644 index 0000000..4f0db20 --- /dev/null +++ b/requirements_dev.txt @@ -0,0 +1,6 @@ +aiohttp==3.8.5 +homeassistant==2023.10.5 +pytest +pytest-homeassistant-custom-component==0.13.36 +pytest-aiohttp +pyhkc==0.4.1 \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_sensor.py b/tests/test_sensor.py new file mode 100644 index 0000000..9d0115f --- /dev/null +++ b/tests/test_sensor.py @@ -0,0 +1,124 @@ +import pytest +from unittest.mock import AsyncMock, patch +from custom_components.hkc_alarm.sensor import HKCSensor +from custom_components.hkc_alarm.const import DOMAIN +from unittest.mock import Mock +from unittest.mock import MagicMock + + +# Mock data for HKCSensor to process +mock_sensor_data = { + "inputId": "1", + "description": "Front Door", + "timestamp": "2023-10-25T08:00:00Z", + "inputState": 1, +} + + +class MockCoordinator: + async_request_refresh = AsyncMock() + last_update_success = True # or False, depending on what you want to test + + +class MockHKCAlarm: + panel_id = "hkc_alarm_instance" + + +@pytest.mark.asyncio +async def test_hkc_sensor_state(hass, aioclient_mock): + with patch.object(HKCSensor, "_panel_data", {"display": "Thu 26 Oct 10:35"}): + mock_coordinator = MockCoordinator() # Create a mock coordinator instance + sensor = HKCSensor( + "hkc_alarm_instance", mock_sensor_data, mock_coordinator + ) # Pass the mock coordinator here + await sensor.async_update() + + print(f"Mock Sensor Data: {mock_sensor_data}") # Print the mock data + print(f"Sensor State: {sensor.state}") # Print the actual state + + assert ( + sensor.state == "Open" + ) # as per your logic, inputState being 1 should result in state "Open" + + +@pytest.mark.asyncio +async def test_hkc_sensor_invalid_timestamp(hass, aioclient_mock): + with patch( + "custom_components.hkc_alarm.sensor.HKCSensor.update_panel_data", + new_callable=AsyncMock, + ) as mock_update: + # Include 'display' field in the mocked return value + mock_update.return_value = { + **mock_sensor_data, + "timestamp": "invalid_timestamp", + "display": "Thu 26 Oct 10:35", + } + + mock_coordinator = MockCoordinator() # Create a mock coordinator instance + sensor = HKCSensor("hkc_alarm_instance", mock_sensor_data, mock_coordinator) + await sensor.async_update() + + assert ( + sensor.state == "Unknown" + ) # as per your logic, an invalid timestamp should result in state "Unknown" + + +@pytest.mark.asyncio +async def test_device_info(): + mock_hkc_alarm = MockHKCAlarm() + mock_coordinator = MockCoordinator() + sensor = HKCSensor(mock_hkc_alarm, mock_sensor_data, mock_coordinator) + expected_device_info = { + "identifiers": {(DOMAIN, "hkc_alarm_instance")}, + "name": "HKC Alarm System", + "manufacturer": "HKC", + "model": "HKC Alarm", + "sw_version": "1.0.0", + } + assert sensor.device_info == expected_device_info + + +@pytest.mark.asyncio +async def test_name(): + mock_coordinator = MockCoordinator() + sensor = HKCSensor("hkc_alarm_instance", mock_sensor_data, mock_coordinator) + assert sensor.name == "Front Door" # Assuming description is 'Front Door' + + +@pytest.mark.asyncio +async def test_should_poll(): + mock_coordinator = MockCoordinator() + sensor = HKCSensor("hkc_alarm_instance", mock_sensor_data, mock_coordinator) + assert sensor.should_poll is False + + +@pytest.mark.asyncio +async def test_handle_coordinator_update(): + mock_coordinator = MockCoordinator() + sensor = HKCSensor("hkc_alarm_instance", mock_sensor_data, mock_coordinator) + + # Create a MagicMock for the hass attribute + mock_hass = MagicMock() + # Set up the get method of the data attribute to return an empty dictionary + mock_hass.data.get.return_value = {} + # Assign the mock_hass object to the hass attribute of the sensor + sensor.hass = mock_hass + + sensor.entity_id = "sensor.front_door" # Set the entity_id manually + new_data = { + "inputId": "1", + "description": "Front Door", + "timestamp": "2023-10-26T08:00:00Z", + "inputState": 0, + } + mock_coordinator.data = [new_data] # Updating the coordinator data + sensor._handle_coordinator_update() + assert sensor._input_data == new_data # Check that _input_data was updated + + +@pytest.mark.asyncio +async def test_async_update(): + mock_coordinator = MockCoordinator() + sensor = HKCSensor("hkc_alarm_instance", mock_sensor_data, mock_coordinator) + await sensor.async_update() + mock_coordinator.async_request_refresh.assert_called() # Verify that a refresh request was made