Home-Assistant, Eastron SDM72, SDM630 modbus config generator

generator.py

#!/usr/bin/env python3
"""
Modbus RTUoverTCP Configuration Generator for Home Assistant

This script generates YAML configuration for Modbus RTUoverTCP integration
in Home Assistant, supporting both 1-phase (SDM120) and 3-phase (SDM630) meters.
"""

import yaml
from typing import List, Dict, Any


# ============================================================================
# CONFIGURATION
# ============================================================================

MODBUS_CONFIG = {
    # Connection settings for the Modbus RTUoverTCP adapter
    "connection": {
        "type": "rtuovertcp",
        "host": "192.168.1.201",
        "port": 502,
        "delay": 10,  # default = 0
        "timeout": 10,  # default = 5
        "message_wait_milliseconds": 100  # default = 30 for serial
    },
    
    # Device name for the Home Assistant configuration
    "device_name": "Meter 200",
    
    # List of Modbus slave addresses for 1-phase meters (SDM120)
    # Note: 1-phase meters only generate L1 sensors (no L2 or L3)
    "single_phase_meters": [1, 2, 3, 4, 5, 6, 7, 8],
    
    # List of Modbus slave addresses for 3-phase meters (SDM630)
    # Note: 3-phase meters generate all three phases (L1, L2, L3)
    "three_phase_meters": [200]
}


# ============================================================================
# SENSOR DEFINITIONS
# ============================================================================

def create_sensor(
    name: str,
    slave: int,
    address: int,
    precision: int,
    data_type: str,
    unit_of_measurement: str = None,
    device_class: str = None,
    state_class: str = None,
    input_type: str = "input",
    scale: float = None
) -> Dict[str, Any]:
    """Create a sensor configuration dictionary."""
    sensor = {
        "name": name,
        "slave": slave,
        "address": address,
        "input_type": input_type,
        "precision": precision,
        "data_type": data_type,
    }
    
    if unit_of_measurement:
        sensor["unit_of_measurement"] = unit_of_measurement
    
    if device_class:
        sensor["device_class"] = device_class
    
    if state_class:
        sensor["state_class"] = state_class
    
    if scale is not None:
        sensor["scale"] = scale
    
    # Generate unique_id
    sensor["unique_id"] = f"modbus_{slave}_address_{address}"
    
    return sensor


# ============================================================================
# METER SENSOR GENERATORS
# ============================================================================

def generate_meter_sensors(meter_id: int, number_of_phases: int) -> List[Dict[str, Any]]:
    """
    Generate sensor configuration for a meter.
    
    Args:
        meter_id: Modbus slave address of the meter
        number_of_phases: Number of phases (1 for SDM120, 3 for SDM630)
                         Determines which phase sensors to generate.
                         - number_of_phases=1: Only L1 sensors (no L2 or L3)
                         - number_of_phases=3: All three phases (L1, L2, L3)
    
    Returns:
        List of sensor configuration dictionaries
    """
    sensors = []
    phase_map = {1: "L1", 2: "L2", 3: "L3"}
    phases_to_generate = list(range(1, number_of_phases + 1))  # [1] or [1, 2, 3]
    
    # Voltage sensors - generate based on number_of_phases
    for phase_num in phases_to_generate:
        phase = phase_map[phase_num]
        sensors.append(create_sensor(
            name=f"Meter_{meter_id}_volts_{phase}/N",
            slave=meter_id,
            address=(phase_num - 1) * 2,  # 0, 2, 4
            precision=2,
            data_type="float32",
            unit_of_measurement="V",
            device_class="voltage",
            state_class="measurement"
        ))
    
    # Average voltage
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_volts_average",
        slave=meter_id,
        address=42,
        precision=2,
        data_type="float32",
        unit_of_measurement="V",
        device_class="voltage",
        state_class="measurement"
    ))
    
    # Current - generate based on number_of_phases
    for phase_num in phases_to_generate:
        phase = phase_map[phase_num]
        sensors.append(create_sensor(
            name=f"Meter_{meter_id}_{phase}_current",
            slave=meter_id,
            address=4 + phase_num * 2,  # 6, 8, 10
            precision=3,
            data_type="float32",
            unit_of_measurement="A",
            device_class="current",
            state_class="measurement"
        ))
    
    # Power - generate based on number_of_phases
    for phase_num in phases_to_generate:
        phase = phase_map[phase_num]
        sensors.append(create_sensor(
            name=f"Meter_{meter_id}_{phase}_power",
            slave=meter_id,
            address=10 + phase_num * 2,  # 12, 14, 16
            precision=3,
            data_type="float32",
            unit_of_measurement="W",
            device_class="power",
            state_class="measurement",
            scale=1 if phase_num == 1 else None
        ))
    
    # Apparent power - generate based on number_of_phases
    for phase_num in phases_to_generate:
        phase = phase_map[phase_num]
        sensors.append(create_sensor(
            name=f"Meter_{meter_id}_{phase}_power_apparent",
            slave=meter_id,
            address=16 + phase_num * 2,  # 18, 20, 22
            precision=2,
            data_type="float32",
            unit_of_measurement="VA",
            device_class="apparent_power",
            state_class="measurement"
        ))
    
    # Reactive power - generate based on number_of_phases
    for phase_num in phases_to_generate:
        phase = phase_map[phase_num]
        sensors.append(create_sensor(
            name=f"Meter_{meter_id}_{phase}_power_reactive",
            slave=meter_id,
            address=22 + phase_num * 2,  # 24, 26, 28
            precision=2,
            data_type="float32",
            unit_of_measurement="var",
            device_class="reactive_power",
            state_class="measurement"
        ))
    
    # Power factor - generate based on number_of_phases
    for phase_num in phases_to_generate:
        phase = phase_map[phase_num]
        sensors.append(create_sensor(
            name=f"Meter_{meter_id}_{phase}_power_factor",
            slave=meter_id,
            address=28 + phase_num * 2,  # 30, 32, 34
            precision=2,
            data_type="float32",
            device_class="power_factor",
            state_class="measurement"
        ))
    
    # Total measurements (same for all meters)
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_total_current",
        slave=meter_id,
        address=48,
        precision=3,
        data_type="float32",
        unit_of_measurement="A",
        device_class="current"
    ))
    
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_total_power",
        slave=meter_id,
        address=52,
        precision=2,
        data_type="float32",
        unit_of_measurement="W",
        device_class="power",
        state_class="measurement"
    ))
    
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_total_power_apparent",
        slave=meter_id,
        address=56,
        precision=2,
        data_type="float32",
        unit_of_measurement="VA",
        device_class="apparent_power",
        state_class="measurement"
    ))
    
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_total_power_reactive",
        slave=meter_id,
        address=60,
        precision=2,
        data_type="float32",
        unit_of_measurement="var",
        device_class="reactive_power",
        state_class="measurement"
    ))
    
    # Frequency
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_frequency",
        slave=meter_id,
        address=70,
        precision=2,
        data_type="float32",
        unit_of_measurement="Hz",
        state_class="measurement",
        device_class="frequency"
    ))
    
    # Energy meters
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_import_active",
        slave=meter_id,
        address=72,
        precision=2,
        data_type="float32",
        unit_of_measurement="kWh",
        device_class="energy",
        state_class="total_increasing"
    ))
    
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_export_active",
        slave=meter_id,
        address=74,
        precision=2,
        data_type="float32",
        unit_of_measurement="kWh",
        device_class="energy",
        state_class="total_increasing"
    ))
    
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_import_reactive",
        slave=meter_id,
        address=78,
        precision=2,
        data_type="float32",
        unit_of_measurement="kvarh",
        device_class="reactive_energy",
        state_class="total_increasing"
    ))
    
    sensors.append(create_sensor(
        name=f"Meter_{meter_id}_export_reactive",
        slave=meter_id,
        address=80,
        precision=2,
        data_type="float32",
        unit_of_measurement="kvarh",
        device_class="reactive_energy",
        state_class="total_increasing"
    ))
    
    return sensors


