Added AuroraWatch.

https://aurorawatch.lancs.ac.uk/

Author: Steve

homeassistant/components/aurorawatch/__init__.py

@@ -0,0 +1,42 @@
+"""The AuroraWatch UK integration."""
+import logging
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+from .coordinator import AurowatchDataUpdateCoordinator
+
+_LOGGER = logging.getLogger(__name__)
+
+PLATFORMS = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up AuroraWatch UK from a config entry."""
+    coordinator = AurowatchDataUpdateCoordinator(hass)
+
+    # Fetch initial data
+    await coordinator.async_config_entry_first_refresh()
+
+    # Store coordinator
+    hass.data.setdefault(DOMAIN, {})
+    hass.data[DOMAIN][entry.entry_id] = coordinator
+
+    # Forward entry setup to sensor platform
+    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    # Unload platforms
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+
+    # Remove coordinator from hass.data
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

homeassistant/components/aurorawatch/config_flow.py

@@ -0,0 +1,32 @@
+"""Config flow for AuroraWatch UK integration."""
+import logging
+
+from homeassistant import config_entries
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AurowatchConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for AuroraWatch UK."""
+
+    VERSION = 1
+
+    async def async_step_user(self, user_input=None) -> FlowResult:
+        """Handle the initial step."""
+        # Check if already configured
+        await self.async_set_unique_id(DOMAIN)
+        self._abort_if_unique_id_configured()
+
+        if user_input is None:
+            # Show the form to confirm setup
+            return self.async_show_form(step_id="user")
+
+        # Create the config entry
+        return self.async_create_entry(
+            title="AuroraWatch UK",
+            data={},
+        )

homeassistant/components/aurorawatch/const.py

@@ -0,0 +1,20 @@
+"""Constants for the AuroraWatch UK integration."""
+from datetime import timedelta
+
+DOMAIN = "aurorawatch"
+ATTRIBUTION = "Data provided by AuroraWatch UK"
+
+# API Configuration
+API_URL = "http://aurorawatch-api.lancs.ac.uk/0.2/status/current-status.xml"
+API_ACTIVITY_URL = "https://aurorawatch-api.lancs.ac.uk/0.2/status/project/awn/sum-activity.xml"
+API_TIMEOUT = 10  # seconds
+
+# Update Configuration
+UPDATE_INTERVAL = timedelta(minutes=5)
+
+# Sensor Attributes
+ATTR_LAST_UPDATED = "last_updated"
+ATTR_PROJECT_ID = "project_id"
+ATTR_SITE_ID = "site_id"
+ATTR_SITE_URL = "site_url"
+ATTR_API_VERSION = "api_version"

homeassistant/components/aurorawatch/coordinator.py

