40-clients/ucepsa/edge-oee-demo/revpi/edge-oee/beacon_manual_test.py

#!/usr/bin/env python3
import json
import sys
import time
from pathlib import Path

from pymodbus.client import ModbusTcpClient

UNIT = 1

DO0_BLUE_COIL = 16
DO1_RED_COIL = 17

WISE_NODES_FILE = Path("wise_nodes.json")


def load_nodes():
    data = json.loads(WISE_NODES_FILE.read_text())
    nodes = data.get("nodes", [])

    result = {}
    for node in nodes:
        machine_id = node.get("machine_id")
        host = node.get("wise_ip") or node.get("wise_host") or node.get("host") or node.get("ip")

        if machine_id and host:
            result[machine_id] = {
                "machine_id": machine_id,
                "source_node": node.get("source_node", ""),
                "host": host,
            }

    return result


def write_coil_checked(client, address, value):
    rr = client.write_coil(address=address, value=value, device_id=UNIT)
    if rr.isError():
        raise RuntimeError(f"Error escribiendo coil {address}: {rr}")


def read_outputs(client):
    rr = client.read_coils(address=DO0_BLUE_COIL, count=2, device_id=UNIT)
    if rr.isError():
        raise RuntimeError(f"Error leyendo coils DO: {rr}")

    do0 = bool(rr.bits[0])
    do1 = bool(rr.bits[1])

    blue_on = do0
    red_on = not do1

    return do0, do1, blue_on, red_on


def set_beacon(client, blue_on, red_on):
    # DO0 gobierna azul directamente.
    write_coil_checked(client, DO0_BLUE_COIL, bool(blue_on))

    # DO1 gobierna roja invertida por contacto NC.
    # red_on=True  -> DO1 False
    # red_on=False -> DO1 True
    write_coil_checked(client, DO1_RED_COIL, not bool(red_on))


def print_state(client, label):
    do0, do1, blue_on, red_on = read_outputs(client)
    print(
        f"{label}: "
        f"DO0_BLUE={do0} DO1_RED={do1} "
        f"=> blue_on={blue_on} red_on={red_on}"
    )


def apply_mode(client, mode, seconds):
    blink_period = 1.0
    end = time.time() + seconds

    if mode == "safe":
        set_beacon(client, blue_on=False, red_on=True)
        print_state(client, "safe")
        time.sleep(seconds)

    elif mode == "ok_running":
        set_beacon(client, blue_on=True, red_on=False)
        print_state(client, "ok_running")
        time.sleep(seconds)

    elif mode == "ok_stopped":
        set_beacon(client, blue_on=True, red_on=True)
        print_state(client, "ok_stopped")
        time.sleep(seconds)

    elif mode == "odoo_bad_running":
        print("odoo_bad_running: azul intermitente, rojo apagado")
        while time.time() < end:
            set_beacon(client, blue_on=True, red_on=False)
            time.sleep(blink_period)
            set_beacon(client, blue_on=False, red_on=False)
            time.sleep(blink_period)

    elif mode == "machine_fault":
        print("machine_fault: azul fijo, rojo intermitente")
        while time.time() < end:
            set_beacon(client, blue_on=True, red_on=True)
            time.sleep(blink_period)
            set_beacon(client, blue_on=True, red_on=False)
            time.sleep(blink_period)

    elif mode == "odoo_bad_and_fault":
        print("odoo_bad_and_fault: azul intermitente, rojo intermitente")
        while time.time() < end:
            set_beacon(client, blue_on=True, red_on=True)
            time.sleep(blink_period)
            set_beacon(client, blue_on=False, red_on=False)
            time.sleep(blink_period)

    elif mode == "system_fault":
        print("system_fault: azul apagado, rojo intermitente")
        while time.time() < end:
            set_beacon(client, blue_on=False, red_on=True)
            time.sleep(blink_period)
            set_beacon(client, blue_on=False, red_on=False)
            time.sleep(blink_period)

    else:
        raise ValueError(f"Modo no reconocido: {mode}")


def test_machine(machine_id, host, mode, seconds):
    print(f"Probando {machine_id} en {host} modo={mode} durante {seconds}s")

    client = ModbusTcpClient(host, port=502, timeout=3)

    if not client.connect():
        raise RuntimeError(f"No conecta con {machine_id} {host}:502")

    try:
        apply_mode(client, mode, seconds)
        print_state(client, "estado final antes de safe")

        # Dejamos siempre estado seguro al terminar la prueba manual.
        set_beacon(client, blue_on=False, red_on=True)
        print_state(client, "safe final")

    finally:
        client.close()


def main():
    if len(sys.argv) < 3:
        print("Uso:")
        print("  python beacon_manual_test.py CORT-00 safe 5")
        print("  python beacon_manual_test.py CORT-00 ok_running 5")
        print("  python beacon_manual_test.py CORT-00 ok_stopped 5")
        print("  python beacon_manual_test.py CORT-00 odoo_bad_running 10")
        print("  python beacon_manual_test.py CORT-00 machine_fault 10")
        print("  python beacon_manual_test.py CORT-00 odoo_bad_and_fault 10")
        print("  python beacon_manual_test.py CORT-00 system_fault 10")
        print("  python beacon_manual_test.py all safe 5")
        raise SystemExit(1)

    target = sys.argv[1]
    mode = sys.argv[2]
    seconds = float(sys.argv[3]) if len(sys.argv) >= 4 else 5.0

    nodes = load_nodes()

    if target == "all":
        for machine_id, node in nodes.items():
            test_machine(machine_id, node["host"], mode, seconds)
        return

    if target not in nodes:
        raise SystemExit(f"No encuentro {target} en wise_nodes.json")

    node = nodes[target]
    test_machine(target, node["host"], mode, seconds)


if __name__ == "__main__":
    main()