wrps-demo-kit/05-tests/fake_plc.py

#!/usr/bin/env python3
"""A fake PLC: serves known values on Modbus TCP so the SCADA side can be
tested on its own.

    python3 fake_plc.py --serve            # a spare port - NOT 502 on the live host
    python fake_plc.py --expect            # anywhere: what CI Server should show

WHY THIS EXISTS
---------------
Everything downstream of the register map - CI Server's Modbus config, the
conversions, the item bindings, the display masks - has only ever been
tested against the simulation, whose values are plausible.  Plausible is
exactly what you cannot check: a level of 70% reads fine whether it came
from the right register or the one next to it.

So: stop the PLC, serve *distinctive* values instead, and read the HMI.
Every register gets a different value, none of them round, none of them
equal to a neighbour, and the pattern is deterministic - the same seed
gives the same numbers, so the expected table can be printed anywhere,
including on a machine that cannot reach the server.

Stdlib only, on purpose.  The Ubuntu box has no pymodbus and a broken
ensurepip, and installing packages on someone's VM to run a test is a
worse trade than writing the 6 bytes of MBAP header out by hand.

WHAT IT SPEAKS
--------------
FC01 read coils, FC02 read discrete inputs, FC03 read holding registers,
FC04 read input registers, FC06 write single register, FC16 write multiple
registers.  Writes are accepted and stored, so a setpoint written from the
HMI can be read back - that tests the write path in the same session.
Anything else gets exception 01 (illegal function).
"""

import argparse
import socket
import socketserver
import struct
import sys
from pathlib import Path

SEED = 20260814          # the date; change it for a fresh pattern
UNIT = 1

# ---------------------------------------------------------------- values


VALUES = Path(__file__).resolve().parent / "fake_values.json"
POINTS = Path(__file__).resolve().parents[1] / "04-scada" / "modbus_points" / "ci-server-points.csv"

# A deliberately lopsided bit pattern.  Not alternating: 1,0,1,0... is the
# one pattern that still looks like itself when the mapping slips, and an
# LCG's low bit produces exactly that.  This one is distinguishable from
# every shift of itself.
COIL_BITS = 0b011011000110101


def lcg(seed):
    """A tiny deterministic generator - only used for the middle digits."""
    x = seed
    while True:
        x = (1103515245 * x + 12345) & 0x7FFFFFFF
        yield x >> 8          # low bits of an LCG are not random; drop them


def make():
    """Build the test values from the point list, and write them out.

    Two constraints fight each other and both matter:

      - a value must be DISTINCTIVE, so a point reading its neighbour's
        register is obvious.  Plausible values are what makes a mis-mapped
        point invisible, which is the whole reason for this exercise.
      - a value must FIT THE DISPLAY MASK.  The masks are sized for the
        plant's real ranges, so a test value that overflows one renders as
        nothing and looks like a fault that is not there.  The first
        attempt did exactly this - CMD_ACK got 2065 against a mask of 99.

    So the value is chosen per point, from its own mask and gain: as large
    and as odd-looking as the mask can render, and no larger.
    """
    import csv
    import json
    g = lcg(SEED)
    coils, holding, notes = {}, {}, {}

    for i, r in enumerate(csv.DictReader(POINTS.open(newline="", encoding="utf-8"))):
        addr, iec, mask = int(r["modbus_address"]), r["iec_address"], r["format_mask"]
        if r["function_code"] == "FC01":
            coils[addr] = bool(COIL_BITS >> addr & 1)
            continue

        gain = float(r["eng_gain"])
        int_digits = len(mask.split(".")[0])
        max_eng = 10 ** int_digits - 1                 # widest the mask renders
        max_raw = min(int(max_eng / gain), 32000)      # ...and a register holds

        if iec == "%QW17":                             # alarm bitmask, unsigned
            holding[addr] = 0xA5C3
            notes[iec] = "bit15 set - proves the point is UNSIGNED"
        elif iec == "%QW7":                            # net accumulation, signed
            holding[addr] = -(next(g) % 900 + 100)
            notes[iec] = "NEGATIVE - proves the point is signed"
        elif iec == "%QW5":                            # time to spill
            holding[addr] = 32767
            notes[iec] = "32767 - the 'drawing down' sentinel"
        elif max_raw <= 9:                             # enums, counts, modes
            holding[addr] = 1 + (next(g) % min(6, max_raw))
        else:
            # SPREAD, not clustered.  The first attempt put the address in
            # the leading digits, which made the eleven setpoints land
            # within 46-55% of each other - so a point reading its
            # neighbour's register looked entirely reasonable, which is the
            # one thing this test must not allow.  Each point now takes a
            # different fraction of its own renderable range, stepping by
            # 37/89 so consecutive points land far apart, with an odd tail
            # so no two are equal.
            frac = ((i * 37 + 11) % 89 + 6) / 100.0
            v = int(max_raw * frac)
            v = v - v % 100 + ((i * 13) % 89) + 7 if v > 200 else v
            holding[addr] = max(1, min(v, max_raw))

    data = {"seed": SEED, "coils": coils, "holding": holding, "notes": notes}
    VALUES.write_text(json.dumps(data, indent=1, sort_keys=True) + "\n",
                      encoding="utf-8")
    print("wrote %s  (%d coils, %d holding registers)"
          % (VALUES.name, len(coils), len(holding)))
    return data


