The DTSU666 bidirectional, three-phase electricity meter and data retrieval via Modbus on the ESP32

#!/usr/bin/env python3
"""
Program do odczytu danych z miernika energii elektrycznej DTSU666 przez Modbus RTU
Wykorzystuje bibliotekę minimalmodbus
"""

import minimalmodbus
import time
import sys
import logging

# Konfiguracja logowania
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class DTSU666:
    """Klasa do obsługi miernika DTSU666 przez Modbus RTU"""

    def __init__(self, port='/dev/ttyUSB0', slave_address=1, baudrate=9600, timeout=1):
        """
        Inicjalizacja połączenia z miernikiem

        Args:
            port: Port szeregowy (np. '/dev/ttyUSB0', 'COM3')
            slave_address: Adres Modbus urządzenia (domyślnie 1)
            baudrate: Prędkość transmisji (domyślnie 9600)
            timeout: Timeout w sekundach
        """
        self.port = port
        self.slave_address = slave_address
        self.baudrate = baudrate
        self.timeout = timeout

        # Inicjalizacja instrumentu Modbus
        self.instrument = minimalmodbus.Instrument(port, slave_address)
        self.instrument.serial.baudrate = baudrate
        self.instrument.serial.bytesize = 8
        self.instrument.serial.parity = minimalmodbus.serial.PARITY_NONE
        self.instrument.serial.stopbits = 1
        self.instrument.serial.timeout = timeout

        # Ustawienie trybu komunikacji (RTU)
        self.instrument.mode = minimalmodbus.MODE_RTU

        # Mapowanie rejestrów z dokumentacji
        self.registers = {
            'Uc': {'address': 0x200A, 'name': 'Napięcie fazy C', 'unit': 'V', 'scale': 0.001},
            'Ia': {'address': 0x200C, 'name': 'Prąd fazy A', 'unit': 'A', 'scale': 0.001},
            'Ib': {'address': 0x200E, 'name': 'Prąd fazy B', 'unit': 'A', 'scale': 0.001},
            'Ic': {'address': 0x2010, 'name': 'Prąd fazy C', 'unit': 'A', 'scale': 0.001},
            'Pt': {'address': 0x2012, 'name': 'Moc czynna całkowita', 'unit': 'W', 'scale': 0.1},
            'Pa': {'address': 0x2014, 'name': 'Moc czynna fazy A', 'unit': 'W', 'scale': 0.1},
            'Pb': {'address': 0x2016, 'name': 'Moc czynna fazy B', 'unit': 'W', 'scale': 0.1},
            'Pc': {'address': 0x2018, 'name': 'Moc czynna fazy C', 'unit': 'W', 'scale': 0.1},
            'Qt': {'address': 0x201A, 'name': 'Moc bierna całkowita', 'unit': 'var', 'scale': 0.1},
            'Qa': {'address': 0x201C, 'name': 'Moc bierna fazy A', 'unit': 'var', 'scale': 0.1},
            'Qb': {'address': 0x201E, 'name': 'Moc bierna fazy B', 'unit': 'var', 'scale': 0.1},
            'Qc': {'address': 0x2020, 'name': 'Moc bierna fazy C', 'unit': 'var', 'scale': 0.1},
            'Pft': {'address': 0x202A, 'name': 'Współczynnik mocy całkowity', 'unit': '', 'scale': 0.001},
            'Pfa': {'address': 0x202C, 'name': 'Współczynnik mocy fazy A', 'unit': '', 'scale': 0.001},
        }

        logger.info(f"Zainicjalizowano DTSU666 na porcie {port}, adres: {slave_address}, baudrate: {baudrate}")

    def read_float(self, register_address, scale=1.0):
        """
        Odczyt wartości zmiennoprzecinkowej z rejestru

        Args:
            register_address: Adres rejestru (hex)
            scale: Współczynnik skali

        Returns:
            float: Odczytana wartość
        """
        try:
            # Odczyt 2 rejestrów (4 bajty) jako float
            value = self.instrument.read_float(register_address, 3, 2)
            return value * scale
        except Exception as e:
            logger.error(f"Błąd odczytu rejestru 0x{register_address:04X}: {e}")
            return None

    def read_all_data(self):
        """
        Odczyt wszystkich dostępnych danych z miernika

        Returns:
            dict: Słownik z odczytanymi wartościami
        """
        data = {}

        logger.info("Rozpoczynanie odczytu danych z DTSU666...")

        for key, reg_info in self.registers.items():
            address = reg_info['address']
            scale = reg_info['scale']
            name = reg_info['name']
            unit = reg_info['unit']

            value = self.read_float(address, scale)

            if value is not None:
                data[key] = value
                # Wyświetlenie odczytanej wartości
                if unit:
                    logger.info(f"{name}: {value:.3f} {unit}")
                else:
                    logger.info(f"{name}: {value:.3f}")
            else:
                data[key] = None
                logger.warning(f"Nie udało się odczytać {name}")

        return data

    def read_specific_data(self, parameters):
        """
        Odczyt wybranych parametrów

        Args:
            parameters: Lista nazw parametrów do odczytu

        Returns:
            dict: Słownik z odczytanymi wartościami
        """
        data = {}

        for param in parameters:
            if param in self.registers:
                reg_info = self.registers[param]
                value = self.read_float(reg_info['address'], reg_info['scale'])

                if value is not None:
                    data[param] = value
                    logger.info(f"{reg_info['name']}: {value:.3f} {reg_info['unit']}")
                else:
                    data[param] = None
                    logger.warning(f"Nie udało się odczytać {reg_info['name']}")
            else:
                logger.error(f"Nieznany parametr: {param}")
                data[param] = None

        return data

    def format_output(self, data):
        """
        Formatowanie danych do czytelnego wyjścia

        Args:
            data: Słownik z danymi

        Returns:
            str: Sformatowany string
        """
        output = []
        output.append("=" * 60)
        output.append(f"DTSU666 - Odczyt danych z {time.strftime('%Y-%m-%d %H:%M:%S')}")
        output.append("=" * 60)

        for key, value in data.items():
            if key in self.registers and value is not None:
                reg_info = self.registers[key]
                name = reg_info['name']
                unit = reg_info['unit']
                output.append(f"{name:30}: {value:>10.3f} {unit}")
            elif key in self.registers:
                output.append(f"{self.registers[key]['name']:30}: {'BRAK DANYCH':>10}")

        output.append("=" * 60)
        return "\n".join(output)


