Hi Everybody,
I created a custom sensor which lets me know if I can go to work by bike by analyzing the rain, temperature and pollution on my route. It generate a score for today. (It relies on the open meteo API)
I would like to see the index for today and tomorrow.
I need to produce this sensor in 2 versions: one for today and one for tomorrow.
How to create 2 sensors with date as parameter?
file : init.py
"""Example Load Platform integration."""
from __future__ import annotations
from homeassistant.core import (
HomeAssistant,
ServiceCall,
ServiceResponse,
SupportsResponse,
)
from homeassistant.helpers.typing import ConfigType
from datetime import datetime, timedelta
import logging
from urllib.request import urlopen
import json
DOMAIN = "ready_to_bike"
# seuils
DATE_OF_THE_DAY = datetime.now().strftime("%Y-%m-%d")
APPARENT_TEMPERATURE_THRESHOLD_MORNING = 23
APPARENT_TEMPERATURE_THRESHOLD_EVENING = 27
PRECIPITATION_PROBABILITY = 10
PRECIPITATION = 0.1
EUROPEAN_AQI = 41
US_AQI = 51
DEFAULT_CONF_UPDATE_INTERVAL = 1
HOUR = [7, 8, 17, 18]
CITIES = {}
CITIES["Bry-sur-marne"] = [48.8381, 2.5249]
CITIES["Nogent-sur-marne"] = [48.8367, 2.4825]
CITIES["Vincennes"] = [48.8486, 2.4377]
CITIES["Paris"] = [48.8534, 2.3488]
_LOGGER = logging.getLogger(__name__)
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Your controller/hub specific code."""
_LOGGER.info("Setup Ready To Bike ")
date_of_the_day = datetime.now().strftime("%Y-%m-%d")
# Data that you want to share with your platforms
hass.data[DOMAIN] = {}
async def call_open_meteo_api(call: ServiceCall) -> ServiceResponse:
return_of_call = await async_call_open_meteo()
return {"return": return_of_call}
async def async_call_open_meteo():
_LOGGER.info("Call API ")
date_of_the_day = datetime.now().strftime("%Y-%m-%d")
hass.data[DOMAIN]["state"] = []
data = {}
for hour in HOUR:
data[hour] = {"hour": 0, "super_indice": True, "data": {}}
for city in CITIES:
data[hour]["data"][city] = {}
super_indice = {}
for hour_index in range(25):
super_indice[hour_index] = 1
for city in CITIES:
_LOGGER.info("Loop " + city)
meteo_result = await fetch_weather_data(hass, date_of_the_day, CITIES[city])
air_result = await fetch_air_quality_data(
hass, date_of_the_day, CITIES[city]
)
for hour in HOUR:
array_of_data = []
# calcul des seuils
if hour < 10:
apparent_temperature = (
meteo_result["apparent_temperature"][hour]
< APPARENT_TEMPERATURE_THRESHOLD_MORNING
)
else:
apparent_temperature = (
meteo_result["apparent_temperature"][hour]
< APPARENT_TEMPERATURE_THRESHOLD_EVENING
)
precipitation_probability = (
meteo_result["precipitation_probability"][hour]
< PRECIPITATION_PROBABILITY
)
precipitation = meteo_result["precipitation"][hour] < PRECIPITATION
european_aqi = air_result["european_aqi"][hour] < EUROPEAN_AQI
us_aqi = air_result["us_aqi"][hour] < US_AQI
# Calcul de l'indice en fonction des conditions spécifiées
indice = (
apparent_temperature
and precipitation_probability
and precipitation
and european_aqi
and us_aqi
)
# Calcul du super indice
super_indice[hour] = indice and super_indice[hour]
hass.data[DOMAIN]["state"].append(super_indice[hour])
data[hour]["hour"] = hour
data[hour]["super_indice"] = super_indice[hour]
data[hour]["data"][city]["indice"] = indice
data[hour]["data"][city]["city"] = city
data[hour]["data"][city]["apparent_temperature"] = meteo_result[
"apparent_temperature"
][hour]
data[hour]["data"][city]["precipitation_probability"] = meteo_result[
"precipitation_probability"
][hour]
data[hour]["data"][city]["precipitation"] = meteo_result[
"precipitation"
][hour]
data[hour]["data"][city]["european_aqi"] = air_result["european_aqi"][
hour
]
data[hour]["data"][city]["us_aqi"] = air_result["us_aqi"][hour]
go_or_not = (data[7]["super_indice"] or data[8]["super_indice"]) and (
data[17]["super_indice"] or data[18]["super_indice"]
)
score = 0
_LOGGER.info("Populate Ready To Bike Sensor")
hass.data[DOMAIN] = {
"score": score,
"indice": go_or_not,
"day": date_of_the_day,
"data": data,
}
_LOGGER.info("End of function call_open_meteo")
return True
# initialize sensor's variables
_LOGGER.info("Launch async_call_open_meteo function")
await async_call_open_meteo()
hass.helpers.discovery.load_platform("sensor", DOMAIN, {}, config)
hass.services.async_register(
DOMAIN,
"call_open_meteo_bike_api",
call_open_meteo_api,
# schema=SEARCH_ITEMS_SCHEMA,
supports_response=SupportsResponse.ONLY,
)
_LOGGER.info("Launch async_call_open_meteo function")
return True
async def fetch_weather_data(hass, date_of_the_day, city):
_LOGGER.info("Call fetch_weather_data " + str(city[0]) + " et " + str(city[1]))
httpAnswer = await hass.async_add_executor_job(
urlopen,
"https://api.open-meteo.com/v1/forecast?latitude=%s&longitude=%s&hourly=temperature_2m,apparent_temperature,precipitation_probability,precipitation,rain,showers,snowfall&timezone=Europe/Berlin&start_date=%s&end_date=%s"
% (city[0], city[1], date_of_the_day, date_of_the_day),
)
result = json.loads(httpAnswer.read())["hourly"]
if result:
return result
else:
raise Exception("Erreur lors de la requête météorologique")
async def fetch_air_quality_data(hass, date_of_the_day, city):
httpAnswer = await hass.async_add_executor_job(
urlopen,
"https://air-quality-api.open-meteo.com/v1/air-quality?latitude=%s&longitude=%s&hourly=alder_pollen,birch_pollen,grass_pollen,mugwort_pollen,olive_pollen,ragweed_pollen,european_aqi,european_aqi_pm2_5,european_aqi_pm10,european_aqi_no2,european_aqi_o3,european_aqi_so2,us_aqi&timezone=Europe/Berlin&start_date=%s&end_date=%s&domains=cams_europe"
% (city[0], city[1], date_of_the_day, date_of_the_day),
)
result = json.loads(httpAnswer.read())["hourly"]
if result:
return result
else:
raise Exception("Erreur lors de la requête air qualité")
File: sensor.py
"""Platform for sensor integration."""
from __future__ import annotations
from homeassistant.components.sensor import SensorEntity
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import DOMAIN
import logging
_LOGGER = logging.getLogger(__name__)
def setup_platform(
hass: HomeAssistant,
config: ConfigType,
add_entities: AddEntitiesCallback,
discovery_info: DiscoveryInfoType | None = None,
) -> None:
"""Set up the sensor platform."""
# We only want this platform to be set up via discovery.
if discovery_info is None:
return
_LOGGER.info("Create Ready To Bike Sensor Platform ")
_LOGGER.debug(f"Config : {config} ")
add_entities([ReadyToBike(config)])
class ReadyToBike(SensorEntity):
"""Representation of a sensor."""
def __init__(self, data) -> None:
"""Initialize the sensor."""
self._state = None
self._attr_extra_state_attributes = {}
@property
def name(self) -> str:
"""Return the name of the sensor."""
return "Ready To Bike Sensor"
@property
def state(self):
"""Return the state of the sensor."""
return self._state
def update(self) -> None:
"""Fetch new state data for the sensor.
This is the only method that should fetch new data for Home Assistant.
"""
_LOGGER.info("Update Ready To Bike Sensor ")
self._state = self.hass.data[DOMAIN].get("score", 0)
self._attr_extra_state_attributes["day"] = self.hass.data[DOMAIN]["day"]
self._attr_extra_state_attributes["indice"] = self.hass.data[DOMAIN]["indice"]
self._attr_extra_state_attributes["data"] = self.hass.data[DOMAIN]["data"]