Inverter Voltronic

Инструкция

Файл:Axpert V PF1 manual.pdf

Снимать данные с линукса

• https://github.com/jblance/mpp-solar

mpp-solar -p /dev/hidraw0 -c QPIGS   -I -P PI30

но работает лучше как минимум у меня через ком-порт и /dev/ttyUSB0
Но для этого нужен переходник USB -> COM
Если переходников несколько то лучше указывать by-id - /dev/serial/by-id/usb-1a86_USB2.0-Ser_-if00-port0

NUT

так как я хотел что бы инвертор прикинулся нормальным UPS а не вот это вот все то набросал простой скрипт для снятия данных (драйвер встроенный в NUT не заработал, написать свой на основе скрипта я не осилил)
Весь код написан на скорую руку, с кучей хардкода, так как надо было прям сейчас, а передывать пока нет времени
Логика работы такая:

• скрипт mpp_nut_bridge.py (через systemd unit) работает в вечном цикле и складывает результат в файл /var/lib/nut/mpp.state
• NUT умеет читать данные из внешнего файла через driver = dummy-ups

[mpp]
    driver = dummy-ups
    port = /var/lib/nut/mpp.state
    desc = "MPP via mppsolar"

• для того что бы отдавать данные на zabbix используется бридж в SNMP

mpp_nut_bridge.py

Кода гавно конечно )

#!/usr/bin/env python3

import subprocess
import time
import tempfile
import os
import json
import logging

class CustomFormatter(logging.Formatter):

    grey = "\x1b[38;20m"
    yellow = "\x1b[33;20m"
    red = "\x1b[31;20m"
    bold_red = "\x1b[31;1m"
    reset = "\x1b[0m"
    format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"

    FORMATS = {
        logging.DEBUG: grey + format + reset,
        logging.INFO: grey + format + reset,
        logging.WARNING: yellow + format + reset,
        logging.ERROR: red + format + reset,
        logging.CRITICAL: bold_red + format + reset
    }

    def format(self, record):
        log_fmt = self.FORMATS.get(record.levelno)
        formatter = logging.Formatter(log_fmt)
        return formatter.format(record)

STATE_FILE = "/var/lib/nut/mpp.state"

LOGLEVEL = os.environ.get('LOGLEVEL', 'INFO').upper()

logger = logging.getLogger("mpp_nut_bridge")
logger.setLevel(LOGLEVEL)

# create console handler with a higher log level
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)

ch.setFormatter(CustomFormatter())

logger.addHandler(ch)

# Команда mppsolar
# /etc/nut/mpp-solar -p /dev/ttyUSB1 -c QPIGS   -I -P PI30  -o json | jq '.'
DEV="/dev/serial/by-id/usb-1a86_USB2.0-Ser_-if00-port0"
#DEV="/dev/hidraw1"
MPPSOLAR_CMD = [
    "/usr/local/virtualenvs/mppsolar/bin/python3",
    "/etc/nut/mpp-solar",
    "-p", DEV,
    "-P", "PI30",
    "-c", "QPIGS",
    "-o", "json",
]


def write_state(vars_dict):
    """Атомарная запись файла состояния для dummy-ups"""
    fd, tmp = tempfile.mkstemp()
    with os.fdopen(fd, "w") as f:
        for k, v in vars_dict.items():
            f.write(f"{k}: {v}\n")
    logger.info(f"Replacing {STATE_FILE}")
    os.replace(tmp, STATE_FILE)

def build_ups_status(data):
    flags = []

    # UPS выключен
    if not data.get("is_switched_on", 1):
        return "OFF"

    ac_in = data.get("ac_input_voltage", 0.0)
    batt_cap = data.get("battery_capacity", 0)
    charging = bool(data.get("is_charging_on", 0))
    load_on = bool(data.get("is_load_on", 0))

    if ac_in > 100:
        flags.append("OL")      # On Line
    else:
        flags.append("OB")      # On Battery

    if batt_cap < 10:
        flags.append("LB")      # Low Battery

    if charging:
        flags.append("CHRG")
    elif load_on:
        flags.append("DISCHRG")

    if not flags:
        return "UNKNOWN"

    return " ".join(flags)

