sp.py

#!/usr/bin/env python3
import csv
import time
import os
import datetime
import sys
from collections import defaultdict, deque

try:
    from fc6a import FC6AMaint
except ImportError:
    print("fc6a module not found, downloading from GitHub...")
    import requests
    url = "https://raw.githubusercontent.com/Makerspace-Bangor/fc6a/main/src/fc6a.py"
    code = requests.get(url, timeout=20).text
    exec(code, globals())
    print("fc6a module loaded from GitHub.")

PLC_CONFIGS = [
    {
        "name": "HP1",
        "ip": "192.168.1.2",
        "device": "FF",
        "endian": "1",
        "registers": [
            ("RIO1", "D0516", "F", True),
        ],
    },
    {
        "name": "HP2",
        "ip": "192.168.1.3",
        "device": "FF",
        "endian": "1",
        "registers": [
            ("RIO2", "D0516", "F", True),
        ],
    },
    {
        "name": "HP3",
        "ip": "192.168.1.4",
        "device": "FF",
        "endian": "1",
        "registers": [
            ("RIO3", "D0516", "F", True),
        ],
    },
    {
        "name": "MCP6",
        "ip": "192.168.1.50",
        "device": "FF",
        "endian": "0",
        "registers": [
            ("HP1", "D0520", "W", True),
            ("HP2", "D0521", "W", True),
            ("HP3", "D0522", "W", True),
        ],
    }
]

# Fixed colors per PLC source
PLC_COLORS = {
    "HP1": "green",
    "HP2": "blue",
    "HP3": "orange",
    "MCP6": "red",
}

SAMPLE_INTERVAL_SECONDS = 1
DISPLAY_HISTORY_POINTS = 4 * 60 * 60
LOG_DIR = "."


def get_csv_filename():
    date_str = datetime.date.today().strftime("%Y-%m-%d")
    return os.path.join(LOG_DIR, f"logs_{date_str}.csv")


def build_combined_headers(active_plcs):
    source_header = ["timestamp"]
    register_header = ["timestamp"]

    for cfg in active_plcs:
        for tag, _reg, _dtype, _visual in cfg["registers"]:
            source_header.append(cfg["name"])
            register_header.append(tag)

    return source_header, register_header


def ensure_csv_header(filename, source_header, register_header):
    if not os.path.exists(filename) or os.path.getsize(filename) == 0:
        with open(filename, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(source_header)
            writer.writerow(register_header)


def read_plc_values(cfg):
    plc = FC6AMaint(cfg["ip"])
    swapped = bool(int(cfg.get("endian", "0")))
    data = {}

    for tag, reg, dtype, _visual in cfg["registers"]:
        try:
            rnum = int(reg[1:])

            if dtype == "F":
                val = round(plc.read_float(rnum, swapped), 2)
            elif dtype == "W":
                val = plc.read_word(rnum)
            elif dtype == "B":
                val = plc.read_bits(rnum)
            else:
                val = None

        except Exception as e:
            ts = datetime.datetime.now().isoformat(timespec="seconds")
            print(
                f"\r{ts}  Error reading {tag} from {cfg['name']}: {e}\033[K",
                file=sys.stderr,
                flush=True
            )
            val = None

        data[tag] = val

    return data


def log_all_plcs(active_plcs):
    now = datetime.datetime.now()
    timestamp = now.isoformat(timespec="seconds")
    readings = {}

    source_header, register_header = build_combined_headers(active_plcs)
    filename = get_csv_filename()

    ensure_csv_header(filename, source_header, register_header)

    row = [timestamp]

    for cfg in active_plcs:
        row_values = read_plc_values(cfg)

        readings[cfg["name"]] = {
            "timestamp": now,
            "values": row_values
        }

        for tag, _reg, _dtype, _visual in cfg["registers"]:
            row.append(row_values.get(tag, ""))

    with open(filename, "a", newline="") as f:
        csv.writer(f).writerow(row)

    return readings


def selected_visual_items(active_plcs):
    items = []

    for cfg in active_plcs:
        for tag, _reg, dtype, visual in cfg["registers"]:
            if visual and dtype in ("F", "W", "B"):
                items.append((cfg["name"], tag))

    return items


def update_visuals(fig, axes, lines, history, visual_items, mdates):
    for ax, (plc_name, tag) in zip(axes, visual_items):
        points = list(history[(plc_name, tag)])
        line = lines[(plc_name, tag)]

        if points:
            x_vals = [p[0] for p in points]
            y_vals = [p[1] for p in points]

            line.set_data(x_vals, y_vals)

            ax.relim()
            ax.autoscale_view()

        ax.xaxis.set_major_locator(mdates.AutoDateLocator())
        ax.xaxis.set_major_formatter(
            mdates.DateFormatter("%Y-%m-%d %H:%M")
        )

    fig.autofmt_xdate()
    fig.tight_layout()
    fig.canvas.draw_idle()


def main():
    active_plcs = PLC_CONFIGS[:5]
    visual_items = selected_visual_items(active_plcs)

    history = defaultdict(
        lambda: deque(maxlen=DISPLAY_HISTORY_POINTS)
    )

    plt = None
    fig = None
    axes = []
    lines = {}

    if visual_items:
        import matplotlib.pyplot as plt_module
        import matplotlib.dates as mdates

        plt = plt_module

        fig, axes_obj = plt.subplots(
            len(visual_items),
            1,
            figsize=(11, max(3, len(visual_items) * 2.4)),
            sharex=True
        )

        axes = (
            axes_obj.flatten().tolist()
            if hasattr(axes_obj, "flatten")
            else [axes_obj]
        )

        for ax, (plc_name, tag) in zip(axes, visual_items):
            line, = ax.plot(
                [],
                [],
                label=f"{plc_name}:{tag}",
                color=PLC_COLORS.get(plc_name, None)
            )

            lines[(plc_name, tag)] = line

            ax.set_ylabel(tag)
            #ax.set_title(plc_name)
            ax.grid(True)
            ax.legend(loc="upper right")

        axes[-1].set_xlabel("Time")

        fig.suptitle("FC6A Live Trends - Last 4 Hours at 1 Hz")

        plt.ion()
        plt.show(block=False)

    print(
        f"Logging {len(active_plcs)} PLC(s) every "
        f"{SAMPLE_INTERVAL_SECONDS} second(s) into: "
        f"{os.path.abspath(LOG_DIR)}"
    )

    print("CSV file pattern: logs_YYYY-MM-DD.csv")

    if visual_items:
        print("Visualizing checked numeric registers:")

        for plc_name, tag in visual_items:
            print(f"  {plc_name}:{tag}")

    else:
        print(
            "No numeric registers checked for visualization. "
            "Logging only."
        )

    while True:
        readings = log_all_plcs(active_plcs)

        if visual_items:
            for plc_name, tag in visual_items:
                reading = readings.get(plc_name)

                if not reading:
                    continue

                val = reading["values"].get(tag)

                if isinstance(val, (int, float)):
                    history[(plc_name, tag)].append(
                        (reading["timestamp"], val)
                    )

            update_visuals(
                fig,
                axes,
                lines,
                history,
                visual_items,
                mdates
            )

            plt.pause(0.05)

        time.sleep(SAMPLE_INTERVAL_SECONDS)


if __name__ == "__main__":
    try:
        main()

    except KeyboardInterrupt:
        print("\nExiting gracefully.")