# ============================================================================
# YAML GENERATION
# ============================================================================

def generate_modbus_config(config: Dict[str, Any] = None) -> Dict[str, Any]:
    """
    Generate the complete Modbus RTUoverTCP configuration for Home Assistant.
    
    Args:
        config: Configuration dictionary. If None, uses MODBUS_CONFIG.
    
    Returns:
        Dictionary containing the full configuration structure.
    """
    if config is None:
        config = MODBUS_CONFIG
    
    # Collect all sensors
    all_sensors = []
    
    # Generate sensors for 1-phase meters
    for meter_id in config["single_phase_meters"]:
        sensors = generate_meter_sensors(meter_id, number_of_phases=1)
        all_sensors.extend(sensors)
    
    # Generate sensors for 3-phase meters
    for meter_id in config["three_phase_meters"]:
        sensors = generate_meter_sensors(meter_id, number_of_phases=3)
        all_sensors.extend(sensors)
    
    # Build the configuration structure
    connection = config["connection"]
    device_name = config.get("device_name", "Modbus Adapter")
    
    yaml_config = [
        {
            "name": device_name,
            "type": connection["type"],
            "host": connection["host"],
            "port": connection["port"],
            "delay": connection["delay"],
            "timeout": connection["timeout"],
            "message_wait_milliseconds": connection["message_wait_milliseconds"],
            "sensors": all_sensors
        }
    ]
    
    return yaml_config


def save_yaml_config(output_file: str = "modbus.yaml", config: Dict[str, Any] = None) -> None:
    """
    Generate and save the Modbus configuration to a YAML file.
    
    Args:
        output_file: Path to the output YAML file.
        config: Configuration dictionary. If None, uses MODBUS_CONFIG.
    """
    yaml_config = generate_modbus_config(config)
    
    with open(output_file, 'w') as f:
        yaml.dump(yaml_config, f, default_flow_style=False, sort_keys=False, allow_unicode=True)
    
    print(f"✓ Generated Modbus configuration: {output_file}")
    print(f"  - Single-phase meters: {len(config['single_phase_meters']) if config else len(MODBUS_CONFIG['single_phase_meters'])}")
    print(f"  - Three-phase meters: {len(config['three_phase_meters']) if config else len(MODBUS_CONFIG['three_phase_meters'])}")
    print(f"  - Total sensors: {len(yaml_config[0]['sensors'])}")


# ============================================================================
# MAIN
# ============================================================================

def main():
    """Main entry point."""
    print("Modbus RTUoverTCP Configuration Generator")
    print("=" * 50)
    
    # Generate and save configuration
    save_yaml_config("modbus.yaml", MODBUS_CONFIG)
    
    print("\n✓ Configuration generated successfully!")


if __name__ == "__main__":
    main()