def main():
    """Funkcja główna"""

    # Konfiguracja połączenia - dostosuj do swojego systemu
    # Linux: /dev/ttyUSB0 lub /dev/ttyS0
    # Windows: COM1, COM2, itp.
    PORT = '/dev/ttyUSB0'  # Zmień na odpowiedni port
    SLAVE_ADDRESS = 1       # Adres Modbus miernika (domyślnie 1)
    BAUDRATE = 9600         # Prędkość transmisji

    # Utworzenie instancji miernika
    meter = DTSU666(port=PORT, slave_address=SLAVE_ADDRESS, baudrate=BAUDRATE)

    try:
        # Pojedynczy odczyt wszystkich danych
        logger.info("=== POJEDYNCZY ODCZYT WSZYSTKICH DANYCH ===")
        data = meter.read_all_data()
        print(meter.format_output(data))

        # Przykład odczytu wybranych parametrów
        logger.info("\n=== ODCZYT WYBRANYCH PARAMETRÓW ===")
        selected_params = ['Uc', 'Ia', 'Ib', 'Ic', 'Pt', 'Pft']
        selected_data = meter.read_specific_data(selected_params)
        print(meter.format_output(selected_data))

        # Przykład ciągłego odczytu (odkomentuj jeśli potrzebujesz)
        """
        logger.info("\n=== ROZPOCZYNAMNIE CIĄGŁEGO ODCZYTU (Ctrl+C aby zatrzymać) ===")
        while True:
            try:
                data = meter.read_all_data()
                print(meter.format_output(data))
                print("\n" + "-" * 60)
                time.sleep(2)  # Odczyt co 2 sekundy
            except KeyboardInterrupt:
                logger.info("Zatrzymano przez użytkownika")
                break
            except Exception as e:
                logger.error(f"Błąd podczas ciągłego odczytu: {e}")
                time.sleep(5)
        """

    except Exception as e:
        logger.error(f"Błąd główny: {e}")
        return 1

    return 0


if __name__ == "__main__":
    # Sprawdzenie czy biblioteka minimalmodbus jest zainstalowana
    try:
        import minimalmodbus
    except ImportError:
        print("Brak biblioteki minimalmodbus. Instaluj: pip install minimalmodbus")
        sys.exit(1)

    sys.exit(main())