@@ -0,0 +1,110 @@
+"""Data coordinator for AuroraWatch UK integration."""
+import logging
+import xml.etree.ElementTree as ET
+from datetime import timedelta
+
+import async_timeout
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import API_ACTIVITY_URL, API_TIMEOUT, API_URL, UPDATE_INTERVAL
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class AurowatchDataUpdateCoordinator(DataUpdateCoordinator):
+    """Class to manage fetching AuroraWatch data from API."""
+
+    def __init__(self, hass: HomeAssistant) -> None:
+        """Initialize the coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="AuroraWatch",
+            update_interval=UPDATE_INTERVAL,
+        )
+
+    async def _async_update_data(self):
+        """Fetch data from AuroraWatch API."""
+        try:
+            session = async_get_clientsession(self.hass)
+
+            # Fetch both status and activity data
+            async with async_timeout.timeout(API_TIMEOUT):
+                status_response = await session.get(API_URL)
+                status_response.raise_for_status()
+                status_xml = await status_response.text()
+
+                activity_response = await session.get(API_ACTIVITY_URL)
+                activity_response.raise_for_status()
+                activity_xml = await activity_response.text()
+
+            # Parse status XML
+            try:
+                status_root = ET.fromstring(status_xml)
+            except ET.ParseError as err:
+                _LOGGER.error("Failed to parse status XML response: %s", err)
+                raise UpdateFailed(f"Invalid XML response: {err}") from err
+
+            # Parse activity XML
+            try:
+                activity_root = ET.fromstring(activity_xml)
+            except ET.ParseError as err:
+                _LOGGER.error("Failed to parse activity XML response: %s", err)
+                raise UpdateFailed(f"Invalid activity XML response: {err}") from err
+
+            # Extract data
+            try:
+                # Get updated datetime from status
+                datetime_element = status_root.find('.//updated/datetime')
+                if datetime_element is None or datetime_element.text is None:
+                    raise UpdateFailed("Missing 'updated/datetime' element in XML")
+                last_updated = datetime_element.text
+
+                # Get site status information
+                status_element = status_root.find('.//site_status')
+                if status_element is None:
+                    raise UpdateFailed("Missing 'site_status' element in XML")
+
+                status_id = status_element.get('status_id')
+                if status_id is None:
+                    raise UpdateFailed("Missing 'status_id' attribute in site_status")
+
+                project_id = status_element.get('project_id', 'Unknown')
+                site_id = status_element.get('site_id', 'Unknown')
+                site_url = status_element.get('site_url', '')
+
+                # Get API version
+                api_version = status_root.get('api_version', 'Unknown')
+
+                # Get current activity value (most recent in the list)
+                activity_elements = activity_root.findall('.//activity')
+                activity_value = None
+                if activity_elements:
+                    # Get the last (most recent) activity reading
+                    last_activity = activity_elements[-1]
+                    value_element = last_activity.find('value')
+                    if value_element is not None and value_element.text is not None:
+                        activity_value = float(value_element.text)
+
+                data = {
+                    "status": status_id,
+                    "last_updated": last_updated,
+                    "project_id": project_id,
+                    "site_id": site_id,
+                    "site_url": site_url,
+                    "api_version": api_version,
+                    "activity": activity_value,
+                }
+
+                _LOGGER.debug("Successfully fetched AuroraWatch data: %s", data)
+                return data
+
+            except (AttributeError, KeyError) as err:
+                _LOGGER.error("Failed to extract data from XML: %s", err)
+                raise UpdateFailed(f"Failed to parse XML structure: {err}") from err
+
+        except Exception as err:
+            _LOGGER.error("Error fetching AuroraWatch data: %s", err)
+            raise UpdateFailed(f"Error communicating with API: {err}") from err

homeassistant/components/aurorawatch/entity.py

@@ -0,0 +1,33 @@
+"""The AuroraWatch component."""
+
+from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
+from homeassistant.helpers.update_coordinator import CoordinatorEntity
+
+from .const import ATTRIBUTION, DOMAIN
+from .coordinator import AurowatchDataUpdateCoordinator
+
+
+class AurowatchEntity(CoordinatorEntity[AurowatchDataUpdateCoordinator]):
+    """Implementation of the base AuroraWatch Entity."""
+
+    _attr_attribution = ATTRIBUTION
+    _attr_has_entity_name = True
+
+    def __init__(
+        self,
+        coordinator: AurowatchDataUpdateCoordinator,
+        translation_key: str,
+    ) -> None:
+        """Initialize the AuroraWatch Entity."""
+
+        super().__init__(coordinator=coordinator)
+
+        self._attr_translation_key = translation_key
+        self._attr_unique_id = f"{DOMAIN}_{translation_key}"
+        self._attr_device_info = DeviceInfo(
+            entry_type=DeviceEntryType.SERVICE,
+            identifiers={(DOMAIN, DOMAIN)},
+            manufacturer="Lancaster University",
+            model="AuroraWatch UK",
+            name="AuroraWatch UK",
+        )

homeassistant/components/aurorawatch/icon.png

@@ -0,0 +1,12 @@
+{
+  "entity": {
+    "sensor": {
+      "aurora_status": {
+        "default": "mdi:weather-night"
+      },
+      "geomagnetic_activity": {
+        "default": "mdi:chart-line"
+      }
+    }
+  }
+}

homeassistant/components/aurorawatch/icons.json

@@ -0,0 +1,11 @@
+{
+  "domain": "aurorawatch",
+  "name": "AuroraWatch UK",
+  "codeowners": ["@agentgonzo"],
+  "config_flow": true,
+  "documentation": "https://github.com/aurorawatch/homeassistant-aurorawatch",
+  "integration_type": "service",
+  "iot_class": "cloud_polling",
+  "requirements": [],
+  "version": "1.0.0"
+}

homeassistant/components/aurorawatch/logo.png

@@ -0,0 +1,80 @@
+"""Sensor platform for AuroraWatch UK integration."""
+import logging
+
+from homeassistant.components.sensor import SensorEntity, SensorStateClass
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+
+from .const import (
+    ATTR_API_VERSION,
+    ATTR_LAST_UPDATED,
+    ATTR_PROJECT_ID,
+    ATTR_SITE_ID,
+    ATTR_SITE_URL,
+    DOMAIN,
+)
+from .coordinator import AurowatchDataUpdateCoordinator
+from .entity import AurowatchEntity
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up the AuroraWatch sensor."""
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    async_add_entities([
+        AurowatchSensor(coordinator),
+        AurowatchActivitySensor(coordinator),
+    ])
+
+
+class AurowatchSensor(AurowatchEntity, SensorEntity):
+    """Representation of an AuroraWatch status sensor."""
+
+    def __init__(self, coordinator: AurowatchDataUpdateCoordinator) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator, "aurora_status")
+
+    @property
+    def native_value(self):
+        """Return the state of the sensor."""
+        if self.coordinator.data:
+            return self.coordinator.data.get("status")
+        return None
+
+    @property
+    def extra_state_attributes(self):
+        """Return the state attributes."""
+        if not self.coordinator.data:
+            return {}
+
+        return {
+            ATTR_LAST_UPDATED: self.coordinator.data.get("last_updated"),
+            ATTR_PROJECT_ID: self.coordinator.data.get("project_id"),
+            ATTR_SITE_ID: self.coordinator.data.get("site_id"),
+            ATTR_SITE_URL: self.coordinator.data.get("site_url"),
+            ATTR_API_VERSION: self.coordinator.data.get("api_version"),
+        }
+
+
+class AurowatchActivitySensor(AurowatchEntity, SensorEntity):
+    """Representation of an AuroraWatch geomagnetic activity sensor."""
+
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_native_unit_of_measurement = "nT"
+
+    def __init__(self, coordinator: AurowatchDataUpdateCoordinator) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator, "geomagnetic_activity")
+
+    @property
+    def native_value(self):
+        """Return the geomagnetic activity value."""
+        if self.coordinator.data:
+            return self.coordinator.data.get("activity")
+        return None