Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
abb5064
commit 070860e
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,120 @@ | ||
| """Binary Sensor entities for the DWD Rain Radar integration.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import logging | ||
| from collections.abc import Callable | ||
| from dataclasses import dataclass | ||
| from datetime import datetime, timedelta | ||
|
|
||
| from homeassistant.core import HomeAssistant | ||
| from homeassistant.config_entries import ConfigEntry | ||
| from homeassistant.helpers.entity_platform import AddEntitiesCallback | ||
| from homeassistant.components.binary_sensor import BinarySensorEntity | ||
| from homeassistant.components.binary_sensor import ( | ||
| BinarySensorDeviceClass, | ||
| BinarySensorEntityDescription | ||
| ) | ||
|
|
||
| from .const import DOMAIN, ATTRIBUTION, FORECAST_MINUTES | ||
| from homeassistant.const import ( | ||
| ATTR_ATTRIBUTION | ||
| ) | ||
| from .coordinator import DwdRainRadarUpdateCoordinator, PrecipitationForecast | ||
| from .entity import DwdCoordinatorEntity | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| @dataclass(frozen=True, kw_only=True) | ||
| class BinarySensorEntityDescription(BinarySensorEntityDescription): | ||
| """Provide a description for a precipitation sensor.""" | ||
|
|
||
| is_on_fn: Callable[[PrecipitationForecast]] | ||
| extra_state_attributes_fn: Callable[[PrecipitationForecast], dict] = lambda _: {} | ||
| exists_fn: Callable[[dict], bool] = lambda _: True | ||
|
|
||
|
|
||
| PRECIPTITATION_SENSORS = [ | ||
| BinarySensorEntityDescription( | ||
| key="raining", | ||
| name="Raining", | ||
| device_class=BinarySensorDeviceClass.MOISTURE, | ||
| is_on_fn=lambda forecasts: next( | ||
| (forecast.precipitation > 0 for forecast in forecasts if | ||
| forecast.prediction_time > datetime.now().astimezone() - timedelta(minutes=5)), | ||
| None | ||
| ), | ||
| extra_state_attributes_fn=lambda forecasts: { | ||
| 'prediction_time': next( | ||
| (forecast.prediction_time for forecast in forecasts if | ||
| forecast.prediction_time > datetime.now().astimezone() - timedelta(minutes=5)), | ||
| None | ||
| ) | ||
| }, | ||
| ), | ||
| *(BinarySensorEntityDescription( | ||
| key=f"raining_in_{forecast_in}_minutes", | ||
| name=f"Raining In {forecast_in} Minutes", | ||
| entity_registry_enabled_default=False, | ||
| device_class=BinarySensorDeviceClass.MOISTURE, | ||
| is_on_fn=lambda forecasts, forecast_in=forecast_in: next( | ||
| (forecast.precipitation > 0 for forecast in forecasts if | ||
| forecast.prediction_time > datetime.now().astimezone() + timedelta(minutes=forecast_in - 5)), | ||
| None | ||
| ), | ||
| extra_state_attributes_fn=lambda forecasts, forecast_in=forecast_in: { | ||
| 'prediction_time': next( | ||
| (forecast.prediction_time for forecast in forecasts if | ||
| forecast.prediction_time > datetime.now().astimezone() + timedelta(minutes=forecast_in - 5)), | ||
| None | ||
| ) | ||
| }, | ||
| ) for forecast_in in FORECAST_MINUTES), | ||
| ] | ||
|
|
||
|
|
||
| async def async_setup_entry( | ||
| hass: HomeAssistant, | ||
| entry: ConfigEntry, | ||
| async_add_entities: AddEntitiesCallback | ||
| ) -> None: | ||
| """Set up the sensor platform.""" | ||
| coordinator = hass.data[DOMAIN][entry.entry_id] | ||
| async_add_entities( | ||
| RainingSensorEntity(coordinator, description) | ||
| for description in PRECIPTITATION_SENSORS | ||
| if description.exists_fn(entry) | ||
| ) | ||
|
|
||
|
|
||
| class RainingSensorEntity(DwdCoordinatorEntity, BinarySensorEntity): | ||
| """Implementation of a precipitation sensor.""" | ||
|
|
||
| def __init__( | ||
| self, | ||
| coordinator: DwdRainRadarUpdateCoordinator, | ||
| description: BinarySensorEntityDescription, | ||
| ) -> None: | ||
| """Initialize the sensor entity.""" | ||
| super().__init__(coordinator, description) | ||
|
|
||
| self._attr_unique_id = ( | ||
| f"{self.coordinator.config_entry.entry_id}" | ||
| + f"_{self.entity_description.key}" | ||
| ) | ||
|
|
||
| @property | ||
| def is_on(self): | ||
| """Return the state of the sensor.""" | ||
| return self.entity_description.is_on_fn(self.coordinator.data) | ||
|
|
||
| @property | ||
| def extra_state_attributes(self): | ||
| """Return the state attributes of the device.""" | ||
| attributes = self.entity_description.extra_state_attributes_fn(self.coordinator.data) | ||
|
|
||
| attributes['latest_update'] = self.coordinator.latest_update | ||
| attributes[ATTR_ATTRIBUTION] = ATTRIBUTION | ||
|
|
||
| return attributes |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| """Test binary sensor for DWD rain radar integration.""" | ||
| import os | ||
|
|
||
| import pytest | ||
| from unittest.mock import AsyncMock, patch, MagicMock | ||
|
|
||
| from freezegun import freeze_time | ||
| from pytest_homeassistant_custom_component.common import MockConfigEntry | ||
| from typing_extensions import Generator | ||
|
|
||
| from custom_components.dwd_rain_radar.const import DOMAIN | ||
|
|
||
| # Example binary data to return | ||
| with open(os.path.dirname(__file__) + '/DE1200_RV_LATEST.tar.bz2', 'rb') as f: | ||
| binary_data = f.read() | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def entity_registry_enabled_by_default() -> Generator[None]: | ||
| """Test fixture that ensures all entities are enabled in the registry.""" | ||
| with patch( | ||
| "homeassistant.helpers.entity.Entity.entity_registry_enabled_default", | ||
| return_value=True, | ||
| ): | ||
| yield | ||
|
|
||
| @pytest.mark.asyncio | ||
| @patch('httpx.AsyncClient.get', new_callable=AsyncMock) | ||
| @freeze_time("2024-08-08T15:47:00", tz_offset=2) | ||
| async def test_binary_sensor(mock_get, hass, enable_custom_integrations, entity_registry_enabled_by_default): | ||
| """Test binary sensor.""" | ||
|
|
||
| # Create a mock response object | ||
| mock_response = MagicMock() | ||
| mock_response.status_code = 200 | ||
| mock_response.read = MagicMock(return_value=binary_data) # Mock the read() method | ||
|
|
||
| # Assign the mock response to the get request | ||
| mock_get.return_value = mock_response | ||
|
|
||
| entry = MockConfigEntry(domain=DOMAIN, data={ | ||
| "name": "test dwd", | ||
| "coordinates": { | ||
| "latitude": 48.07530, | ||
| "longitude": 11.32589 | ||
| } | ||
| }) | ||
| entry.add_to_hass(hass) | ||
| await hass.config_entries.async_setup(entry.entry_id) | ||
| await hass.async_block_till_done() | ||
|
|
||
| raining = hass.states.get("binary_sensor.mock_title_raining") | ||
|
|
||
| assert raining | ||
| assert raining.state == 'on' | ||
| assert raining.attributes['prediction_time'].isoformat() == '2024-08-08T17:50:00+02:00' | ||
|
|
||
| raining_in_120_minutes = hass.states.get("binary_sensor.mock_title_raining_in_120_minutes") | ||
| assert raining_in_120_minutes | ||
| assert raining_in_120_minutes.state == 'off' | ||
| assert raining_in_120_minutes.attributes['prediction_time'].isoformat() == '2024-08-08T19:45:00+02:00' | ||
Oops, something went wrong.