starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 09:23:17 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add precipitation forecast service to Environment Canada (#182135)

Browse Source 
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Michael Davie
2026-09-19 21:14:25 +02:00
committed by GitHub
co-authored by Claude Sonnet 5
parent 02553e9779
commit 920e59e65a
6 changed files with 307 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/environment_canada/icons.json
+3
View File
@@ -25,6 +25,9 @@
"get_forecasts": {
"service": "mdi:weather-cloudy-clock"
},
"get_precipitation_forecast": {
"service": "mdi:weather-pouring"
},
"set_radar_type": {
"service": "mdi:radar"
}
homeassistant/components/environment_canada/services.py
+65 -2
View File
@@ -2,10 +2,15 @@
from typing import Any
from env_canada import ECWeather
from env_canada import ECPrecipForecast, ECWeather
import probatio
from homeassistant.const import ATTR_CONFIG_ENTRY_ID
from homeassistant.const import (
ATTR_CONFIG_ENTRY_ID,
CONF_LANGUAGE,
CONF_LATITUDE,
CONF_LONGITUDE,
)
from homeassistant.core import HomeAssistant, ServiceCall, SupportsResponse, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import config_validation as cv, service
@@ -17,11 +22,29 @@ SERVICE_GET_ALERTS_SCHEMA = probatio.Schema(
{probatio.Required(ATTR_CONFIG_ENTRY_ID): cv.string}
)
SERVICE_GET_PRECIPITATION_FORECAST = "get_precipitation_forecast"
SERVICE_GET_PRECIPITATION_FORECAST_SCHEMA = probatio.Schema(
{
probatio.Required(ATTR_CONFIG_ENTRY_ID): cv.string,
probatio.Optional("precip_type"): probatio.In(["auto", "rain", "snow"]),
probatio.Optional("past_minutes"): probatio.All(int, probatio.Range(0, 180)),
probatio.Optional("future_minutes"): probatio.All(int, probatio.Range(0, 72)),
probatio.Optional("hourly_hours"): probatio.All(int, probatio.Range(0, 48)),
}
)
SNAKE_MAPPING = {
"alertColourLevel": "alert_colour_level",
"expiryTime": "expiry_time",
}
PRECIP_FORECAST_OPTIONS = (
"precip_type",
"past_minutes",
"future_minutes",
"hourly_hours",
)
async def _async_get_alerts(call: ServiceCall) -> dict[str, Any]:
"""Return the active alerts."""
@@ -46,6 +69,39 @@ async def _async_get_alerts(call: ServiceCall) -> dict[str, Any]:
}
async def _async_get_precipitation_forecast(call: ServiceCall) -> dict[str, Any]:
"""Return the precipitation forecast series."""
entry = service.async_get_config_entry(
call.hass, DOMAIN, call.data[ATTR_CONFIG_ENTRY_ID]
)
# A fresh object per call, rather than one shared across calls: its
# options would otherwise leak between calls that omit them, and
# concurrent calls could race on the same instance's attributes.
kwargs: dict[str, Any] = {
"coordinates": (entry.data[CONF_LATITUDE], entry.data[CONF_LONGITUDE]),
"language": entry.data.get(CONF_LANGUAGE, "English").lower(),
}
for option in PRECIP_FORECAST_OPTIONS:
if option in call.data:
kwargs[option] = call.data[option]
precip = ECPrecipForecast(**kwargs)
await precip.update()
return {
"nowcast": [
{**item, "timestamp": item["timestamp"].isoformat()}
for item in precip.nowcast
],
"hourly": [
{**item, "timestamp": item["timestamp"].isoformat()}
for item in precip.hourly
],
"metadata": precip.metadata,
}
@callback
def async_setup_services(hass: HomeAssistant) -> None:
"""Set up the services for the Environment Canada integration."""
@@ -56,3 +112,10 @@ def async_setup_services(hass: HomeAssistant) -> None:
schema=SERVICE_GET_ALERTS_SCHEMA,
supports_response=SupportsResponse.ONLY,
)
hass.services.async_register(
DOMAIN,
SERVICE_GET_PRECIPITATION_FORECAST,
_async_get_precipitation_forecast,
schema=SERVICE_GET_PRECIPITATION_FORECAST_SCHEMA,
supports_response=SupportsResponse.ONLY,
)
homeassistant/components/environment_canada/services.yaml
+33
View File
@@ -12,6 +12,39 @@ get_forecasts:
integration: environment_canada
domain: weather
get_precipitation_forecast:
fields:
config_entry_id:
required: true
selector:
config_entry:
integration: environment_canada
precip_type:
selector:
select:
options:
- "auto"
- "rain"
- "snow"
past_minutes:
selector:
number:
min: 0
max: 180
unit_of_measurement: min
future_minutes:
selector:
number:
min: 0
max: 72
unit_of_measurement: min
hourly_hours:
selector:
number:
min: 0
max: 48
unit_of_measurement: h
set_radar_type:
target:
entity:
homeassistant/components/environment_canada/strings.json
+26
View File
@@ -201,6 +201,32 @@
"description": "Retrieves the forecast from selected weather services.",
"name": "Get forecasts"
},
"get_precipitation_forecast": {
"description": "Retrieves the precipitation forecast series from the selected weather service.",
"fields": {
"config_entry_id": {
"description": "The Environment Canada service to retrieve the precipitation forecast from.",
"name": "Environment Canada service"
},
"future_minutes": {
"description": "How far ahead the short-interval series reaches.",
"name": "Future minutes"
},
"hourly_hours": {
"description": "Length of the hourly series. Zero disables it.",
"name": "Hourly hours"
},
"past_minutes": {
"description": "How far back the short-interval series reaches.",
"name": "Past minutes"
},
"precip_type": {
"description": "Which precipitation type to report. Automatically detects it from radar.",
"name": "Precipitation type"
}
},
"name": "Get precipitation forecast"
},
"set_radar_type": {
"description": "Sets the type of radar image to retrieve.",
"fields": {
tests/components/environment_canada/snapshots/test_services.ambr
+29
View File
@@ -37,3 +37,32 @@
]),
})
# ---
# name: test_get_precipitation_forecast
dict({
'hourly': list([
dict({
'amount': 1.726,
'conditional_amount': 0.909,
'expected_amount': 0.518,
'label': '0.5 - 1 mm',
'precip_type': 'Rain',
'probability': 57,
'timestamp': '2022-10-04T13:00:00+00:00',
}),
]),
'metadata': dict({
'attribution': 'Data provided by Environment Canada',
'timestamp': '2022-10-04T12:00:00+00:00',
}),
'nowcast': list([
dict({
'forecast': False,
'label': '1.0 - 2.0 (mm/h)',
'precip_type': 'rain',
'rate': 1.2391,
'timestamp': '2022-10-04T12:00:00+00:00',
'unit': 'mm/h',
}),
]),
})
# ---
tests/components/environment_canada/test_services.py
+151 -1
View File
@@ -1,17 +1,54 @@
"""Tests for the Environment Canada services."""
from datetime import UTC, datetime
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import probatio
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.environment_canada.const import DOMAIN
from homeassistant.const import CONF_LANGUAGE, CONF_LATITUDE, CONF_LONGITUDE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from . import init_integration
from . import FIXTURE_USER_INPUT, init_integration
SERVICE_GET_ALERTS = "get_alerts"
SERVICE_GET_PRECIPITATION_FORECAST = "get_precipitation_forecast"
def _precip_mock() -> MagicMock:
"""Build an ECPrecipForecast constructor mock returning sample data."""
instance = MagicMock()
instance.update = AsyncMock()
instance.nowcast = [
{
"timestamp": datetime(2022, 10, 4, 12, 0, tzinfo=UTC),
"rate": 1.2391,
"unit": "mm/h",
"label": "1.0 - 2.0 (mm/h)",
"precip_type": "rain",
"forecast": False,
}
]
instance.hourly = [
{
"timestamp": datetime(2022, 10, 4, 13, 0, tzinfo=UTC),
"amount": 1.726,
"probability": 57,
"conditional_amount": 0.909,
"expected_amount": 0.518,
"precip_type": "Rain",
"label": "0.5 - 1 mm",
}
]
instance.metadata = {
"attribution": "Data provided by Environment Canada",
"timestamp": "2022-10-04T12:00:00+00:00",
}
return MagicMock(return_value=instance)
async def test_get_alerts(
@@ -45,3 +82,116 @@ async def test_get_alerts_not_connected(
blocking=True,
return_response=True,
)
async def test_get_precipitation_forecast(
hass: HomeAssistant, snapshot: SnapshotAssertion, ec_data: dict[str, Any]
) -> None:
"""Test the get_precipitation_forecast service returns the series."""
config_entry = await init_integration(hass, ec_data)
constructor = _precip_mock()
with patch(
"homeassistant.components.environment_canada.services.ECPrecipForecast",
constructor,
):
response = await hass.services.async_call(
DOMAIN,
SERVICE_GET_PRECIPITATION_FORECAST,
{"config_entry_id": config_entry.entry_id},
blocking=True,
return_response=True,
)
assert response == snapshot
constructor.assert_called_once_with(
coordinates=(
FIXTURE_USER_INPUT[CONF_LATITUDE],
FIXTURE_USER_INPUT[CONF_LONGITUDE],
),
language=FIXTURE_USER_INPUT[CONF_LANGUAGE].lower(),
)
constructor.return_value.update.assert_awaited_once()
async def test_get_precipitation_forecast_options(
hass: HomeAssistant, ec_data: dict[str, Any]
) -> None:
"""Test the get_precipitation_forecast service passes requested options."""
config_entry = await init_integration(hass, ec_data)
constructor = _precip_mock()
with patch(
"homeassistant.components.environment_canada.services.ECPrecipForecast",
constructor,
):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_PRECIPITATION_FORECAST,
{
"config_entry_id": config_entry.entry_id,
"precip_type": "snow",
"past_minutes": 30,
"future_minutes": 15,
"hourly_hours": 6,
},
blocking=True,
return_response=True,
)
constructor.assert_called_once_with(
coordinates=(
FIXTURE_USER_INPUT[CONF_LATITUDE],
FIXTURE_USER_INPUT[CONF_LONGITUDE],
),
language=FIXTURE_USER_INPUT[CONF_LANGUAGE].lower(),
precip_type="snow",
past_minutes=30,
future_minutes=15,
hourly_hours=6,
)
async def test_get_precipitation_forecast_options_do_not_leak(
hass: HomeAssistant, ec_data: dict[str, Any]
) -> None:
"""Test that options from one call are not carried over to the next."""
config_entry = await init_integration(hass, ec_data)
constructor = _precip_mock()
with patch(
"homeassistant.components.environment_canada.services.ECPrecipForecast",
constructor,
):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_PRECIPITATION_FORECAST,
{"config_entry_id": config_entry.entry_id, "precip_type": "snow"},
blocking=True,
return_response=True,
)
await hass.services.async_call(
DOMAIN,
SERVICE_GET_PRECIPITATION_FORECAST,
{"config_entry_id": config_entry.entry_id},
blocking=True,
return_response=True,
)
assert "precip_type" not in constructor.call_args_list[1].kwargs
async def test_get_precipitation_forecast_invalid_option(
hass: HomeAssistant, ec_data: dict[str, Any]
) -> None:
"""Test the get_precipitation_forecast service rejects out-of-range options."""
config_entry = await init_integration(hass, ec_data)
with pytest.raises(probatio.MultipleInvalid):
await hass.services.async_call(
DOMAIN,
SERVICE_GET_PRECIPITATION_FORECAST,
{"config_entry_id": config_entry.entry_id, "hourly_hours": 100},
blocking=True,
return_response=True,
)