def pattern():
    import json
    if not VALUES.is_file():
        sys.exit("missing %s - run:  python fake_plc.py --make" % VALUES.name)
    d = json.loads(VALUES.read_text(encoding="utf-8"))
    return {"coils": {int(k): v for k, v in d["coils"].items()},
            "holding": {int(k): v for k, v in d["holding"].items()},
            "notes": d.get("notes", {})}


# ---------------------------------------------------------------- server
class Handler(socketserver.BaseRequestHandler):
    def handle(self):
        data = self.server.data
        while True:
            # MBAP is 7 bytes: transaction, protocol, length, unit.  `length`
            # counts the unit byte plus the PDU, so the PDU is length - 1.
            head = self._recv(7)
            if not head:
                return
            tid, pid, length, unit = struct.unpack(">HHHB", head)
            body = self._recv(length - 1)
            if not body:
                return
            fc = body[0]
            try:
                resp = self.dispatch(fc, body[1:], data)
            except Exception:
                resp = bytes([fc | 0x80, 0x02])
            self.request.sendall(
                struct.pack(">HHHB", tid, pid, len(resp) + 1, unit) + resp)

    def _recv(self, n):
        buf = b""
        while len(buf) < n:
            chunk = self.request.recv(n - len(buf))
            if not chunk:
                return b""
            buf += chunk
        return buf

    def dispatch(self, fc, args, data):
        if fc in (1, 2):                                   # read bits
            addr, count = struct.unpack(">HH", args[:4])
            bits = [data["coils"].get(addr + i, False) for i in range(count)]
            packed = bytearray((count + 7) // 8)
            for i, b in enumerate(bits):
                if b:
                    packed[i // 8] |= 1 << (i % 8)
            return bytes([fc, len(packed)]) + bytes(packed)

        if fc in (3, 4):                                   # read registers
            addr, count = struct.unpack(">HH", args[:4])
            out = b""
            for i in range(count):
                v = data["holding"].get(addr + i, 0) if fc == 3 else 0
                out += struct.pack(">H", v & 0xFFFF)
            return bytes([fc, len(out)]) + out

        if fc == 6:                                        # write one
            addr, value = struct.unpack(">HH", args[:4])
            data["holding"][addr] = value - 65536 if value > 32767 else value
            data["writes"].append((addr, data["holding"][addr]))
            print("   WRITE  holding %d <- %d" % (addr, data["holding"][addr]),
                  flush=True)
            return bytes([fc]) + args[:4]

        if fc == 16:                                       # write many
            addr, count = struct.unpack(">HH", args[:4])
            for i in range(count):
                v, = struct.unpack(">H", args[5 + i * 2:7 + i * 2])
                data["holding"][addr + i] = v - 65536 if v > 32767 else v
                data["writes"].append((addr + i, data["holding"][addr + i]))
                print("   WRITE  holding %d <- %d" % (addr + i,
                                                      data["holding"][addr + i]),
                      flush=True)
            return bytes([fc]) + args[:4]

        return bytes([fc | 0x80, 0x01])


class Server(socketserver.ThreadingTCPServer):
    allow_reuse_address = True
    daemon_threads = True


def serve(port):
    data = pattern()
    data["writes"] = []
    srv = Server(("0.0.0.0", port), Handler)
    srv.data = data
    print("fake PLC listening on 0.0.0.0:%d  (seed %d)" % (port, SEED), flush=True)
    print("   %d coils, %d holding registers served"
          % (len(data["coils"]), len(data["holding"])), flush=True)
    print("   writes from SCADA are stored and logged below", flush=True)
    try:
        srv.serve_forever()
    except KeyboardInterrupt:
        pass


# ---------------------------------------------------------------- expect
def half_up(value, decimals):
    """Round the way CI Server does, not the way Python does.

    Python's %.1f rounds an exact half to even: 113.25 -> "113.2".
    CI Server rounds it up: 113.3.  Two of the 49 test values landed
    exactly on a half, and both were reported as mismatches by the first
    run of this table when the SCADA side was in fact correct.  A test
    that cries wolf twice in fifty is worse than no test.
    """
    from decimal import Decimal, ROUND_HALF_UP
    q = Decimal("1." + "0" * decimals) if decimals else Decimal("1")
    return Decimal(repr(value)).quantize(q, rounding=ROUND_HALF_UP)


def expect():
    """What CI Server should display, in engineering units."""
    import csv
    pts = Path(__file__).resolve().parents[1] / "04-scada" / "modbus_points" / "ci-server-points.csv"
    rows = list(csv.DictReader(pts.open(newline="", encoding="utf-8")))
    data = pattern()

    # ci-server-points.csv carries the CI Server item name directly, so the
    # name shown here is the one an operator sees on a display.  (An earlier
    # version fell back to a `scada_tag` column of PS_* placeholder names;
    # those named nothing that exists and have been removed - see
    # 04-scada/modbus_points/README.md.)
    leaf = {r["iec_address"]: r["ci_item"].replace("AID.WRPS.", "") for r in rows}

    print("Expected CI Server readings while fake_plc.py is serving "
          "(seed %d)\n" % SEED)
    print("%-22s %-10s %8s   %-14s %s"
          % ("ITEM", "ADDRESS", "RAW", "SHOULD READ", "NOTE"))
    print("-" * 78)
    for r in rows:
        a = int(r["modbus_address"])
        iec = r["iec_address"]
        if r["function_code"] == "FC01":
            raw = int(data["coils"].get(a, False))
            shown = str(raw)
        else:
            raw = data["holding"].get(a, 0)
            gain = float(r["eng_gain"])
            dec = len(r["format_mask"].split(".")[1]) if "." in r["format_mask"] else 0
            eng = (raw & 0xFFFF) * gain if iec == "%QW17" else raw * gain
            shown = "%s %s" % (half_up(eng, dec), r["eng_units"])
        print("%-22s %-10s %8d   %-14s %s"
              % (leaf.get(iec, r["ci_item"].replace("AID.WRPS.", "")),
                 "%s:%d" % (r["function_code"], a), raw, shown,
                 data["notes"].get(iec, "")))


def main():
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    ap.add_argument("--make", action="store_true",
                    help="generate the test values from the point list")
    ap.add_argument("--serve", action="store_true", help="run the server")
    ap.add_argument("--expect", action="store_true", help="print expected readings")
    ap.add_argument("--port", type=int, default=502)
    a = ap.parse_args()
    if a.make:
        make()
    elif a.serve:
        serve(a.port)
    elif a.expect:
        expect()
    else:
        ap.print_help()
        sys.exit(1)


if __name__ == "__main__":
    main()