pepsi.py

#!/usr/bin/env python3
import csv, time, os, datetime
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": "MCP6",
        "ip": "192.168.1.50",
        "device": "FF",
        "endian": "0",
        "registers": [
            ("Active1", "D0520", "W", True),
            ("Active2", "D0521", "W", True),
            ("Active3", "D0522", "W", True),
        ],
    },
    {
        "name": "HP1",
        "ip": "192.168.1.2",
        "device": "FF",
        "endian": "1",
        "registers": [
            ("LWT1", "D3938", "F", True),
        ],
    },
    {
        "name": "Hp2",
        "ip": "192.168.1.3",
        "device": "FF",
        "endian": "1",
        "registers": [
            ("LWT2", "D3938", "F", True),
        ],
    },
    {
        "name": "Hp3",
        "ip": "192.168.1.4",
        "device": "FF",
        "endian": "1",
        "registers": [
            ("LWT3", "D3938", "F", True),
        ],
    }
]

SAMPLE_INTERVAL_SECONDS = 1
DISPLAY_HISTORY_POINTS = 4 * 60 * 60  # 4 hours at 1 Hz
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:
            print(f"Error reading {tag} from {cfg['name']}: {e}")
            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"]:
            # objects to plot
            if visual and dtype in ("F", "W", "B"):
                items.append((cfg["name"], tag))
    return items

def update_visuals(fig, axes, history, visual_items):
    for ax, (plc_name, tag) in zip(axes, visual_items):
        ax.clear()
        points = list(history[(plc_name, tag)])
        if points:
            ax.plot([p[0] for p in points], [p[1] for p in points], label=f"{plc_name}:{tag}")
        ax.set_ylabel(tag)
        #ax.set_title(plc_name)
        if ax is axes[0]:
            ax.set_title(plc_name)
        else:
            ax.set_title("")

        ax.grid(True) # I think the grid is helpful
        ax.legend(loc="upper right") # matplotlib does some annoying bs I havent fixed yet.

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

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

def main():
    # Limit to 5 PLCs max
    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 = []

    if visual_items:
        import matplotlib.pyplot as plt_module
        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]

        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 {SAMPLE_INTERVAL_SECONDS} second(s) into: {os.path.abspath(LOG_DIR)}")
    print(f"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, history, visual_items)
            plt.pause(0.05)

        time.sleep(SAMPLE_INTERVAL_SECONDS)

if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\nExiting gracefully.")