123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""Sensor platform for Olimex ESP32-C6-EVB."""
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
import aiohttp
|
|
from homeassistant.components.sensor import SensorEntity
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddEntitiesCallback
|
|
from homeassistant.helpers.update_coordinator import (
|
|
CoordinatorEntity,
|
|
DataUpdateCoordinator,
|
|
UpdateFailed,
|
|
)
|
|
|
|
from .const import DOMAIN, DEVICE_INFO, DEFAULT_SCAN_INTERVAL
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
async_add_entities: AddEntitiesCallback,
|
|
) -> None:
|
|
"""Set up Olimex ESP32-C6-EVB sensors."""
|
|
host = entry.data["host"]
|
|
port = entry.data.get("port", 80)
|
|
|
|
coordinator = OlimexDataUpdateCoordinator(hass, host, port)
|
|
await coordinator.async_config_entry_first_refresh()
|
|
|
|
sensors = [
|
|
OlimexTemperatureSensor(coordinator, entry),
|
|
OlimexWiFiSignalSensor(coordinator, entry),
|
|
]
|
|
|
|
async_add_entities(sensors)
|
|
|
|
class OlimexDataUpdateCoordinator(DataUpdateCoordinator):
|
|
"""Data coordinator for Olimex ESP32-C6-EVB."""
|
|
|
|
def __init__(self, hass: HomeAssistant, host: str, port: int) -> None:
|
|
"""Initialize coordinator."""
|
|
super().__init__(
|
|
hass,
|
|
_LOGGER,
|
|
name=DOMAIN,
|
|
update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
|
|
)
|
|
self.host = host
|
|
self.port = port
|
|
|
|
async def _async_update_data(self):
|
|
"""Fetch data from ESP32-C6."""
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
# TODO: Update this URL based on your ESP32 firmware API
|
|
async with session.get(
|
|
f"http://{self.host}:{self.port}/api/status",
|
|
timeout=aiohttp.ClientTimeout(total=10),
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
raise UpdateFailed(f"Error fetching data: {response.status}")
|
|
except aiohttp.ClientError as err:
|
|
raise UpdateFailed(f"Error communicating with device: {err}")
|
|
|
|
class OlimexTemperatureSensor(CoordinatorEntity, SensorEntity):
|
|
"""Temperature sensor for ESP32-C6."""
|
|
|
|
def __init__(self, coordinator, entry):
|
|
"""Initialize the sensor."""
|
|
super().__init__(coordinator)
|
|
self._entry = entry
|
|
self._attr_name = "Temperature"
|
|
self._attr_unique_id = f"{entry.entry_id}_temperature"
|
|
self._attr_native_unit_of_measurement = "°C"
|
|
self._attr_device_class = "temperature"
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Return device information."""
|
|
return {
|
|
"identifiers": {(DOMAIN, self._entry.entry_id)},
|
|
"name": f"Olimex ESP32-C6 ({self._entry.data['host']})",
|
|
**DEVICE_INFO,
|
|
}
|
|
|
|
@property
|
|
def native_value(self):
|
|
"""Return the temperature value."""
|
|
if self.coordinator.data:
|
|
return self.coordinator.data.get("temperature")
|
|
return None
|
|
|
|
class OlimexWiFiSignalSensor(CoordinatorEntity, SensorEntity):
|
|
"""WiFi signal sensor for ESP32-C6."""
|
|
|
|
def __init__(self, coordinator, entry):
|
|
"""Initialize the sensor."""
|
|
super().__init__(coordinator)
|
|
self._entry = entry
|
|
self._attr_name = "WiFi Signal"
|
|
self._attr_unique_id = f"{entry.entry_id}_wifi_signal"
|
|
self._attr_native_unit_of_measurement = "dBm"
|
|
self._attr_device_class = "signal_strength"
|
|
|
|
@property
|
|
def device_info(self):
|
|
"""Return device information."""
|
|
return {
|
|
"identifiers": {(DOMAIN, self._entry.entry_id)},
|
|
"name": f"Olimex ESP32-C6 ({self._entry.data['host']})",
|
|
**DEVICE_INFO,
|
|
}
|
|
|
|
@property
|
|
def native_value(self):
|
|
"""Return the WiFi signal strength."""
|
|
if self.coordinator.data:
|
|
return self.coordinator.data.get("wifi_rssi")
|
|
return None
|