diff --git a/tests/test_owon.py b/tests/test_owon.py new file mode 100644 index 0000000000..51566de09d --- /dev/null +++ b/tests/test_owon.py @@ -0,0 +1,144 @@ +"""Test units for Owon units.""" + +import pytest +from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement +from zigpy.zcl.clusters.smartenergy import Metering + +import zhaquirks +from zhaquirks.const import ENDPOINTS, INPUT_CLUSTERS +from zhaquirks.owon.pc321 import ( + ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES, + METERING_CONSTANT_ATTRIBUTES, + Owon_PC321, + OwonElectricalMeasurementPhaseA, + OwonElectricalMeasurementPhaseB, + OwonElectricalMeasurementPhaseC, + OwonManufacturerSpecific, + OwonMeteringPhaseA, + OwonMeteringPhaseB, + OwonMeteringPhaseC, +) + +zhaquirks.setup() + + +def test_pc321_signature(assert_signature_matches_quirk): + """Test Owon_PC321 cover signature is matched to its quirk.""" + + signature = { + "node_descriptor": "NodeDescriptor(logical_type=, complex_descriptor_available=0, user_descriptor_available=0, reserved=0, aps_flags=0, frequency_band=, mac_capability_flags=, manufacturer_code=4412, maximum_buffer_size=82, maximum_incoming_transfer_size=82, server_mask=11264, maximum_outgoing_transfer_size=82, descriptor_capability_field=, *allocate_address=True, *is_alternate_pan_coordinator=False, *is_coordinator=False, *is_end_device=False, *is_full_function_device=True, *is_mains_powered=True, *is_receiver_on_when_idle=True, *is_router=True, *is_security_capable=False)", + "endpoints": { + "1": { + "profile_id": 0x0104, + "device_type": "0x000d", + "in_clusters": [ + "0x0000", + "0x0003", + "0x0702", + ], + "out_clusters": ["0x0003"], + } + }, + "manufacturer": "OWON Technology Inc.", + "model": "PC321", + "class": "zigpy.device.Device", + } + assert_signature_matches_quirk(Owon_PC321, signature) + + +BUS_NAMES = [ + "energy_consumption_ph_a_bus", + "energy_consumption_ph_b_bus", + "energy_consumption_ph_c_bus", + "total_energy_consumption_bus", + "active_power_bus", + "active_power_ph_b_bus", + "active_power_ph_c_bus", + "total_active_power_bus", + "reactive_power_bus", + "reactive_power_ph_b_bus", + "reactive_power_ph_c_bus", + "total_reactive_power_bus", + "rms_voltage_bus", + "rms_voltage_ph_b_bus", + "rms_voltage_ph_c_bus", + "active_current_bus", + "active_current_ph_b_bus", + "active_current_ph_c_bus", + "instantaneous_active_current_bus", + "reactive_energy_consumption_ph_a_bus", + "reactive_energy_consumption_ph_b_bus", + "reactive_energy_consumption_ph_c_bus", + "total_reactive_energy_consumption_bus", + "ac_frequency_bus", + "leakage_current_bus", +] + + +@pytest.mark.parametrize("quirk", (zhaquirks.owon.pc321.Owon_PC321,)) +def test_pc321_replacement(zigpy_device_from_quirk, quirk): + """Test that the quirk has appropriate replacement endpoints.""" + device = zigpy_device_from_quirk(quirk) + assert device.replacement is not None + + # check that the replacement first endpoint has ManufacturerSpecificCluster + assert quirk.replacement[ENDPOINTS][1][INPUT_CLUSTERS].pop().cluster_id == 0xFD32 + + # check that quirk has necessary buses + for bus in BUS_NAMES: + assert isinstance(device.__dict__[bus], zhaquirks.Bus) + + for endpoint_id, endpoint in quirk.replacement[ENDPOINTS].items(): + if endpoint_id == 1: + continue + for cluster in endpoint[INPUT_CLUSTERS]: + # must expose only Metering or ElectricalMeasurement subclasses + assert issubclass(cluster, Metering) or issubclass( + cluster, ElectricalMeasurement + ) + + +def test_ManufacturerSpecific_cluster(): + """Test Manufacturer specific cluster has necessary functions.""" + + for fn in BUS_NAMES: + assert callable( + getattr(OwonManufacturerSpecific, fn.replace("_bus", "_reported")) + ) + + +def test_ElectricalMeasurement_clusters(): + """Test ElectricalMeasurement cluster has necessary functions.""" + + assert ( + OwonElectricalMeasurementPhaseA._CONSTANT_ATTRIBUTES + == ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES + ) + assert ( + OwonElectricalMeasurementPhaseB._CONSTANT_ATTRIBUTES + == ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES + ) + assert ( + OwonElectricalMeasurementPhaseC._CONSTANT_ATTRIBUTES + == ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES + ) + assert callable(OwonElectricalMeasurementPhaseA.rms_voltage_reported) + assert callable(OwonElectricalMeasurementPhaseA.active_power_reported) + assert callable(OwonElectricalMeasurementPhaseB.rms_voltage_ph_b_reported) + assert callable(OwonElectricalMeasurementPhaseB.active_power_ph_b_reported) + assert callable(OwonElectricalMeasurementPhaseC.rms_voltage_ph_c_reported) + assert callable(OwonElectricalMeasurementPhaseC.active_power_ph_c_reported) + + +def test_Metering_clusters(): + """Test Metering cluster has necessary functions.""" + + assert OwonMeteringPhaseA._CONSTANT_ATTRIBUTES == METERING_CONSTANT_ATTRIBUTES + assert OwonMeteringPhaseB._CONSTANT_ATTRIBUTES == METERING_CONSTANT_ATTRIBUTES + assert OwonMeteringPhaseC._CONSTANT_ATTRIBUTES == METERING_CONSTANT_ATTRIBUTES + assert callable(OwonMeteringPhaseA.energy_consumption_ph_a_reported) + assert callable(OwonMeteringPhaseA.active_power_reported) + assert callable(OwonMeteringPhaseB.energy_consumption_ph_b_reported) + assert callable(OwonMeteringPhaseB.active_power_ph_b_reported) + assert callable(OwonMeteringPhaseC.energy_consumption_ph_c_reported) + assert callable(OwonMeteringPhaseC.active_power_ph_c_reported) diff --git a/zhaquirks/owon/__init__.py b/zhaquirks/owon/__init__.py new file mode 100644 index 0000000000..9c2717d41d --- /dev/null +++ b/zhaquirks/owon/__init__.py @@ -0,0 +1,3 @@ +"""Module for OWON quirks implementations.""" + +OWON = "OWON Technology Inc." diff --git a/zhaquirks/owon/pc321.py b/zhaquirks/owon/pc321.py new file mode 100644 index 0000000000..d4eb2ed964 --- /dev/null +++ b/zhaquirks/owon/pc321.py @@ -0,0 +1,708 @@ +"""QUIRK FOR OWON PC321 (non-Tuya version).""" + +from zigpy.profiles import zha +from zigpy.quirks import CustomCluster, CustomDevice +import zigpy.types as t +from zigpy.zcl.clusters.general import Basic, Identify +from zigpy.zcl.clusters.homeautomation import ElectricalMeasurement +from zigpy.zcl.clusters.smartenergy import Metering + +from zhaquirks import Bus, LocalDataCluster +from zhaquirks.const import ( + DEVICE_TYPE, + ENDPOINTS, + INPUT_CLUSTERS, + MODELS_INFO, + OUTPUT_CLUSTERS, + PROFILE_ID, +) +from zhaquirks.owon import OWON + +OWON_PC321 = "PC321" +OWON_PC321_CLUSTER_ID = 0xFD32 + +"""Owon PC321 Attributes""" + +# Consumption +OWON_METERING_ENERGY_CONSUMPTION_PH_A_ATTR = 0x4000 +OWON_METERING_ENERGY_CONSUMPTION_PH_B_ATTR = 0x4001 +OWON_METERING_ENERGY_CONSUMPTION_PH_C_ATTR = 0x4002 +OWON_METERING_TOTAL_ENERGY_CONSUMPTION_ATTR = 0x0000 +# Active power +OWON_METERING_ACTIVE_POWER_PH_A_ATTR = 0x2000 +OWON_METERING_ACTIVE_POWER_PH_B_ATTR = 0x2001 +OWON_METERING_ACTIVE_POWER_PH_C_ATTR = 0x2002 +OWON_METERING_TOTAL_ACTIVE_POWER_ATTR = 0x0400 +# Reactive power +OWON_METERING_REACTIVE_POWER_PH_A_ATTR = 0x2100 +OWON_METERING_REACTIVE_POWER_PH_B_ATTR = 0x2101 +OWON_METERING_REACTIVE_POWER_PH_C_ATTR = 0x2102 +OWON_METERING_TOTAL_REACTIVE_POWER_ATTR = 0x2103 +# Voltage +OWON_METERING_VOLTAGE_PH_A_ATTR = 0x3000 +OWON_METERING_VOLTAGE_PH_B_ATTR = 0x3001 +OWON_METERING_VOLTAGE_PH_C_ATTR = 0x3002 +# Active current +OWON_METERING_ACTIVE_CURRENT_PH_A_ATTR = 0x3100 +OWON_METERING_ACTIVE_CURRENT_PH_B_ATTR = 0x3101 +OWON_METERING_ACTIVE_CURRENT_PH_C_ATTR = 0x3102 +OWON_METERING_TOTAL_ACTIVE_CURRENT_ATTR = 0x3103 +# Reactive energy consumption +OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_A_ATTR = 0x4100 +OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_B_ATTR = 0x4101 +OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_C_ATTR = 0x4102 +OWON_METERING_TOTAL_REACTIVE_ENERGY_CONSUMPTION_ATTR = 0x4103 +# Frequency +OWON_METERING_AC_FREQUENCY_ATTR = 0x5005 +# Leakage +OWON_METERING_LEAKAGE_CURRENT_ATTR = 0x3104 + +METERING_CONSTANT_ATTRIBUTES = { + Metering.AttributeDefs.unit_of_measure.id: 0, + Metering.AttributeDefs.multiplier.id: 1, + Metering.AttributeDefs.divisor.id: 1000, + Metering.AttributeDefs.summation_formatting.id: 251, + Metering.AttributeDefs.metering_device_type.id: 0, + Metering.AttributeDefs.status: 0, +} + +ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES = { + ElectricalMeasurement.AttributeDefs.ac_voltage_multiplier.id: 1, + ElectricalMeasurement.AttributeDefs.ac_voltage_divisor.id: 10, +} + + +class OwonMetering(CustomCluster, Metering): + """Owon non-standard Metering cluster attributes, to be mapped into bus for later use in another clusters.""" + + def __init__(self, *args, **kwargs): + """Init.""" + self._current_state = {} + super().__init__(*args, **kwargs) + + def _update_attribute(self, attrid, value): + super()._update_attribute(attrid, value) + + # Consumption + if attrid == OWON_METERING_ENERGY_CONSUMPTION_PH_A_ATTR: + self.endpoint.device.energy_consumption_ph_a_bus.listener_event( + "energy_consumption_ph_a_reported", value + ) + elif attrid == OWON_METERING_ENERGY_CONSUMPTION_PH_B_ATTR: + self.endpoint.device.energy_consumption_ph_b_bus.listener_event( + "energy_consumption_ph_b_reported", value + ) + elif attrid == OWON_METERING_ENERGY_CONSUMPTION_PH_C_ATTR: + self.endpoint.device.energy_consumption_ph_c_bus.listener_event( + "energy_consumption_ph_c_reported", value + ) + elif attrid == OWON_METERING_TOTAL_ENERGY_CONSUMPTION_ATTR: + self.endpoint.device.total_energy_consumption_bus.listener_event( + "total_energy_consumption_reported", value + ) + # Active power + elif attrid == OWON_METERING_ACTIVE_POWER_PH_A_ATTR: + self.endpoint.device.active_power_bus.listener_event( + "active_power_reported", value + ) + elif attrid == OWON_METERING_ACTIVE_POWER_PH_B_ATTR: + self.endpoint.device.active_power_ph_b_bus.listener_event( + "active_power_ph_b_reported", value + ) + elif attrid == OWON_METERING_ACTIVE_POWER_PH_C_ATTR: + self.endpoint.device.active_power_ph_c_bus.listener_event( + "active_power_ph_c_reported", value + ) + elif attrid == OWON_METERING_TOTAL_ACTIVE_POWER_ATTR: + self.endpoint.device.total_active_power_bus.listener_event( + "total_active_power_reported", value + ) + # Reactive power + elif attrid == OWON_METERING_REACTIVE_POWER_PH_A_ATTR: + self.endpoint.device.reactive_power_bus.listener_event( + "reactive_power_reported", value + ) + elif attrid == OWON_METERING_REACTIVE_POWER_PH_B_ATTR: + self.endpoint.device.reactive_power_ph_b_bus.listener_event( + "reactive_power_ph_b_reported", value + ) + elif attrid == OWON_METERING_REACTIVE_POWER_PH_C_ATTR: + self.endpoint.device.reactive_power_ph_c_bus.listener_event( + "reactive_power_ph_c_reported", value + ) + elif attrid == OWON_METERING_TOTAL_REACTIVE_POWER_ATTR: + self.endpoint.device.total_reactive_power_bus.listener_event( + "total_reactive_power_reported", value + ) + # Voltage + elif attrid == OWON_METERING_VOLTAGE_PH_A_ATTR: + self.endpoint.device.rms_voltage_bus.listener_event( + "rms_voltage_reported", value + ) + elif attrid == OWON_METERING_VOLTAGE_PH_B_ATTR: + self.endpoint.device.rms_voltage_ph_b_bus.listener_event( + "rms_voltage_ph_b_reported", value + ) + elif attrid == OWON_METERING_VOLTAGE_PH_C_ATTR: + self.endpoint.device.rms_voltage_ph_c_bus.listener_event( + "rms_voltage_ph_c_reported", value + ) + # Active current + elif attrid == OWON_METERING_ACTIVE_CURRENT_PH_A_ATTR: + self.endpoint.device.active_current_bus.listener_event( + "active_current_reported", value + ) + elif attrid == OWON_METERING_ACTIVE_CURRENT_PH_B_ATTR: + self.endpoint.device.active_current_ph_b_bus.listener_event( + "active_current_ph_b_reported", value + ) + elif attrid == OWON_METERING_ACTIVE_CURRENT_PH_C_ATTR: + self.endpoint.device.active_current_ph_c_bus.listener_event( + "active_current_ph_c_reported", value + ) + elif attrid == OWON_METERING_TOTAL_ACTIVE_CURRENT_ATTR: + self.endpoint.device.instantaneous_active_current_bus.listener_event( + "instantaneous_active_current_reported", value + ) + # Reactive energy consumption + elif attrid == OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_A_ATTR: + self.endpoint.device.reactive_energy_consumption_ph_a_bus.listener_event( + "reactive_energy_consumption_ph_a_reported", value + ) + elif attrid == OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_B_ATTR: + self.endpoint.device.reactive_energy_consumption_ph_b_bus.listener_event( + "reactive_energy_consumption_ph_b_reported", value + ) + elif attrid == OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_C_ATTR: + self.endpoint.device.reactive_energy_consumption_ph_c_bus.listener_event( + "reactive_energy_consumption_ph_c_reported", value + ) + elif attrid == OWON_METERING_TOTAL_REACTIVE_ENERGY_CONSUMPTION_ATTR: + self.endpoint.device.total_reactive_energy_consumption_bus.listener_event( + "total_reactive_energy_consumption_reported", value + ) + # Frequency + elif attrid == OWON_METERING_AC_FREQUENCY_ATTR: + self.endpoint.device.ac_frequency_bus.listener_event( + "ac_frequency_reported", value + ) + # Leakage + elif attrid == OWON_METERING_LEAKAGE_CURRENT_ATTR: + self.endpoint.device.leakage_current_bus.listener_event( + "leakage_current_reported", value + ) + + +class OwonManufacturerSpecific(LocalDataCluster): + """Owon manufacturer specific attributes.""" + + cluster_id = OWON_PC321_CLUSTER_ID + ep_attribute = "owon_pc321_manufacturer_specific_cluster" + + attributes = { + # Energy Consumption + OWON_METERING_ENERGY_CONSUMPTION_PH_A_ATTR: ( + "energy_consumption_ph_a", + t.uint48_t, + True, + ), + OWON_METERING_ENERGY_CONSUMPTION_PH_B_ATTR: ( + "energy_consumption_ph_b", + t.uint48_t, + True, + ), + OWON_METERING_ENERGY_CONSUMPTION_PH_C_ATTR: ( + "energy_consumption_ph_c", + t.uint48_t, + True, + ), + OWON_METERING_TOTAL_ENERGY_CONSUMPTION_ATTR: ( + "total_energy_consumption", + t.uint48_t, + True, + ), + # Active power + OWON_METERING_ACTIVE_POWER_PH_A_ATTR: ("active_power_ph_a", t.uint24_t, True), + OWON_METERING_ACTIVE_POWER_PH_B_ATTR: ("active_power_ph_b", t.uint24_t, True), + OWON_METERING_ACTIVE_POWER_PH_C_ATTR: ("active_power_ph_c", t.uint24_t, True), + OWON_METERING_TOTAL_ACTIVE_POWER_ATTR: ("total_active_power", t.uint24_t, True), + # Reactive power + OWON_METERING_REACTIVE_POWER_PH_A_ATTR: ( + "reactive_power_ph_a", + t.uint24_t, + True, + ), + OWON_METERING_REACTIVE_POWER_PH_B_ATTR: ( + "reactive_power_ph_b", + t.uint24_t, + True, + ), + OWON_METERING_REACTIVE_POWER_PH_C_ATTR: ( + "reactive_power_ph_c", + t.uint24_t, + True, + ), + OWON_METERING_TOTAL_REACTIVE_POWER_ATTR: ( + "total_reactive_power", + t.uint24_t, + True, + ), + # Voltage + OWON_METERING_VOLTAGE_PH_A_ATTR: ("rms_voltage_ph_a", t.uint24_t, True), + OWON_METERING_VOLTAGE_PH_B_ATTR: ("rms_voltage_ph_b", t.uint24_t, True), + OWON_METERING_VOLTAGE_PH_C_ATTR: ("rms_voltage_ph_c", t.uint24_t, True), + # Active current + OWON_METERING_ACTIVE_CURRENT_PH_A_ATTR: ( + "active_current_ph_a", + t.uint24_t, + True, + ), + OWON_METERING_ACTIVE_CURRENT_PH_B_ATTR: ( + "active_current_ph_b", + t.uint24_t, + True, + ), + OWON_METERING_ACTIVE_CURRENT_PH_C_ATTR: ( + "active_current_ph_c", + t.uint24_t, + True, + ), + OWON_METERING_TOTAL_ACTIVE_CURRENT_ATTR: ( + "total_active_current", + t.uint24_t, + True, + ), + # Reactive energy consumption + OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_A_ATTR: ( + "reactive_energy_consumption_ph_a", + t.uint48_t, + True, + ), + OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_B_ATTR: ( + "reactive_energy_consumption_ph_b", + t.uint48_t, + True, + ), + OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_C_ATTR: ( + "reactive_energy_consumption_ph_c", + t.uint48_t, + True, + ), + OWON_METERING_TOTAL_REACTIVE_ENERGY_CONSUMPTION_ATTR: ( + "total_reactive_energy_consumption", + t.uint48_t, + True, + ), + # Frequency + OWON_METERING_AC_FREQUENCY_ATTR: ("ac_frequency", t.uint8_t, True), + # Leakage + OWON_METERING_LEAKAGE_CURRENT_ATTR: ("leakage_current", t.uint24_t, True), + } + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + # Energy Consumption + self.endpoint.device.energy_consumption_ph_a_bus.add_listener(self) + self.endpoint.device.energy_consumption_ph_b_bus.add_listener(self) + self.endpoint.device.energy_consumption_ph_c_bus.add_listener(self) + self.endpoint.device.total_energy_consumption_bus.add_listener(self) + # Active power + self.endpoint.device.active_power_bus.add_listener(self) + self.endpoint.device.active_power_ph_b_bus.add_listener(self) + self.endpoint.device.active_power_ph_c_bus.add_listener(self) + self.endpoint.device.total_active_power_bus.add_listener(self) + # Reactive power + self.endpoint.device.reactive_power_bus.add_listener(self) + self.endpoint.device.reactive_power_ph_b_bus.add_listener(self) + self.endpoint.device.reactive_power_ph_c_bus.add_listener(self) + self.endpoint.device.total_reactive_power_bus.add_listener(self) + # Voltage + self.endpoint.device.rms_voltage_bus.add_listener(self) + self.endpoint.device.rms_voltage_ph_b_bus.add_listener(self) + self.endpoint.device.rms_voltage_ph_c_bus.add_listener(self) + # Active current + self.endpoint.device.active_current_bus.add_listener(self) + self.endpoint.device.active_current_ph_b_bus.add_listener(self) + self.endpoint.device.active_current_ph_c_bus.add_listener(self) + self.endpoint.device.instantaneous_active_current_bus.add_listener(self) + # Reactive Energy Consumption + self.endpoint.device.reactive_energy_consumption_ph_a_bus.add_listener(self) + self.endpoint.device.reactive_energy_consumption_ph_b_bus.add_listener(self) + self.endpoint.device.reactive_energy_consumption_ph_c_bus.add_listener(self) + self.endpoint.device.total_reactive_energy_consumption_bus.add_listener(self) + # Frequency + self.endpoint.device.ac_frequency_bus.add_listener(self) + # Leakage + self.endpoint.device.leakage_current_bus.add_listener(self) + + # Energy Consumption + def energy_consumption_ph_a_reported(self, value): + """Energy Consumption Phase A reported.""" + self._update_attribute(OWON_METERING_ENERGY_CONSUMPTION_PH_A_ATTR, value) + + def energy_consumption_ph_b_reported(self, value): + """Energy Consumption Phase B reported.""" + self._update_attribute(OWON_METERING_ENERGY_CONSUMPTION_PH_B_ATTR, value) + + def energy_consumption_ph_c_reported(self, value): + """Energy Consumption Phase C reported.""" + self._update_attribute(OWON_METERING_ENERGY_CONSUMPTION_PH_C_ATTR, value) + + def total_energy_consumption_reported(self, value): + """Total Energy Consumption reported.""" + self._update_attribute(OWON_METERING_TOTAL_ENERGY_CONSUMPTION_ATTR, value) + + # Active power + def active_power_reported(self, value): + """Active Power Phase A reported.""" + self._update_attribute(OWON_METERING_ACTIVE_POWER_PH_A_ATTR, value) + + def active_power_ph_b_reported(self, value): + """Active Power Phase B reported.""" + self._update_attribute(OWON_METERING_ACTIVE_POWER_PH_B_ATTR, value) + + def active_power_ph_c_reported(self, value): + """Active Power Phase C reported.""" + self._update_attribute(OWON_METERING_ACTIVE_POWER_PH_C_ATTR, value) + + def total_active_power_reported(self, value): + """Total Active Power reported.""" + self._update_attribute(OWON_METERING_TOTAL_ACTIVE_POWER_ATTR, value) + + # Reactive power + def reactive_power_reported(self, value): + """Reactive Power Phase A reported.""" + self._update_attribute(OWON_METERING_REACTIVE_POWER_PH_A_ATTR, value) + + def reactive_power_ph_b_reported(self, value): + """Reactive Power Phase B reported.""" + self._update_attribute(OWON_METERING_REACTIVE_POWER_PH_B_ATTR, value) + + def reactive_power_ph_c_reported(self, value): + """Reactive Power Phase C reported.""" + self._update_attribute(OWON_METERING_REACTIVE_POWER_PH_C_ATTR, value) + + def total_reactive_power_reported(self, value): + """Total Reactive Power reported.""" + self._update_attribute(OWON_METERING_TOTAL_REACTIVE_POWER_ATTR, value) + + # Voltage + def rms_voltage_reported(self, value): + """RMS Voltage Phase A reported.""" + self._update_attribute(OWON_METERING_VOLTAGE_PH_A_ATTR, value) + + def rms_voltage_ph_b_reported(self, value): + """RMS Voltage Phase B reported.""" + self._update_attribute(OWON_METERING_VOLTAGE_PH_B_ATTR, value) + + def rms_voltage_ph_c_reported(self, value): + """RMS Voltage Phase C reported.""" + self._update_attribute(OWON_METERING_VOLTAGE_PH_C_ATTR, value) + + # Active current + def active_current_reported(self, value): + """Active Current Phase A reported.""" + self._update_attribute(OWON_METERING_ACTIVE_CURRENT_PH_A_ATTR, value) + + def active_current_ph_b_reported(self, value): + """Active Current Phase B reported.""" + self._update_attribute(OWON_METERING_ACTIVE_CURRENT_PH_B_ATTR, value) + + def active_current_ph_c_reported(self, value): + """Active Current Phase C reported.""" + self._update_attribute(OWON_METERING_ACTIVE_CURRENT_PH_C_ATTR, value) + + def instantaneous_active_current_reported(self, value): + """Instantaneous Total Active Current reported.""" + self._update_attribute(OWON_METERING_TOTAL_ACTIVE_CURRENT_ATTR, value) + + # Reactive Energy Consumption + def reactive_energy_consumption_ph_a_reported(self, value): + """Reactive Energy Consumption Phase A reported.""" + self._update_attribute( + OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_A_ATTR, value + ) + + def reactive_energy_consumption_ph_b_reported(self, value): + """Reactive Energy Consumption Phase B reported.""" + self._update_attribute( + OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_B_ATTR, value + ) + + def reactive_energy_consumption_ph_c_reported(self, value): + """Reactive Energy Consumption Phase C reported.""" + self._update_attribute( + OWON_METERING_REACTIVE_ENERGY_CONSUMPTION_PH_C_ATTR, value + ) + + def total_reactive_energy_consumption_reported(self, value): + """Total Reactive Energy Consumption reported.""" + self._update_attribute( + OWON_METERING_TOTAL_REACTIVE_ENERGY_CONSUMPTION_ATTR, value + ) + + # Frequency + def ac_frequency_reported(self, value): + """AC Frequency reported.""" + self._update_attribute(OWON_METERING_AC_FREQUENCY_ATTR, value) + + # Leakage + def leakage_current_reported(self, value): + """Leakage Current reported.""" + self._update_attribute(OWON_METERING_LEAKAGE_CURRENT_ATTR, value) + + +class OwonMeteringPhaseA(LocalDataCluster, Metering): + """Owon metering Phase A - only attributes.""" + + _CONSTANT_ATTRIBUTES = METERING_CONSTANT_ATTRIBUTES.copy() + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + # Energy Consumption + self.endpoint.device.energy_consumption_ph_a_bus.add_listener(self) + self.endpoint.device.active_power_bus.add_listener(self) + + def energy_consumption_ph_a_reported(self, value): + """Energy Consumption Phase A reported.""" + self._update_attribute(Metering.AttributeDefs.current_summ_delivered.id, value) + + def active_power_reported(self, value): + """Total Active Power reported.""" + self._update_attribute(Metering.AttributeDefs.instantaneous_demand.id, value) + + +class OwonMeteringPhaseB(LocalDataCluster, Metering): + """Owon metering Phase B - only attributes.""" + + _CONSTANT_ATTRIBUTES = METERING_CONSTANT_ATTRIBUTES.copy() + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + # Energy Consumption + self.endpoint.device.energy_consumption_ph_b_bus.add_listener(self) + self.endpoint.device.active_power_ph_b_bus.add_listener(self) + + def energy_consumption_ph_b_reported(self, value): + """Energy Consumption Phase B reported.""" + self._update_attribute(Metering.AttributeDefs.current_summ_delivered.id, value) + + def active_power_ph_b_reported(self, value): + """Active Power phase B reported.""" + self._update_attribute(Metering.AttributeDefs.instantaneous_demand.id, value) + + +class OwonMeteringPhaseC(LocalDataCluster, Metering): + """Owon metering Phase C - only attributes.""" + + _CONSTANT_ATTRIBUTES = METERING_CONSTANT_ATTRIBUTES.copy() + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + # Energy Consumption + self.endpoint.device.energy_consumption_ph_c_bus.add_listener(self) + self.endpoint.device.active_power_ph_c_bus.add_listener(self) + + def energy_consumption_ph_c_reported(self, value): + """Energy Consumption Phase C reported.""" + self._update_attribute(Metering.AttributeDefs.current_summ_delivered.id, value) + + def active_power_ph_c_reported(self, value): + """Active Power phase C reported.""" + self._update_attribute(Metering.AttributeDefs.instantaneous_demand.id, value) + + +class OwonElectricalMeasurementPhaseA(LocalDataCluster, ElectricalMeasurement): + """Owon PC321 PhaseA-only attributes that can be mapped to ElectricalMeasurement cluster.""" + + # attributes = ElectricalMeasurement.attributes.copy() + # attributes.pop(ElectricalMeasurement.AttributeDefs.power_factor.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.ac_frequency.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.apparent_power.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.rms_current.id) + _CONSTANT_ATTRIBUTES = ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES.copy() + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + self.endpoint.device.rms_voltage_bus.add_listener(self) + self.endpoint.device.active_power_bus.add_listener(self) + self.endpoint.device.active_current_bus.add_listener(self) + + def rms_voltage_reported(self, value): + """Voltage Phase A reported.""" + self._update_attribute( + ElectricalMeasurement.AttributeDefs.rms_voltage.id, value + ) + + def active_power_reported(self, value): + """Active Power Phase A reported.""" + self._update_attribute( + ElectricalMeasurement.AttributeDefs.active_power.id, value + ) + + +class OwonElectricalMeasurementPhaseB(LocalDataCluster, ElectricalMeasurement): + """Owon PC321 PhaseB-only attributes that can be mapped to ElectricalMeasurement cluster.""" + + # attributes = ElectricalMeasurement.attributes.copy() + # attributes.pop(ElectricalMeasurement.AttributeDefs.power_factor.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.ac_frequency.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.apparent_power.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.rms_current.id) + _CONSTANT_ATTRIBUTES = ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES.copy() + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + self.endpoint.device.rms_voltage_ph_b_bus.add_listener(self) + self.endpoint.device.active_power_ph_b_bus.add_listener(self) + + def rms_voltage_ph_b_reported(self, value): + """Voltage Phase B reported.""" + self._update_attribute( + ElectricalMeasurement.AttributeDefs.rms_voltage.id, value + ) + + def active_power_ph_b_reported(self, value): + """Active Power Phase B reported.""" + self._update_attribute( + ElectricalMeasurement.AttributeDefs.active_power.id, value + ) + + +class OwonElectricalMeasurementPhaseC(LocalDataCluster, ElectricalMeasurement): + """Owon PC321 PhaseC-only attributes that can be mapped to ElectricalMeasurement cluster.""" + + # attributes = ElectricalMeasurement.attributes.copy() + # attributes.pop(ElectricalMeasurement.AttributeDefs.power_factor.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.ac_frequency.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.apparent_power.id) + # attributes.pop(ElectricalMeasurement.AttributeDefs.rms_current.id) + _CONSTANT_ATTRIBUTES = ELECTRICAL_MEASUREMENT_CONSTANT_ATTRIBUTES.copy() + + def __init__(self, *args, **kwargs): + """Init.""" + super().__init__(*args, **kwargs) + self.endpoint.device.rms_voltage_ph_c_bus.add_listener(self) + self.endpoint.device.active_power_ph_c_bus.add_listener(self) + + def rms_voltage_ph_c_reported(self, value): + """Voltage Phase C reported.""" + self._update_attribute( + ElectricalMeasurement.AttributeDefs.rms_voltage.id, value + ) + + def active_power_ph_c_reported(self, value): + """Active Power Phase C reported.""" + self._update_attribute( + ElectricalMeasurement.AttributeDefs.active_power.id, value + ) + + +class Owon_PC321(CustomDevice): + """OwonPC321 Custom Device.""" + + def __init__(self, *args, **kwargs): + """Init.""" + # Active Consumption + self.energy_consumption_ph_a_bus = Bus() + self.energy_consumption_ph_b_bus = Bus() + self.energy_consumption_ph_c_bus = Bus() + self.total_energy_consumption_bus = Bus() + # Active power + self.active_power_bus = Bus() + self.active_power_ph_b_bus = Bus() + self.active_power_ph_c_bus = Bus() + self.total_active_power_bus = Bus() + # Reactive power + self.reactive_power_bus = Bus() + self.reactive_power_ph_b_bus = Bus() + self.reactive_power_ph_c_bus = Bus() + self.total_reactive_power_bus = Bus() + # Voltage + self.rms_voltage_bus = Bus() + self.rms_voltage_ph_b_bus = Bus() + self.rms_voltage_ph_c_bus = Bus() + # Active current + self.active_current_bus = Bus() + self.active_current_ph_b_bus = Bus() + self.active_current_ph_c_bus = Bus() + self.instantaneous_active_current_bus = Bus() + # Reactive consumption + self.reactive_energy_consumption_ph_a_bus = Bus() + self.reactive_energy_consumption_ph_b_bus = Bus() + self.reactive_energy_consumption_ph_c_bus = Bus() + self.total_reactive_energy_consumption_bus = Bus() + # Frequency + self.ac_frequency_bus = Bus() + # Leakage + self.leakage_current_bus = Bus() + + super().__init__(*args, **kwargs) + + signature = { + MODELS_INFO: [(OWON, OWON_PC321)], + ENDPOINTS: { + # device_version="0x000d" + # input_clusters=["0x0000", "0x0003", "0x0702"] + # output_clusters=["0x0003"] + 1: { + PROFILE_ID: zha.PROFILE_ID, + DEVICE_TYPE: zha.DeviceType.CONSUMPTION_AWARENESS_DEVICE, + INPUT_CLUSTERS: [ + Basic.cluster_id, + Identify.cluster_id, + Metering.cluster_id, + ], + OUTPUT_CLUSTERS: [Identify.cluster_id], + }, + }, + } + + replacement = { + ENDPOINTS: { + 1: { + PROFILE_ID: zha.PROFILE_ID, + DEVICE_TYPE: zha.DeviceType.CONSUMPTION_AWARENESS_DEVICE, + INPUT_CLUSTERS: [ + Basic.cluster_id, + Identify.cluster_id, + OwonMetering, + OwonManufacturerSpecific, + ], + OUTPUT_CLUSTERS: [Identify.cluster_id], + }, + 11: { + PROFILE_ID: zha.PROFILE_ID, + DEVICE_TYPE: zha.DeviceType.CONSUMPTION_AWARENESS_DEVICE, + INPUT_CLUSTERS: [ + OwonMeteringPhaseA, + OwonElectricalMeasurementPhaseA, + ], + OUTPUT_CLUSTERS: [], + }, + 12: { + PROFILE_ID: zha.PROFILE_ID, + DEVICE_TYPE: zha.DeviceType.CONSUMPTION_AWARENESS_DEVICE, + INPUT_CLUSTERS: [ + OwonMeteringPhaseB, + OwonElectricalMeasurementPhaseB, + ], + OUTPUT_CLUSTERS: [], + }, + 13: { + PROFILE_ID: zha.PROFILE_ID, + DEVICE_TYPE: zha.DeviceType.CONSUMPTION_AWARENESS_DEVICE, + INPUT_CLUSTERS: [ + OwonMeteringPhaseC, + OwonElectricalMeasurementPhaseC, + ], + OUTPUT_CLUSTERS: [], + }, + }, + }