Add service alerts sensor

custom_components/nysse/__init__.py

@@ -44,7 +44,7 @@ async def async_unload_entry(
  )
  )
  # Remove options_update_listener.
- # hass.data[DOMAIN][entry.entry_id]["unsub_options_update_listener"]()
+ hass.data[DOMAIN][entry.entry_id]["unsub_options_update_listener"]()
 
  # Remove config entry from domain.
  if unload_ok:

custom_components/nysse/sensor.py

@@ -1,7 +1,7 @@
 """Platform for sensor integration."""
 from __future__ import annotations
 
-from datetime import datetime, timedelta
+from datetime import timedelta
 import json
 import logging
 
@@ -19,11 +19,13 @@
  DEFAULT_MAX,
  DEFAULT_TIMELIMIT,
  DEPARTURE,
+ DOMAIN,
  EXPECTED_ARRIVAL_TIME,
  EXPECTED_DEPARTURE_TIME,
  JOURNEY,
  JOURNEYS_URL,
  PLATFORM_NAME,
+ SERVICE_ALERTS_URL,
  STOP_URL,
  TRAM_LINES,
  WEEKDAYS,
@@ -43,6 +45,11 @@ async def async_setup_entry(
 ) -> None:
  """Setups sensors from a config entry created in the integrations UI."""
  sensors = []
+ configs = hass.data[DOMAIN]
+ if len(configs) > 0:
+ if config_entry.entry_id == next(iter(configs)):
+ sensors.append(ServiceAlertSensor())
+
  if "station" in config_entry.options:
  sensors.append(
  NysseSensor(
@@ -54,7 +61,6 @@ async def async_setup_entry(
  if "timelimit" in config_entry.options
  else DEFAULT_TIMELIMIT,
  config_entry.options["lines"],
- hass.config.time_zone,
  )
  )
  else:
@@ -66,7 +72,6 @@ async def async_setup_entry(
  if "timelimit" in config_entry.data
  else DEFAULT_TIMELIMIT,
  config_entry.data["lines"],
- hass.config.time_zone,
  )
  )
 
@@ -76,7 +81,7 @@ async def async_setup_entry(
 class NysseSensor(SensorEntity):
  """Representation of a Sensor."""
 
- def __init__(self, stop_code, maximum, timelimit, lines, time_zone) -> None:
+ def __init__(self, stop_code, maximum, timelimit, lines) -> None:
  """Initialize the sensor."""
  self._unique_id = PLATFORM_NAME + "_" + stop_code
  self.stop_code = stop_code
@@ -90,7 +95,6 @@ def __init__(self, stop_code, maximum, timelimit, lines, time_zone) -> None:
 
  self._current_weekday_int = -1
  self._last_update_time = None
- self._time_zone = dt_util.get_time_zone(time_zone)
 
  self._fetch_fail_counter = 0
  self._fetch_pause_counter = 0
@@ -319,7 +323,7 @@ def get_departure_time(
 
  async def async_update(self) -> None:
  """Fetch new state data for the sensor."""
- self._last_update_time = datetime.now().astimezone(self._time_zone)
+ self._last_update_time = dt_util.now()
  self._current_weekday_int = self._last_update_time.weekday()
 
  try:
@@ -447,3 +451,104 @@ def extra_state_attributes(self):
  "station_name": self._stops[self.stop_code],
  }
  return attributes
+
+
+class ServiceAlertSensor(SensorEntity):
+ """Representation of a service alert sensor."""
+
+ def __init__(self) -> None:
+ """Initialize the sensor."""
+ self._last_update = ""
+ self._alerts = []
+ self._empty_response_counter = 0
+
+ def timestamp_to_local(self, timestamp):
+ """Convert timestamp to local datetime."""
+ utc = dt_util.utc_from_timestamp(int(str(timestamp)[:10]))
+ return dt_util.as_local(utc)
+
+ def conditionally_clear_alerts(self):
+ """Clear alerts if none received in 20 tries."""
+ if self._empty_response_counter >= 20:
+ self._empty_response_counter = 0
+ self._alerts.clear()
+
+ async def fetch_service_alerts(self):
+ """Fetch service alerts."""
+ try:
+ alerts = []
+ data = await get(SERVICE_ALERTS_URL)
+ if not data:
+ _LOGGER.warning(
+ "Can't fetch service alerts. Incorrect response from %s",
+ SERVICE_ALERTS_URL,
+ )
+ return
+ json_data = json.loads(data)
+
+ self._last_update = self.timestamp_to_local(
+ json_data["header"]["timestamp"]
+ )
+
+ for item in json_data["entity"]:
+ start_time = self.timestamp_to_local(
+ item["alert"]["active_period"][0]["start"]
+ )
+ end_time = self.timestamp_to_local(
+ item["alert"]["active_period"][0]["end"]
+ )
+ description = item["alert"]["description_text"]["translation"][0][
+ "text"
+ ]
+
+ formatted_alert = {
+ "description": description,
+ "start": start_time,
+ "end": end_time,
+ }
+ alerts.append(formatted_alert)
+
+ return alerts
+
+ except KeyError:
+ self._empty_response_counter += 1
+ self.conditionally_clear_alerts()
+ return self._alerts
+ except OSError as err:
+ _LOGGER.error("Failed to update service alerts: %s", err)
+ return []
+
+ async def async_update(self) -> None:
+ """Fetch new state data for the sensor."""
+ self._alerts = await self.fetch_service_alerts()
+
+ @property
+ def unique_id(self) -> str:
+ """Unique id for the sensor."""
+ return "service_alerts"
+
+ @property
+ def name(self) -> str:
+ """Return the name of the sensor."""
+ return "Nysse Service Alerts"
+
+ @property
+ def icon(self) -> str:
+ """Icon of the sensor."""
+ return "mdi:bus-alert"
+
+ @property
+ def state(self) -> str:
+ """Return the state of the sensor."""
+ if len(self._alerts) > 0:
+ return len(self._alerts)
+ return 0
+
+ @property
+ def extra_state_attributes(self):
+ """Sensor attributes."""
+ attributes = {
+ "last_refresh": self._last_update,
+ "alerts": self._alerts,
+ }
+ return attributes