first commit

custom_components/strike/sensor.py

@@ -0,0 +1,181 @@
+"""Support for Strike Bitcoin price sensor."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta
+import logging
+from typing import Any
+
+import aiohttp
+
+from homeassistant.components.sensor import (
+    SensorDeviceClass,
+    SensorEntity,
+    SensorStateClass,
+)
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_SCAN_INTERVAL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.update_coordinator import (
+    CoordinatorEntity,
+    DataUpdateCoordinator,
+    UpdateFailed,
+)
+
+from .const import (
+    API_BASE_URL,
+    API_TIMEOUT,
+    ATTR_CURRENCY,
+    ATTR_LAST_UPDATE,
+    ATTR_SOURCE,
+    CONF_API_KEY,
+    CONF_CURRENCY_PAIR,
+    DEFAULT_CURRENCY_PAIR,
+    DEFAULT_SCAN_INTERVAL,
+    DOMAIN,
+    MANUFACTURER,
+    MODEL,
+    TICKER_ENDPOINT,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: ConfigEntry,
+    async_add_entities: AddEntitiesCallback,
+) -> None:
+    """Set up Strike Bitcoin sensor based on a config entry."""
+    coordinator = StrikeBitcoinCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+
+    async_add_entities([StrikeBitcoinSensor(coordinator, entry)])
+
+
+class StrikeBitcoinCoordinator(DataUpdateCoordinator):
+    """Class to manage fetching Strike Bitcoin data."""
+
+    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
+        """Initialize the coordinator."""
+        self.api_key = entry.data[CONF_API_KEY]
+        self.currency_pair = entry.data.get(CONF_CURRENCY_PAIR, DEFAULT_CURRENCY_PAIR)
+        self.session = hass.data[DOMAIN][entry.entry_id]["session"]
+
+        scan_interval = entry.data.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL)
+
+        super().__init__(
+            hass,
+            _LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(seconds=scan_interval),
+        )
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        """Fetch data from Strike API."""
+        headers = {
+            "Accept": "application/json",
+            "Authorization": f"Bearer {self.api_key}",
+        }
+
+        # Parse currency pair (e.g., BTCUSD -> BTC and USD)
+        source_currency = self.currency_pair[:3]
+        target_currency = self.currency_pair[3:]
+
+        try:
+            async with self.session.get(
+                f"{API_BASE_URL}{TICKER_ENDPOINT}",
+                headers=headers,
+                params={
+                    "sourceCurrency": source_currency,
+                    "targetCurrency": target_currency,
+                },
+                timeout=aiohttp.ClientTimeout(total=API_TIMEOUT),
+            ) as response:
+                if response.status == 401:
+                    raise UpdateFailed("Invalid API key")
+                if response.status == 403:
+                    raise UpdateFailed("API key forbidden")
+                if response.status != 200:
+                    raise UpdateFailed(f"API request failed with status {response.status}")
+
+                data = await response.json()
+
+                # Strike API returns the ticker data
+                # Expected format: {"amount": "50000.00"}
+                if "amount" not in data:
+                    raise UpdateFailed("Invalid response format from Strike API")
+
+                return {
+                    "price": float(data["amount"]),
+                    "currency": target_currency,
+                    "source_currency": source_currency,
+                    "last_update": datetime.now().isoformat(),
+                }
+
+        except aiohttp.ClientError as err:
+            _LOGGER.error("Error connecting to Strike API: %s", err)
+            raise UpdateFailed(f"Error connecting to Strike API: {err}") from err
+        except Exception as err:
+            _LOGGER.exception("Unexpected error fetching Strike data: %s", err)
+            raise UpdateFailed(f"Unexpected error: {err}") from err
+
+
+class StrikeBitcoinSensor(CoordinatorEntity, SensorEntity):
+    """Representation of a Strike Bitcoin Price Sensor."""
+
+    _attr_has_entity_name = True
+    _attr_name = None
+    _attr_device_class = SensorDeviceClass.MONETARY
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_icon = "mdi:bitcoin"
+
+    def __init__(
+        self, coordinator: StrikeBitcoinCoordinator, entry: ConfigEntry
+    ) -> None:
+        """Initialize the sensor."""
+        super().__init__(coordinator)
+
+        self._attr_unique_id = f"{entry.entry_id}_btc_price"
+        self._entry_id = entry.entry_id
+        self.currency_pair = entry.data.get(CONF_CURRENCY_PAIR, DEFAULT_CURRENCY_PAIR)
+
+        # Set device info
+        self._attr_device_info = {
+            "identifiers": {(DOMAIN, entry.entry_id)},
+            "name": f"Strike {self.currency_pair}",
+            "manufacturer": MANUFACTURER,
+            "model": MODEL,
+            "entry_type": "service",
+        }
+
+    @property
+    def native_value(self) -> float | None:
+        """Return the state of the sensor."""
+        if self.coordinator.data is None:
+            return None
+        return self.coordinator.data.get("price")
+
+    @property
+    def native_unit_of_measurement(self) -> str | None:
+        """Return the unit of measurement."""
+        if self.coordinator.data is None:
+            return None
+        return self.coordinator.data.get("currency", "USD")
+
+    @property
+    def extra_state_attributes(self) -> dict[str, Any]:
+        """Return the state attributes."""
+        if self.coordinator.data is None:
+            return {}
+
+        return {
+            ATTR_CURRENCY: self.coordinator.data.get("currency"),
+            ATTR_SOURCE: f"{self.coordinator.data.get('source_currency')}/{self.coordinator.data.get('currency')}",
+            ATTR_LAST_UPDATE: self.coordinator.data.get("last_update"),
+        }
+
+    @property
+    def available(self) -> bool:
+        """Return True if entity is available."""
+        return self.coordinator.last_update_success and self.coordinator.data is not None