-
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #20 from PiotrMachowski/dev
v2.0.0
- Loading branch information
Showing
13 changed files
with
489 additions
and
214 deletions.
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
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 |
|---|---|---|
| @@ -1 +1,60 @@ | ||
| """Antistorm""" | ||
| import logging | ||
|
|
||
| import homeassistant.helpers.config_validation as cv | ||
| import voluptuous as vol | ||
| from homeassistant.components.sensor import PLATFORM_SCHEMA | ||
| from homeassistant.config_entries import ConfigEntry | ||
| from homeassistant.const import CONF_NAME | ||
| from homeassistant.core import HomeAssistant | ||
| from homeassistant.exceptions import ConfigEntryNotReady | ||
|
|
||
| from .const import ( | ||
| DOMAIN, PLATFORMS, CONF_CITY_ID | ||
| ) | ||
| from .update_coordinator import AntistormUpdateCoordinator | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
| PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend( | ||
| { | ||
| vol.Required(CONF_CITY_ID): cv.positive_int, | ||
| vol.Required(CONF_NAME): cv.string, | ||
| } | ||
| ) | ||
|
|
||
|
|
||
| async def async_setup(_hass, _config): | ||
| return True | ||
|
|
||
|
|
||
| async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry): | ||
| if hass.data.get(DOMAIN) is None: | ||
| hass.data.setdefault(DOMAIN, {}) | ||
|
|
||
| city_id = config_entry.data.get(CONF_CITY_ID) | ||
|
|
||
| coordinator = AntistormUpdateCoordinator(hass, city_id) | ||
| await coordinator.async_refresh() | ||
|
|
||
| if not coordinator.last_update_success: | ||
| raise ConfigEntryNotReady | ||
|
|
||
| hass.data[DOMAIN][config_entry.entry_id] = coordinator | ||
|
|
||
| await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS) | ||
| config_entry.async_on_unload(config_entry.add_update_listener(async_reload_entry)) | ||
| return True | ||
|
|
||
|
|
||
| async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry): | ||
| """Unload a config entry.""" | ||
| unloaded = await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS) | ||
| if unloaded: | ||
| hass.data[DOMAIN].pop(config_entry.entry_id) | ||
| return unloaded | ||
|
|
||
|
|
||
| async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: | ||
| """Reload config entry.""" | ||
| await async_unload_entry(hass, entry) | ||
| await async_setup_entry(hass, entry) |
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 |
|---|---|---|
| @@ -1,89 +1,80 @@ | ||
| import logging | ||
| import requests | ||
| from dataclasses import dataclass | ||
| from typing import Callable | ||
|
|
||
| import voluptuous as vol | ||
| from homeassistant.config_entries import ConfigEntry | ||
|
|
||
| from homeassistant.components.binary_sensor import DEVICE_CLASS_SAFETY, PLATFORM_SCHEMA, ENTITY_ID_FORMAT | ||
| from homeassistant.const import CONF_MONITORED_CONDITIONS, CONF_NAME, ATTR_ATTRIBUTION | ||
| import homeassistant.helpers.config_validation as cv | ||
| try: | ||
| from homeassistant.components.binary_sensor import BinarySensorEntity | ||
| except ImportError: | ||
| from homeassistant.components.binary_sensor import BinarySensorDevice as BinarySensorEntity | ||
| from homeassistant.helpers.entity import async_generate_entity_id | ||
| from homeassistant.core import HomeAssistant | ||
| from homeassistant.components.binary_sensor import BinarySensorDeviceClass, BinarySensorEntityDescription, DOMAIN as BS_DOMAIN | ||
| from homeassistant.components.binary_sensor import BinarySensorEntity | ||
| from homeassistant.helpers.entity_platform import AddEntitiesCallback | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
| CONF_STATION_ID = 'station_id' | ||
|
|
||
| DEFAULT_NAME = 'Antistorm' | ||
| ATTRIBUTION = 'Information provided by Antistorm.eu.' | ||
|
|
||
| SENSOR_TYPES = { | ||
| 'storm_alarm': ['a_b', 'Alarm burzowy', 'mdi:weather-lightning'], | ||
| 'rain_alarm': ['a_o', 'Alarm opadów', 'mdi:weather-pouring'], | ||
| 'storm_active': ['s', 'Aktywna burza', 'mdi:weather-lightning-rainy'], | ||
| } | ||
|
|
||
| PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ | ||
| vol.Required(CONF_NAME, default=DEFAULT_NAME): cv.string, | ||
| vol.Required(CONF_STATION_ID): cv.string, | ||
| vol.Required(CONF_MONITORED_CONDITIONS, default=[]): | ||
| vol.All(cv.ensure_list, [vol.In(SENSOR_TYPES)]) | ||
| }) | ||
|
|
||
|
|
||
| def setup_platform(hass, config, add_entities, discovery_info=None): | ||
| station_id = config.get(CONF_STATION_ID) | ||
| name = config.get(CONF_NAME) | ||
| address = 'http://antistorm.eu/webservice.php?id=' + str(station_id) | ||
| request = requests.get(address) | ||
| request.encoding = 'utf-8' | ||
| city_name = request.json()['m'] | ||
| dev = [] | ||
| for monitored_condition in config[CONF_MONITORED_CONDITIONS]: | ||
| uid = '{}_{}_{}'.format(name, station_id, monitored_condition) | ||
| entity_id = async_generate_entity_id(ENTITY_ID_FORMAT, uid, hass=hass) | ||
| dev.append(AntistormBinarySensor(entity_id, name, city_name, monitored_condition, station_id)) | ||
| add_entities(dev, True) | ||
|
|
||
|
|
||
| class AntistormBinarySensor(BinarySensorEntity): | ||
| def __init__(self, entity_id, name, city_name, sensor_type, station_id): | ||
| self.entity_id = entity_id | ||
| self._name = name | ||
| self.city_name = city_name | ||
| self.station_id = station_id | ||
| self.sensor_type = sensor_type | ||
| self.data = None | ||
| self._state = None | ||
| self._jsonParameter = SENSOR_TYPES[sensor_type][0] | ||
| self._name = SENSOR_TYPES[sensor_type][1] | ||
| from .const import DOMAIN | ||
| from .update_coordinator import AntistormUpdateCoordinator, AntistormData | ||
| from .entity import AntistormEntity | ||
|
|
||
| @property | ||
| def extra_state_attributes(self): | ||
| output = dict() | ||
| output[ATTR_ATTRIBUTION] = ATTRIBUTION | ||
| return output | ||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
| @property | ||
| def name(self): | ||
| return '{} {} - {}'.format(self._name, self.city_name, SENSOR_TYPES[self.sensor_type][1]) | ||
|
|
||
| @property | ||
| def icon(self): | ||
| return SENSOR_TYPES[self.sensor_type][2] | ||
| @dataclass(frozen=True) | ||
| class AntistormBinarySensorDescriptionMixin: | ||
| value_fn: Callable[[AntistormData], bool] | ||
|
|
||
|
|
||
| @dataclass(frozen=True) | ||
| class AntistormBinarySensorEntityDescription(BinarySensorEntityDescription, AntistormBinarySensorDescriptionMixin): | ||
| device_class = BinarySensorDeviceClass.SAFETY | ||
| has_entity_name = True | ||
|
|
||
|
|
||
| entity_descriptions = [ | ||
| AntistormBinarySensorEntityDescription( | ||
| key='storm_alarm', | ||
| translation_key='storm_alarm', | ||
| icon="mdi:weather-lightning", | ||
| value_fn=lambda data: data.storm_alarm, | ||
| ), | ||
| AntistormBinarySensorEntityDescription( | ||
| key='precipitation_alarm', | ||
| translation_key='precipitation_alarm', | ||
| icon="mdi:weather-pouring", | ||
| value_fn=lambda data: data.precipitation_alarm, | ||
| ), | ||
| AntistormBinarySensorEntityDescription( | ||
| key='storm_active', | ||
| translation_key='storm_active', | ||
| icon="mdi:weather-lightning-rainy", | ||
| value_fn=lambda data: data.storm_active, | ||
| ), | ||
| ] | ||
|
|
||
|
|
||
| async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, | ||
| async_add_entities: AddEntitiesCallback) -> bool: | ||
| coordinator: AntistormUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id] | ||
| entities = [] | ||
| for entity_description in entity_descriptions: | ||
| entities.append(AntistormBinarySensor(coordinator, config_entry, entity_description)) | ||
| async_add_entities(entities) | ||
| return True | ||
|
|
||
|
|
||
| class AntistormBinarySensor(AntistormEntity, BinarySensorEntity): | ||
| entity_description: AntistormBinarySensorEntityDescription | ||
|
|
||
| def __init__(self, coordinator: AntistormUpdateCoordinator, config_entry: ConfigEntry, | ||
| description: AntistormBinarySensorEntityDescription) -> None: | ||
| super().__init__(coordinator, config_entry) | ||
| self.entity_description = description | ||
| self._attr_unique_id = f"{DOMAIN}_{description.key}" | ||
| self.entity_id = f"{BS_DOMAIN}.{DOMAIN}_{self.city_id}_{description.key}" | ||
|
|
||
| @property | ||
| def is_on(self): | ||
| return self.data is not None and int(self.data[self._jsonParameter]) > 0 | ||
| def is_on(self) -> bool | None: | ||
| if self.get_data() is None: | ||
| return None | ||
| return self.entity_description.value_fn(self.get_data()) | ||
|
|
||
| @property | ||
| def device_class(self): | ||
| return DEVICE_CLASS_SAFETY | ||
|
|
||
| def update(self): | ||
| address = 'http://antistorm.eu/webservice.php?id=' + str(self.station_id) | ||
| request = requests.get(address) | ||
| if request.status_code == 200 and request.content.__len__() > 0: | ||
| self.data = request.json() | ||
| def unique_id(self) -> str: | ||
| return f"{super().unique_id}_{self.entity_description.key}" |
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,47 @@ | ||
| """Config flow to configure Antistorm integration.""" | ||
|
|
||
| import logging | ||
|
|
||
| import voluptuous as vol | ||
| from homeassistant import config_entries | ||
| from homeassistant.const import CONF_NAME | ||
| from homeassistant.data_entry_flow import FlowResult | ||
|
|
||
| from .connector import AntistormConnector | ||
| from .const import CONF_CITY_ID, CONF_CITY_NAME, DOMAIN | ||
|
|
||
| _LOGGER = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| class AntistormFlowHandler(config_entries.ConfigFlow, domain=DOMAIN): | ||
| VERSION = 1 | ||
|
|
||
| async def get_city_details(self, city_name: str) -> tuple[int, str] | None: | ||
| city_id = await self.hass.async_add_executor_job(AntistormConnector.get_city_id, city_name) | ||
| if city_id is None: | ||
| return None | ||
| city_name = (await self.hass.async_add_executor_job(lambda: AntistormConnector(city_id).get_data())).city | ||
| return city_id, city_name | ||
|
|
||
| async def async_step_user(self, user_input=None) -> FlowResult: | ||
| errors = {} | ||
| usr_city_name = "" | ||
| if user_input is not None: | ||
| usr_city_name = user_input[CONF_CITY_NAME].strip() | ||
| details = await self.get_city_details(usr_city_name) | ||
| if details is not None: | ||
| city_id = details[0] | ||
| city_name = details[1] | ||
| return self.async_create_entry( | ||
| title=city_name, | ||
| data={ | ||
| CONF_CITY_ID: city_id, | ||
| CONF_NAME: city_name | ||
| }, | ||
| ) | ||
| else: | ||
| errors[CONF_CITY_NAME] = "city_not_found" | ||
| schema = vol.Schema({ | ||
| vol.Required(CONF_CITY_NAME, default=usr_city_name): str, | ||
| }) | ||
| return self.async_show_form(step_id="user", data_schema=schema, errors=errors) |
Oops, something went wrong.