Files

541 lines
25 KiB
Python

"""Support for Gardena Smart System sensors."""
from __future__ import annotations
import logging
from datetime import datetime, timedelta, timezone
from typing import Any
from homeassistant.components.sensor import SensorEntity, SensorDeviceClass, SensorStateClass
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import PERCENTAGE, UnitOfTemperature, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import DeviceInfo, EntityCategory
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import (
DOMAIN,
ATTR_BATTERY_STATE,
ATTR_RF_LINK_LEVEL,
ATTR_RF_LINK_STATE,
MOWER_INFORMATIONAL_CODES,
)
from .coordinator import GardenaSmartSystemCoordinator
from .entities import GardenaEntity
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Gardena Smart System sensors."""
coordinator: GardenaSmartSystemCoordinator = hass.data[DOMAIN][entry.entry_id]
# Create sensor entities for each device
entities = []
for location in coordinator.locations.values():
for device in location.devices.values():
_LOGGER.debug(f"Checking device {device.name} ({device.id}) - Services: {list(device.services.keys())}")
# Add battery and RF link sensors for devices that have COMMON service
if "COMMON" in device.services:
common_services = device.services["COMMON"]
_LOGGER.debug(f"Found {len(common_services)} common services for device: {device.name} ({device.id})")
for common_service in common_services:
# Only create battery sensor if device has a battery.
# Include when battery_level is present even if battery_state is not yet
# populated (soil sensors can return null battery_state at initial load).
has_battery = (
common_service.battery_state not in [None, "NO_BATTERY"]
or common_service.battery_level is not None
)
if has_battery:
_LOGGER.debug(f"Creating battery sensor for device with battery: {device.name} (battery_state: {common_service.battery_state})")
entities.append(GardenaBatterySensor(coordinator, device, common_service))
else:
_LOGGER.debug(f"Skipping battery sensor for device without battery: {device.name} (battery_state: {common_service.battery_state})")
# RF Link Quality sensor
if common_service.rf_link_level is not None:
entities.append(GardenaRFLinkLevelSensor(coordinator, device, common_service))
# Add mower sensors
if "MOWER" in device.services:
mower_services = device.services["MOWER"]
for mower_service in mower_services:
entities.append(GardenaMowerErrorSensor(coordinator, device, mower_service))
if mower_service.operating_hours is not None:
entities.append(GardenaMowerOperatingHoursSensor(coordinator, device, mower_service))
# Add watering end time sensors for valves
if "VALVE" in device.services:
valve_services = device.services["VALVE"]
for valve_service in valve_services:
entities.append(GardenaValveRemainingTimeSensor(coordinator, device, valve_service))
# Add sensor entities if available
if "SENSOR" in device.services:
sensor_services = device.services["SENSOR"]
_LOGGER.debug(f"Found {len(sensor_services)} sensor services for device: {device.name} ({device.id})")
for sensor_service in sensor_services:
_LOGGER.debug(f"Creating sensor entities for service: {sensor_service.id}")
# Check if this is a soil sensor (has soil_humidity or soil_temperature)
is_soil_sensor = (sensor_service.soil_humidity is not None or
sensor_service.soil_temperature is not None)
# Create temperature sensors
if sensor_service.soil_temperature is not None:
entities.append(GardenaTemperatureSensor(coordinator, device, sensor_service, "soil_temperature", is_soil_sensor))
if sensor_service.ambient_temperature is not None:
entities.append(GardenaTemperatureSensor(coordinator, device, sensor_service, "ambient_temperature", is_soil_sensor))
# Create humidity sensor (only for soil sensors)
if sensor_service.soil_humidity is not None:
entities.append(GardenaHumiditySensor(coordinator, device, sensor_service))
# Create light sensor
if sensor_service.light_intensity is not None:
entities.append(GardenaLightSensor(coordinator, device, sensor_service))
# Add API usage diagnostic sensor (one per config entry)
entities.append(GardenaAPIUsageSensor(coordinator, entry.entry_id))
_LOGGER.debug(f"Created {len(entities)} sensor entities")
async_add_entities(entities)
class GardenaBatterySensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena battery sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, common_service) -> None:
"""Initialize the battery sensor."""
super().__init__(coordinator, device, "COMMON")
self._common_service = common_service
self._device_id = device.id
self._attr_name = f"{device.name} Battery Level"
self._attr_unique_id = f"{device.id}_{common_service.id}_battery_level"
self._attr_native_unit_of_measurement = PERCENTAGE
self._attr_device_class = SensorDeviceClass.BATTERY
self._attr_icon = "mdi:battery"
def _get_current_common_service(self):
"""Get current common service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "COMMON" in device.services:
for service in device.services["COMMON"]:
if service.id == self._common_service.id:
return service
return None
@property
def native_value(self) -> int | None:
"""Return the battery level."""
current_service = self._get_current_common_service()
return current_service.battery_level if current_service else None
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return entity specific state attributes."""
attrs = super().extra_state_attributes
current_service = self._get_current_common_service()
if current_service:
attrs.update({
ATTR_BATTERY_STATE: current_service.battery_state,
ATTR_RF_LINK_LEVEL: current_service.rf_link_level,
ATTR_RF_LINK_STATE: current_service.rf_link_state,
})
return attrs
class GardenaMowerErrorSensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena mower error code sensor."""
_attr_translation_key = "mower_error"
_attr_has_entity_name = True
_attr_device_class = SensorDeviceClass.ENUM
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, mower_service) -> None:
"""Initialize the mower error sensor."""
super().__init__(coordinator, device, "MOWER")
self._mower_service = mower_service
self._device_id = device.id
self._attr_name = None
self._attr_unique_id = f"{device.id}_{mower_service.id}_last_error_code"
self._attr_icon = "mdi:alert-circle-outline"
self._attr_options = [
"uninitialised", "no_message", "outside_working_area", "no_loop_signal",
"no_charging_station_signal", "wrong_loop_signal",
"loop_sensor_problem_front", "loop_sensor_problem_rear",
"trapped", "upside_down", "low_battery", "empty_battery", "no_drive",
"lifted", "stuck_in_charging_station", "charging_station_blocked",
"collision_sensor_problem_rear", "collision_sensor_problem_front",
"wheel_motor_blocked_right", "wheel_motor_blocked_left",
"wheel_drive_problem_right", "wheel_drive_problem_left",
"cutting_system_blocked", "invalid_sub_device_combination",
"settings_restored", "charging_system_problem", "tilt_sensor_problem",
"mower_tilted", "wheel_motor_overloaded_right",
"wheel_motor_overloaded_left", "charging_current_too_high",
"electronic_problem", "cutting_height_blocked", "cutting_height_problem",
"temporary_problem", "guide_1_not_found", "guide_2_not_found",
"guide_3_not_found", "gps_tracker_module_error", "weak_gps_signal",
"guide_calibration_failed", "temporary_battery_problem",
"battery_problem", "alarm_mower_switched_off", "alarm_mower_stopped",
"alarm_mower_lifted", "alarm_mower_tilted", "com_board_not_available",
"slipped", "invalid_battery_combination", "safety_function_faulty",
"invalid_system_conf", "lift_sensor_defect", "mobile_loop_defect",
"left_loop_sensor", "right_loop_sensor", "wrong_pin", "temporary_lift",
"cutting_drive", "steep_slope", "stop_button_fail",
"angle_cutting_means_off", "slave_mcu_lost", "cutting_overload",
"cutting_height_range", "cutting_height_drift", "cutting_height_limited",
"cutting_height_drive", "cutting_height_current",
"cutting_height_direction", "mower_to_cs_com", "ultrasonic_error",
"high_low_bat_temp_a", "high_low_bat_temp_b",
"too_low_voltage_bat_a", "too_low_voltage_bat_b",
"alarm_motion", "alarm_geofence",
"rr_wheel_blocked", "rl_wheel_blocked",
"rr_wheel_drive", "rl_wheel_drive",
"rear_right_wheel_overloaded", "rear_left_wheel_overloaded",
"angular_sensor_defect", "no_power_in_cs", "switch_cord_sensor_defect",
"map_not_valid", "no_position", "no_rs_communication",
"folding_sensor_activated",
"ultrasonic_sensor_1_defect", "ultrasonic_sensor_2_defect",
"ultrasonic_sensor_3_defect", "ultrasonic_sensor_4_defect",
"cutting_drive_motor_1_defect", "cutting_drive_motor_2_defect",
"cutting_drive_motor_3_defect",
"collision_sensor_defect", "docking_sensor_defect",
"folding_cutting_deck_sensor_defect", "loop_sensor_defect",
"collision_sensor_error", "no_confirmed_position",
"major_cutting_disk_imbalance", "complex_working_area",
"invalid_sw_configuration", "radar_error", "work_area_tampered",
"destination_not_reachable", "wait_stop_pressed", "wait_for_safety_pin",
"destination_not_reachable_warning", "battery_near_end_of_life",
"edgemotor_blocked", "no_correction_data", "invalid_correction_data",
"wait_updating", "wait_power_up", "off_disabled", "off_hatch_open",
"off_hatch_closed", "parked_daily_limit_reached",
"vision_system_malfunction", "poor_vision_system_performance",
"vision_processing_failed",
"loop_sensor_problem_left", "loop_sensor_problem_right",
"wrong_pin_code", "lost_wheel_brush", "accessory_power_anomaly",
"loop_wire_broken", "battery_fet_error", "imbalanced_cutting_disc",
"cutting_motor_problem", "limited_cutting_height_range",
"cutting_motor_drive_defect", "memory_circuit_problem",
"stop_button_problem", "difficult_finding_home",
"guide_calibration_accomplished", "too_many_batteries",
"alarm_mower_in_motion", "alarm_outside_geofence",
"connection_changed", "connection_not_changed",
"unknown",
]
def _get_current_mower_service(self):
"""Get current mower service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "MOWER" in device.services:
for service in device.services["MOWER"]:
if service.id == self._mower_service.id:
return service
return None
@property
def native_value(self) -> str | None:
"""Return the last error code, or 'no_message' for informational states."""
current_service = self._get_current_mower_service()
if not current_service:
return None
error_code = current_service.last_error_code
if error_code:
code = error_code.lower()
if code in MOWER_INFORMATIONAL_CODES:
return "no_message"
return code
return "no_message"
@property
def icon(self) -> str:
"""Return icon based on error state."""
value = self.native_value
if value and value != "no_message":
return "mdi:alert-circle"
return "mdi:check-circle-outline"
class GardenaMowerOperatingHoursSensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena mower operating hours sensor."""
_attr_state_class = SensorStateClass.TOTAL_INCREASING
_attr_device_class = SensorDeviceClass.DURATION
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, mower_service) -> None:
"""Initialize the operating hours sensor."""
super().__init__(coordinator, device, "MOWER")
self._mower_service = mower_service
self._device_id = device.id
self._attr_name = f"{device.name} Operating Hours"
self._attr_unique_id = f"{device.id}_{mower_service.id}_operating_hours"
self._attr_native_unit_of_measurement = UnitOfTime.HOURS
self._attr_icon = "mdi:clock-outline"
def _get_current_mower_service(self):
"""Get current mower service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "MOWER" in device.services:
for service in device.services["MOWER"]:
if service.id == self._mower_service.id:
return service
return None
@property
def native_value(self) -> int | None:
"""Return the operating hours."""
current_service = self._get_current_mower_service()
return current_service.operating_hours if current_service else None
class GardenaRFLinkLevelSensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena RF link quality sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, common_service) -> None:
"""Initialize the RF link level sensor."""
super().__init__(coordinator, device, "COMMON")
self._common_service = common_service
self._device_id = device.id
self._attr_name = f"{device.name} RF Link Quality"
self._attr_unique_id = f"{device.id}_{common_service.id}_rf_link_level"
self._attr_native_unit_of_measurement = PERCENTAGE
self._attr_icon = "mdi:signal"
def _get_current_common_service(self):
"""Get current common service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "COMMON" in device.services:
for service in device.services["COMMON"]:
if service.id == self._common_service.id:
return service
return None
@property
def native_value(self) -> int | None:
"""Return the RF link level."""
current_service = self._get_current_common_service()
return current_service.rf_link_level if current_service else None
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return entity specific state attributes."""
attrs = super().extra_state_attributes
current_service = self._get_current_common_service()
if current_service:
attrs[ATTR_RF_LINK_STATE] = current_service.rf_link_state
return attrs
class GardenaTemperatureSensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena temperature sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, sensor_service, temp_attr: str, is_soil_sensor: bool = False) -> None:
"""Initialize the temperature sensor."""
super().__init__(coordinator, device, "SENSOR")
self._sensor_service = sensor_service
self._device_id = device.id
self._temp_attr = temp_attr
if temp_attr == "soil_temperature":
self._attr_name = f"{device.name} Soil Temperature"
self._attr_unique_id = f"{device.id}_{sensor_service.id}_soil_temperature"
else:
self._attr_name = f"{device.name} Ambient Temperature"
self._attr_unique_id = f"{device.id}_{sensor_service.id}_ambient_temperature"
# Add soil sensor indicator to name if it's a soil sensor
if is_soil_sensor and temp_attr == "soil_temperature":
self._attr_name = f"{device.name} Soil Temperature (Soil Sensor)"
self._attr_native_unit_of_measurement = UnitOfTemperature.CELSIUS
self._attr_device_class = SensorDeviceClass.TEMPERATURE
self._attr_icon = "mdi:thermometer"
def _get_current_sensor_service(self):
"""Get current sensor service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "SENSOR" in device.services:
for service in device.services["SENSOR"]:
if service.id == self._sensor_service.id:
return service
return None
@property
def native_value(self) -> int | None:
"""Return the temperature value."""
current_service = self._get_current_sensor_service()
if not current_service:
return None
if self._temp_attr == "soil_temperature":
return current_service.soil_temperature
else:
return current_service.ambient_temperature
class GardenaHumiditySensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena humidity sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, sensor_service) -> None:
"""Initialize the humidity sensor."""
super().__init__(coordinator, device, "SENSOR")
self._sensor_service = sensor_service
self._device_id = device.id
self._attr_name = f"{device.name} Soil Humidity"
self._attr_unique_id = f"{device.id}_{sensor_service.id}_soil_humidity"
self._attr_native_unit_of_measurement = PERCENTAGE
self._attr_device_class = SensorDeviceClass.MOISTURE
self._attr_icon = "mdi:water-percent"
def _get_current_sensor_service(self):
"""Get current sensor service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "SENSOR" in device.services:
for service in device.services["SENSOR"]:
if service.id == self._sensor_service.id:
return service
return None
@property
def native_value(self) -> int | None:
"""Return the humidity value."""
current_service = self._get_current_sensor_service()
return current_service.soil_humidity if current_service else None
class GardenaLightSensor(GardenaEntity, SensorEntity):
"""Representation of a Gardena light sensor."""
_attr_state_class = SensorStateClass.MEASUREMENT
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, sensor_service) -> None:
"""Initialize the light sensor."""
super().__init__(coordinator, device, "SENSOR")
self._sensor_service = sensor_service
self._device_id = device.id
self._attr_name = f"{device.name} Light Intensity"
self._attr_unique_id = f"{device.id}_{sensor_service.id}_light_intensity"
self._attr_native_unit_of_measurement = "lx"
self._attr_device_class = SensorDeviceClass.ILLUMINANCE
self._attr_icon = "mdi:white-balance-sunny"
def _get_current_sensor_service(self):
"""Get current sensor service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "SENSOR" in device.services:
for service in device.services["SENSOR"]:
if service.id == self._sensor_service.id:
return service
return None
@property
def native_value(self) -> int | None:
"""Return the light intensity value."""
current_service = self._get_current_sensor_service()
return current_service.light_intensity if current_service else None
class GardenaValveRemainingTimeSensor(GardenaEntity, SensorEntity):
"""Sensor that shows the watering end time for a valve."""
_attr_device_class = SensorDeviceClass.TIMESTAMP
def __init__(self, coordinator: GardenaSmartSystemCoordinator, device, valve_service) -> None:
"""Initialize the valve remaining time sensor."""
super().__init__(coordinator, device, "VALVE")
self._valve_service = valve_service
self._device_id = device.id
valve_name = valve_service.name or device.name
self._attr_name = f"{valve_name} Watering End"
self._attr_unique_id = f"{device.id}_{valve_service.id}_watering_end"
self._attr_icon = "mdi:timer-sand"
def _get_current_valve_service(self):
"""Get current valve service from coordinator (fresh data)."""
device = self.coordinator.get_device_by_id(self._device_id)
if device and "VALVE" in device.services:
for service in device.services["VALVE"]:
if service.id == self._valve_service.id:
return service
return None
@property
def native_value(self) -> datetime | None:
"""Return the watering end timestamp."""
current_service = self._get_current_valve_service()
if not current_service:
return None
if current_service.activity in ("MANUAL_WATERING", "SCHEDULED_WATERING"):
if current_service.duration and current_service.duration_timestamp:
try:
start = datetime.fromisoformat(
current_service.duration_timestamp.replace("Z", "+00:00")
)
return start + timedelta(seconds=current_service.duration)
except (ValueError, TypeError):
return None
return None
class GardenaAPIUsageSensor(SensorEntity):
"""Sensor tracking Gardena API usage against the 700 requests/week quota."""
_attr_has_entity_name = True
_attr_translation_key = "api_requests_week"
_attr_icon = "mdi:api"
_attr_state_class = SensorStateClass.TOTAL
_attr_entity_category = EntityCategory.DIAGNOSTIC
def __init__(self, coordinator: GardenaSmartSystemCoordinator, entry_id: str) -> None:
"""Initialize the API usage sensor."""
self.coordinator = coordinator
self._entry_id = entry_id
self._attr_unique_id = f"gardena_api_usage_{entry_id}"
self._attr_name = "API Requests (Week)"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, f"websocket_status_{entry_id}")},
)
@property
def available(self) -> bool:
"""Always available."""
return True
@property
def native_value(self) -> int:
"""Return the number of API requests this week."""
return self.coordinator.client.api_tracker.requests_this_week
@property
def extra_state_attributes(self) -> dict[str, Any]:
"""Return diagnostic attributes with request breakdown."""
tracker = self.coordinator.client.api_tracker
attrs: dict[str, Any] = {
"requests_today": tracker.requests_today,
"requests_this_week": tracker.requests_this_week,
"quota_weekly": 700,
"requests_by_endpoint": tracker.requests_by_endpoint(),
"recent_requests": tracker.recent_requests,
}
return attrs
async def async_update(self) -> None:
"""No-op: value is computed on read from the shared tracker."""