homeassistant-powersensor/custom_components/powersensor/PowersensorHouseholdEntity.py at 3b1428007aed02224d4b24633ae8a1f065c16ecc · DiUS/homeassistant-powersensor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
from datetime import date, datetime
from dataclasses import dataclass
from enum import Enum
from typing import Callable

from homeassistant.components.sensor import SensorEntity, SensorDeviceClass, SensorStateClass
from homeassistant.const import UnitOfPower, UnitOfEnergy
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.typing import StateType

from powersensor_local import VirtualHousehold

from .const import DOMAIN

class HouseholdMeasurements(Enum):
    POWER_HOME_USE = 1
    POWER_FROM_GRID = 2
    POWER_TO_GRID = 3
    POWER_SOLAR_GENERATION = 4
    ENERGY_HOME_USE = 5
    ENERGY_FROM_GRID = 6
    ENERGY_TO_GRID = 7
    ENERGY_SOLAR_GENERATION = 8

ConsumptionMeasurements = [
    HouseholdMeasurements.POWER_HOME_USE,
    HouseholdMeasurements.POWER_FROM_GRID,
    HouseholdMeasurements.ENERGY_HOME_USE,
    HouseholdMeasurements.ENERGY_FROM_GRID
]
ProductionMeasurements = [
    HouseholdMeasurements.POWER_TO_GRID,
    HouseholdMeasurements.POWER_SOLAR_GENERATION,
    HouseholdMeasurements.ENERGY_TO_GRID,
    HouseholdMeasurements.ENERGY_SOLAR_GENERATION
]

@dataclass
class EntityConfig:
    name : str
    device_class : SensorDeviceClass
    state_class : SensorStateClass | None
    unit : UnitOfPower | UnitOfEnergy
    formatter: Callable
    precision: int
    event: str

FMT_INT = lambda f: int(f)
FMT_WS_TO_KWH = lambda f: float(f)/3600000

class PowersensorHouseholdEntity(SensorEntity):
    """Powersensor Virtual Household entity"""

    should_poll = False
    _attr_has_entity_name = True
    _attr_available = True

    _ENTITY_CONFIGS  = {
        HouseholdMeasurements.POWER_HOME_USE : EntityConfig(
            "Power - Home use", SensorDeviceClass.POWER, None, UnitOfPower.WATT, FMT_INT, 0, "home_usage"),
        HouseholdMeasurements.POWER_FROM_GRID : EntityConfig(
            "Power - From grid", SensorDeviceClass.POWER, None, UnitOfPower.WATT, FMT_INT, 0, "from_grid"),
        HouseholdMeasurements.POWER_TO_GRID : EntityConfig(
            "Power - To grid", SensorDeviceClass.POWER, None, UnitOfPower.WATT, FMT_INT, 0, "to_grid"),
        HouseholdMeasurements.POWER_SOLAR_GENERATION : EntityConfig(
            "Power - Solar generation", SensorDeviceClass.POWER, None, UnitOfPower.WATT, FMT_INT, 0, "solar_generation"),

        HouseholdMeasurements.ENERGY_HOME_USE : EntityConfig(
            "Energy - Home usage", SensorDeviceClass.ENERGY, SensorStateClass.TOTAL, UnitOfEnergy.KILO_WATT_HOUR, FMT_WS_TO_KWH, 3, "home_usage_summation"),
        HouseholdMeasurements.ENERGY_FROM_GRID : EntityConfig(
            "Energy - From grid", SensorDeviceClass.ENERGY, SensorStateClass.TOTAL, UnitOfEnergy.KILO_WATT_HOUR, FMT_WS_TO_KWH, 3, "from_grid_summation"),
        HouseholdMeasurements.ENERGY_TO_GRID : EntityConfig(
            "Energy - To grid", SensorDeviceClass.ENERGY, SensorStateClass.TOTAL, UnitOfEnergy.KILO_WATT_HOUR, FMT_WS_TO_KWH, 3, "to_grid_summation"),
        HouseholdMeasurements.ENERGY_SOLAR_GENERATION : EntityConfig(
            "Energy - Solar generation", SensorDeviceClass.ENERGY, SensorStateClass.TOTAL, UnitOfEnergy.KILO_WATT_HOUR, FMT_WS_TO_KWH, 3, "solar_generation_summation"),
    }

    def __init__(self, vhh: VirtualHousehold, measurement_type: HouseholdMeasurements):
        """Initialize the entity."""
        self._vhh = vhh
        self._config = self._ENTITY_CONFIGS[measurement_type]

        self._attr_name = self._config.name
        self._attr_unique_id = f"{DOMAIN}_vhh_{self._config.event}"
        self._attr_device_class = self._config.device_class
        self._attr_state_class = self._config.state_class
        self._attr_native_unit_of_measurement = self._config.unit
        self._attr_suggested_display_precision = self._config.precision

    @property
    def device_info(self) -> DeviceInfo:
        return {
            'identifiers': {(DOMAIN, "vhh")},
            'manufacturer': "Powersensor",
            'model': "Virtual",
            'name': "Powersensor Household View 🏠",
        }

    async def async_added_to_hass(self):
        self._vhh.subscribe(self._config.event, self._on_event)

    async def async_will_remove_from_hass(self):
        self._vhh.unsubscribe(self._config.event, self._on_event)

    async def _on_event(self, _, msg):
        val = None
        if self._config.unit == UnitOfPower.WATT:
            key = "watts"
            if key in msg:
                val = msg[key]
        elif self._config.unit == UnitOfEnergy.KILO_WATT_HOUR:
            key = "summation_joules"
            if key in msg:
                val = msg[key]
            key = "summation_resettime_utc"
            if key in msg:
                self._attr_last_reset = datetime.fromtimestamp(msg[key])
        if val is not None:
            self._attr_native_value = self._config.formatter(val)
            self.async_write_ha_state()