diff options
| -rw-r--r-- | src/home/mqtt/payload/base_payload.py | 44 | ||||
| -rw-r--r-- | src/home/mqtt/payload/inverter.py | 14 |
2 files changed, 27 insertions, 31 deletions
diff --git a/src/home/mqtt/payload/base_payload.py b/src/home/mqtt/payload/base_payload.py index c9ec907..108e0c0 100644 --- a/src/home/mqtt/payload/base_payload.py +++ b/src/home/mqtt/payload/base_payload.py @@ -1,7 +1,8 @@ import abc import struct +import re -from typing import Generic, TypeVar +from typing import Optional, Tuple class MQTTPayload(abc.ABC): @@ -20,15 +21,9 @@ class MQTTPayload(abc.ABC): bf_progress = 0 for field, field_type in self.__class__.__annotations__.items(): - field_type_origin = None - if hasattr(field_type, '__extra__') or hasattr(field_type, '__origin__'): - try: - field_type_origin = field_type.__extra__ - except AttributeError: - field_type_origin = field_type.__origin__ - - if field_type_origin is not None and issubclass(field_type_origin, MQTTPayloadBitField): - n, s, b = field_type.__args__ + bfp = _bit_field_params(field_type) + if bfp: + n, s, b = bfp if n != bf_number: if bf_number != -1: args.append(bf_arg) @@ -61,15 +56,9 @@ class MQTTPayload(abc.ABC): bf_progress = 0 for field, field_type in cls.__annotations__.items(): - field_type_origin = None - if hasattr(field_type, '__extra__') or hasattr(field_type, '__origin__'): - try: - field_type_origin = field_type.__extra__ - except AttributeError: - field_type_origin = field_type.__origin__ - - if field_type_origin is not None and issubclass(field_type_origin, MQTTPayloadBitField): - n, s, b = field_type.__args__ + bfp = _bit_field_params(field_type) + if bfp: + n, s, b = bfp if n != bf_number: bf_number = n bf_progress = 0 @@ -86,6 +75,7 @@ class MQTTPayload(abc.ABC): else: kwargs[field] = cls._unpack_field(field, data[i]) i += 1 + return cls(**kwargs) def _pack_field(self, name): @@ -120,10 +110,16 @@ class MQTTPayloadCustomField(abc.ABC): pass -NT = TypeVar('NT') # number of bit field -ST = TypeVar('ST') # size in bytes -BT = TypeVar('BT') # size in bits of particular value +def bit_field(seq_no: int, total_bits: int, bits: int): + return type(f'MQTTPayloadBitField_{seq_no}_{total_bits}_{bits}', (object,), { + 'seq_no': seq_no, + 'total_bits': total_bits, + 'bits': bits + }) -class MQTTPayloadBitField(int, Generic[NT, ST, BT]): - pass +def _bit_field_params(cl) -> Optional[Tuple[int, ...]]: + match = re.match(r'MQTTPayloadBitField_(\d+)_(\d+)_(\d)$', cl.__name__) + if match is not None: + return tuple([int(match.group(i)) for i in range(1, 4)]) + return None diff --git a/src/home/mqtt/payload/inverter.py b/src/home/mqtt/payload/inverter.py index b3f4edd..1d4099c 100644 --- a/src/home/mqtt/payload/inverter.py +++ b/src/home/mqtt/payload/inverter.py @@ -1,6 +1,6 @@ import struct -from .base_payload import MQTTPayload, MQTTPayloadBitField +from .base_payload import MQTTPayload, bit_field from typing import Tuple _mult_10 = lambda n: int(n*10) @@ -57,12 +57,12 @@ class Status(MQTTPayload): pv2_input_voltage: float # H - mppt1_charger_status: MQTTPayloadBitField[0, 16, 2] - mppt2_charger_status: MQTTPayloadBitField[0, 16, 2] - battery_power_direction: MQTTPayloadBitField[0, 16, 2] - dc_ac_power_direction: MQTTPayloadBitField[0, 16, 2] - line_power_direction: MQTTPayloadBitField[0, 16, 2] - load_connected: MQTTPayloadBitField[0, 16, 1] + mppt1_charger_status: bit_field(0, 16, 2) + mppt2_charger_status: bit_field(0, 16, 2) + battery_power_direction: bit_field(0, 16, 2) + dc_ac_power_direction: bit_field(0, 16, 2) + line_power_direction: bit_field(0, 16, 2) + load_connected: bit_field(0, 16, 1) class Generation(MQTTPayload): |