from typing import Any
import html
from .label_matcher import ActionLabelMatcher
from .nina_api import NinaAPI
from .warning import Warning
class WarningFetcher:
"""Class to fetch warnings for a region."""
def __init__(self, api: NinaAPI, label_matcher: ActionLabelMatcher):
"""Initialize."""
self._api = api
self._label_matcher = label_matcher
async def fetch_warnings(self, region_code: str) -> list[Warning]:
"""Fetch warnings for the given region."""
warnings: list[Warning] = []
raw_warnings = await self._api.get_warnings(region_code)
for warning_data in raw_warnings:
warning_details = await self._api.get_warning_details(
warning_data["payload"]["id"]
)
warnings.append(self._build_from_data(warning_data, warning_details))
return warnings
def _build_from_data(
self, warning_data: dict[str, Any], warning_details: dict[str, Any]
) -> Warning:
"""Build a Warning object from the data."""
infos = warning_details["info"][0]
warning_id: str = warning_data["payload"]["id"]
headline: str = warning_data["payload"]["data"]["headline"]
severity: str = warning_data["payload"]["data"]["severity"]
description: str | None = (
html.unescape(infos.get("description", ""))
if "description" in infos
else None
)
sender: str | None = infos.get("senderName", None)
areas: list[str] = [area["areaDesc"] for area in infos.get("area", [])]
web: str | None = infos.get("web", None)
time_send: str = warning_data["sent"]
time_start: str | None = warning_data.get(
"effective", warning_data.get("onset", None)
)
time_expire: str | None = warning_data.get("expires", None)
recommended_actions: list[str] = (
[infos.get("instruction", "").replace("
", " ")]
if "instruction" in infos
else []
)
if len(recommended_actions) == 0:
recommended_actions_codes: list[str] = [
parameter["value"].split(" ")
for parameter in infos.get("parameter", {})
if parameter["valueName"] == "instructionCode"
]
recommended_actions = [
self._label_matcher.match(code) for code in recommended_actions_codes
]
return Warning(
warning_id,
headline,
severity,
description,
sender,
areas,
recommended_actions,
web,
time_send,
time_start,
time_expire,
)