def poll_once():
    attempts=3
    while attempts>0:
        try:
            # Запускаем mppsolar и читаем JSON
            res = subprocess.run(
                MPPSOLAR_CMD,
                capture_output=True,
                text=True,
                check=True,
            )
            logger.info(f"STDOUT: {res.stdout}")
            data = json.loads(res.stdout)
            error = data.get("error", False)
            logger.info(f"DATA: {data} Error: {error}")
            # Собираем переменные NUT
            if error:
                logger.info("ERROR DETECTED")
            vars_dict = {
                "battery.charge": int(data.get("battery_capacity")),
                "battery.voltage": float(data.get("battery_voltage")),
                "input.voltage": float(data.get("ac_input_voltage")),
                "input.frequency": float(data.get("ac_input_frequency")),
                "output.voltage": float(data.get("ac_output_voltage")),
                "output.frequency": float(data.get("ac_output_frequency")),
                "ups.load": int(data.get("ac_output_load")),
                "ups.power": int(data.get("ac_output_apparent_power")),
                "ups.realpower": int(data.get("ac_output_active_power")),
                "inverter_heat_sink_temperature": int(data.get("inverter_heat_sink_temperature")),
                "ups.status": build_ups_status(data),
                # Можно добавить свои:
                # "device.mfr": "MPP Solar",
                # "device.model": "PI30",
            }
            write_state(vars_dict)
            return
        except Exception as e:
            logger.info(f"Error during data collection: {e}")
            logger.info(f"Sleep 30 sec. before next attempt")
            time.sleep(30)
            attempts = attempts - 1
    # If exited on attempts < 0
    vars_dict = {
        "battery.charge": 0,
        "battery.voltage": 0,
        "input.voltage": 0,
        "input.frequency": 0,
        "output.voltage": 0,
        "output.frequency": 0,
        "ups.load": 0,
        "ups.power": 0,
        "ups.realpower": 0,
        "inverter_heat_sink_temperature": 0,
        "ups.status": "OFF"
    }
    write_state(vars_dict)


def main():
    logger.info("Starting")
    SLEEP = 20
    while True:
        try:
            poll_once()
        except Exception as e:
            # При ошибке помечаем UPS как OFF
            try:
                logger.info(f"Error during poll: {e}")
                write_state({"ups.status": "OFF"})
            except Exception:
                pass
        logger.info(f"Sleeping {SLEEP} sec brfore next check")
        time.sleep(SLEEP)  # опрос каждые 5 секунд

if __name__ == "__main__":
    main()

mpp-nut-bridge.service

• /usr/local/virtualenvs/mppsolar/bin/python3 - этот путь к virtualenv

[Unit]
Description=MPP Solar -> NUT bridge
After=network.target

[Service]
User=nut
Group=nut
ExecStart=/usr/local/virtualenvs/mppsolar/bin/python3 /etc/nut/mpp_nut_bridge.py
Restart=always
RestartSec=3


[Install]
WantedBy=multi-user.target

snmpd.conf

Часть конфига ответвенная за "проброс" запросов к snmpd
Тут три ветки

• .1.3.6.1.2.1.33 "Стандартный" MIB - просто для теста, у меня в заббиксе он не используется
• .1.3.6.1.4.1.318 MIB специфичный для APC - все части взяты из конфига темплейта zabbix, возможно в оригинальном MIB больше данных
• .1.3.6.1.4.1.418 - ветка выбрана от фонаря и используется для мониторинга BMS батареи (это отдельная задача - мониторинг батареи Daly BMS)
• ups-snmp-passpersist.py имя файла который дергать на запрос (ему передается тип изапроса GET/GETNEXT и OID, другие типы игнорируем)

pass .1.3.6.1.2.1.33 /etc/nut/ups-snmp-passpersist.py
pass .1.3.6.1.4.1.318 /etc/nut/ups-snmp-passpersist.py
pass .1.3.6.1.4.1.418 /etc/nut/batt-dale-snmp.py

По сути этот конфиг на каждый запрос вызывает скрипт с параметрами (хорошо бы конечно сделать постоянно висящего демона что бы не форкать на каждый запрос но пока и так сойдет)