rio_debug_scan.py

#!/usr/bin/env python3
'''
Debug/health scanner for groov RIO devices over OptoMMP/TCP.

Usage:

rio_debug_scan.py 192.168.1.50 --mode once
rio_debug_scan.py 192.168.1.50 --mode watch --interval 5
rio_debug_scan.py --host 192.168.1.50 --mode once
rio_debug_scan.py 192.168.1.50 --mode watch --heal-timeout 75
python3 /mnt/data/rio_debug_scan.py 192.168.1.50 --mode once --json

About Features:
- once mode: single scan and report
- watch mode: continuous scan and report
- optional host as positional arg or --host
- optional ICMP ping check before each scan
- "heal" attempt: reads the module-build area when ping fails or a read stalls
  because the user observed that reading module information can restore network response
- uses the generated CSV files when available, with safe hardcoded fallbacks
- json file output
This is intentionally conservative: it only reads addresses and never writes anything
except an optional Powerup Clear if you explicitly add that later.
'''

from __future__ import annotations

import argparse
import csv
import datetime as dt
import ipaddress
import json
import os
import platform
import socket
import struct
import subprocess
import sys
import time
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional, Tuple

DEFAULT_PORT = 2001
DEFAULT_TIMEOUT = 5.0
DEFAULT_INTERVAL = 5.0
DEFAULT_HEAL_TIMEOUT = 65.0
DEFAULT_MODULE_BUILD_LENGTH = 0x800  # From module-build area last valid address pattern.

# Fallback addresses from the OptoMMP guide / generated CSV.
FALLBACK_REGS: Dict[str, Tuple[int, int, str]] = {
    # name: (address, length, type)
    "mem_map_version": (0xFFFFF0300000, 4, "UI"),
    "powerup_clear_flag": (0xFFFFF0300004, 4, "UI"),
    "busy_flag": (0xFFFFF0300008, 4, "UI"),
    "last_error_code": (0xFFFFF030000C, 4, "I"),
    "txn_label": (0xFFFFF0300010, 2, "UI"),
    "source_address": (0xFFFFF0300012, 2, "UI"),
    "error_address": (0xFFFFF0300014, 4, "UI"),
    "loader_revision": (0xFFFFF0300018, 4, "UI"),
    "firmware_revision": (0xFFFFF030001C, 4, "IP"),
    "unit_type": (0xFFFFF0300020, 4, "UI"),
    "hw_month": (0xFFFFF0300024, 1, "UI"),
    "hw_day": (0xFFFFF0300025, 1, "UI"),
    "hw_year": (0xFFFFF0300026, 2, "UI"),
    "installed_ram": (0xFFFFF0300028, 4, "UI"),
    "mac_enet1": (0xFFFFF030002E, 6, "MAC"),
    "ip_enet1": (0xFFFFF0300034, 4, "IP"),
    "subnet_enet1": (0xFFFFF0300038, 4, "IP"),
    "gateway_enet1": (0xFFFFF030003C, 4, "IP"),
    "dns_enet1": (0xFFFFF0300040, 4, "IP"),
    "bootp_dhcp_status": (0xFFFFF0300058, 1, "UI"),
    "uptime_ms": (0xFFFFF030010C, 4, "UI"),
    "quality_mask": (0xFFFFF1002000, 8, "M"),
    "quality_code_0_0": (0xFFFFF1002080, 4, "UI"),
    "module0_type": (0xFFFFF8110000, 4, "UI"),
    "module0_firmware_mode": (0xFFFFF8110008, 4, "UI"),
    "module0_serial": (0xFFFFF811000C, 4, "UI"),
}

UNIT_TYPES = {
    0x00000103: "GRV-R7-MM1001-10",
    0x00000108: "GRV-R7-MM2001-10",
    0x00000101: "GRV-EPIC-PR1",
    0x00000107: "GRV-EPIC-PR2",
}

MODULE_TYPES = {
    0xF0000022: "groov RIO base module",
}

QUALITY_CODES = {
    0: "good/none",
    1: "bad value",
    2: "clamped high",
    3: "clamped low",
    30: "no channel present",
}


@dataclass
class RegisterDef:
    name: str
    address: int
    length: int
    typ: str
    description: str = ""


class OptoMMPClient:
    def __init__(self, host: str, port: int = DEFAULT_PORT, timeout: float = DEFAULT_TIMEOUT):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.sock: Optional[socket.socket] = None

    def connect(self) -> None:
        self.close()
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.settimeout(self.timeout)
        self.sock.connect((self.host, self.port))

    def close(self) -> None:
        if self.sock is not None:
            try:
                self.sock.close()
            except OSError:
                pass
            self.sock = None

    def _ensure(self) -> socket.socket:
        if self.sock is None:
            self.connect()
        assert self.sock is not None
        return self.sock

    def _recv_exact(self, n: int) -> bytes:
        sock = self._ensure()
        chunks = bytearray()
        while len(chunks) < n:
            chunk = sock.recv(n - len(chunks))
            if not chunk:
                raise ConnectionError("socket closed while receiving data")
            chunks.extend(chunk)
        return bytes(chunks)

    def read_block(self, address: int, length: int) -> bytes:
        if not (0 < length <= 0xFFFF):
            raise ValueError(f"invalid read length: {length}")
        sock = self._ensure()
        # Read Block Request packet per OptoMMP examples.
        request = bytearray(16)
        request[0:4] = bytes([0x00, 0x00, 0x04, 0x50])
        request[4:8] = bytes([0x00, 0x00, 0xFF, 0xFF])
        request[8:12] = address.to_bytes(4, "big")
        request[12:14] = length.to_bytes(2, "big")
        request[14:16] = bytes([0x00, 0x00])
        sock.sendall(request)

        # Read response header first.
        header = self._recv_exact(16)
        # rcode lives in low nibble of byte 3 in IEEE-1394 style responses.
        rcode = header[3] & 0x0F
        if rcode != 0:
            raise RuntimeError(f"read response returned rcode={rcode}")

        # Most devices return the requested data immediately after the 16-byte header.
        data = self._recv_exact(length)
        return data

    def write_quadlet(self, address: int, value: int) -> None:
        sock = self._ensure()
        request = bytearray(20)
        request[0:4] = bytes([0x00, 0x00, 0x04, 0x10])
        request[4:8] = bytes([0x00, 0x00, 0xFF, 0xFF])
        request[8:12] = address.to_bytes(4, "big")
        request[12:16] = (0).to_bytes(4, "big")
        request[16:20] = value.to_bytes(4, "big", signed=False)
        sock.sendall(request)
        resp = self._recv_exact(12)
        rcode = resp[3] & 0x0F
        if rcode != 0:
            raise RuntimeError(f"write response returned rcode={rcode}")


def parse_args() -> argparse.Namespace:
    p = argparse.ArgumentParser(description="groov RIO OptoMMP debug scanner")
    p.add_argument("host", nargs="?", default=None, help="Target RIO IP address (or use --host)")
    p.add_argument("--host", dest="host_kw", default=None, help="Target RIO IP address")
    p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OptoMMP TCP port (default {DEFAULT_PORT})")
    p.add_argument("--mode", choices=["once", "watch"], default="once", help="Scan once or continuously")
    p.add_argument("--interval", type=float, default=DEFAULT_INTERVAL, help="Seconds between scans in watch mode")
    p.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help="Socket timeout in seconds")
    p.add_argument("--heal-timeout", type=float, default=DEFAULT_HEAL_TIMEOUT, help="How long to keep retrying the heal read")
    p.add_argument("--no-ping", action="store_true", help="Skip ICMP ping check")
    p.add_argument("--csv", default=None, help="Path to groov_rio_memory_map_filtered.csv")
    p.add_argument("--patterns", default=None, help="Path to groov_rio_address_patterns.csv")
    p.add_argument("--json", action="store_true", help="Emit JSON per scan instead of text")
    p.add_argument("--quiet-heal", action="store_true", help="Suppress heal progress messages")
    args = p.parse_args()
    args.host = args.host_kw or args.host
    if not args.host:
        p.error("provide a target IP either as arg[1] or with --host")
    try:
        ipaddress.ip_address(args.host)
    except ValueError as exc:
        p.error(f"invalid IP address: {exc}")
    return args


def default_csv_path(filename: str) -> str:
    return os.path.join(os.path.dirname(os.path.abspath(__file__)), filename)


def load_register_defs(csv_path: Optional[str]) -> Dict[str, RegisterDef]:
    path = csv_path or default_csv_path("groov_rio_memory_map_filtered.csv")
    regs: Dict[str, RegisterDef] = {}
    if os.path.exists(path):
        try:
            with open(path, newline="", encoding="utf-8") as fh:
                reader = csv.DictReader(fh)
                for row in reader:
                    area = (row.get("area") or "").strip().lower()
                    desc = (row.get("description") or "").strip()
                    address_hex = (row.get("address_hex") or "").strip()
                    length_hex = (row.get("length_hex") or "").strip()
                    typ = (row.get("type") or "").strip()
                    if not address_hex or not length_hex:
                        continue
                    key = infer_register_name(area, desc, address_hex)
                    if not key:
                        continue
                    try:
                        address = int(address_hex.replace(" ", ""), 16)
                        length = int(length_hex.replace(" ", ""), 16)
                    except ValueError:
                        continue
                    regs[key] = RegisterDef(key, address, length, typ, desc)
        except Exception:
            pass
    for key, (address, length, typ) in FALLBACK_REGS.items():
        regs.setdefault(key, RegisterDef(key, address, length, typ, key.replace("_", " ")))
    return regs


def infer_register_name(area: str, desc: str, address_hex: str) -> Optional[str]:
    d = desc.lower()
    if area == "status_read":
        if "memory map revision number" in d:
            return "mem_map_version"
        if "powerup clear flag" in d:
            return "powerup_clear_flag"
        if "busy flag" in d:
            return "busy_flag"
        if "last error code" in d:
            return "last_error_code"
        if "transaction label" in d:
            return "txn_label"
        if "source address of the unit" in d:
            return "source_address"
        if "error address for last error" in d:
            return "error_address"
        if d.startswith("loader revision"):
            return "loader_revision"
        if d.startswith("firmware (kernel) revision"):
            return "firmware_revision"
        if d.startswith("unit type"):
            return "unit_type"
        if "hardware revision (month)" in d:
            return "hw_month"
        if "hardware revision (day)" in d:
            return "hw_day"
        if "hardware revision (year)" in d:
            return "hw_year"
        if "installed ram" in d:
            return "installed_ram"
        if "mac address" in d or "enet1 mac address" in d:
            return "mac_enet1"
        if "enet1 tcp/ip address" in d:
            return "ip_enet1"
        if "enet1 tcp/ip subnet mask" in d:
            return "subnet_enet1"
        if "enet1 tcp/ip default gateway" in d:
            return "gateway_enet1"
        if "enet1 dns server address" in d:
            return "dns_enet1"
        if "bootp" in d or "dhcp" in d:
            return "bootp_dhcp_status"
        if "uptime" in d:
            return "uptime_ms"
    if area == "status_write" and "operation code" in d:
        return "operation_code"
    if area == "quality":
        if "quality indicators present" in d:
            return "quality_mask"
        if "(0,0) quality indicator code" in d:
            return "quality_code_0_0"
    if area == "module_build":
        if "module 0. module type" in d:
            return "module0_type"
        if "firmware mode" in d:
            return "module0_firmware_mode"
        if "manufacturer’s serial number" in d or "manufacturer's serial number" in d:
            return "module0_serial"
    return None


def ping(host: str, timeout_s: float = 1.0) -> bool:
    system = platform.system().lower()
    if system == "windows":
        cmd = ["ping", "-n", "1", "-w", str(int(timeout_s * 1000)), host]
    else:
        cmd = ["ping", "-c", "1", "-W", str(max(1, int(timeout_s))), host]
    try:
        proc = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return proc.returncode == 0
    except Exception:
        return False


def decode_value(data: bytes, typ: str):
    typ = (typ or "").upper()
    n = len(data)
    if typ == "IP" and n == 4:
        return ".".join(str(b) for b in data)
    if typ == "MAC" and n == 6:
        return ":".join(f"{b:02X}" for b in data)
    if typ == "I" and n == 4:
        return struct.unpack(">i", data)[0]
    if typ == "F" and n == 4:
        return struct.unpack(">f", data)[0]
    if typ in {"UI", "M", "B"}:
        if n == 1:
            return data[0]
        if n == 2:
            return int.from_bytes(data, "big", signed=False)
        if n == 4:
            return int.from_bytes(data, "big", signed=False)
        if n == 8:
            return int.from_bytes(data, "big", signed=False)
    return data.hex()


def decode_firmware_word_as_text(value: int) -> str:
    b = value.to_bytes(4, "big", signed=False)
    return f"{b[0]}.{b[1]} rev={b[2]} letter={b[3]}"


def decode_hw_date(month: int, day: int, year: int) -> str:
    try:
        return f"{year:04d}-{month:02d}-{day:02d}"
    except Exception:
        return f"{month}/{day}/{year}"


def module_type_name(code: int) -> str:
    return MODULE_TYPES.get(code, f"0x{code:08X}")


def unit_type_name(code: int) -> str:
    return UNIT_TYPES.get(code, f"0x{code:08X}")


def quality_channels_from_mask(mask_value: int, max_channels: int = 64) -> List[int]:
    channels: List[int] = []
    for ch in range(max_channels):
        if (mask_value >> ch) & 1:
            channels.append(ch)
    return channels


def read_quality_codes(client: OptoMMPClient, base_addr: int, count: int = 64) -> List[int]:
    # For groov RIO all channels are effectively module position 0 with channel stride 4.
    block = client.read_block(base_addr, count * 4)
    return [int.from_bytes(block[i:i + 4], "big", signed=False) for i in range(0, len(block), 4)]


def heal_read_all_modules(host: str, port: int, timeout: float, heal_timeout: float, quiet: bool = False) -> bool:
    start = time.time()
    attempts = 0
    while time.time() - start < heal_timeout:
        attempts += 1
        try:
            client = OptoMMPClient(host, port=port, timeout=timeout)
            client.connect()
            _ = client.read_block(0xFFFFF8110000, DEFAULT_MODULE_BUILD_LENGTH)
            client.close()
            if not quiet:
                print(f"[heal] module-build read succeeded after {attempts} attempt(s)", file=sys.stderr)
            return True
        except Exception as exc:
            if not quiet:
                print(f"[heal] attempt {attempts}: {exc}", file=sys.stderr)
            time.sleep(1.0)
    return False


def read_named(client: OptoMMPClient, reg: RegisterDef):
    raw = client.read_block(reg.address, reg.length)
    return decode_value(raw, reg.typ)


def perform_scan(host: str, port: int, timeout: float, regs: Dict[str, RegisterDef], do_ping: bool,
                 heal_timeout: float, quiet_heal: bool) -> Dict[str, object]:
    result: Dict[str, object] = {
        "timestamp": dt.datetime.now().isoformat(timespec="seconds"),
        "host": host,
        "port": port,
    }

    ping_ok = True
    if do_ping:
        ping_ok = ping(host)
        result["ping_ok"] = ping_ok

    if do_ping and not ping_ok:
        healed = heal_read_all_modules(host, port, timeout, heal_timeout, quiet=quiet_heal)
        result["heal_attempted"] = True
        result["heal_succeeded"] = healed
        result["ping_ok_after_heal"] = ping(host)
    else:
        result["heal_attempted"] = False

    client = OptoMMPClient(host, port=port, timeout=timeout)
    try:
        client.connect()
        status: Dict[str, object] = {}
        for key in [
            "mem_map_version", "powerup_clear_flag", "busy_flag", "last_error_code", "txn_label",
            "source_address", "error_address", "loader_revision", "firmware_revision", "unit_type",
            "hw_month", "hw_day", "hw_year", "installed_ram", "mac_enet1", "ip_enet1",
            "subnet_enet1", "gateway_enet1", "dns_enet1", "bootp_dhcp_status", "uptime_ms",
        ]:
            reg = regs.get(key)
            if reg is None:
                continue
            try:
                status[key] = read_named(client, reg)
            except Exception as exc:
                status[key] = f"ERROR: {exc}"

        # Derived fields.
        unit_type = status.get("unit_type")
        if isinstance(unit_type, int):
            status["unit_type_name"] = unit_type_name(unit_type)
        if isinstance(status.get("loader_revision"), int):
            status["loader_revision_text"] = decode_firmware_word_as_text(status["loader_revision"])  # type: ignore[index]
        if isinstance(status.get("firmware_revision"), str):
            status["firmware_revision_text"] = status["firmware_revision"]
        if all(isinstance(status.get(k), int) for k in ("hw_month", "hw_day", "hw_year")):
            status["hardware_revision_date"] = decode_hw_date(
                status["hw_month"], status["hw_day"], status["hw_year"]
            )
        if isinstance(status.get("uptime_ms"), int):
            status["uptime_s"] = round(status["uptime_ms"] / 1000.0, 3)

        result["status"] = status

        # Module info / the healing read.
        module_info: Dict[str, object] = {}
        try:
            module_block = client.read_block(0xFFFFF8110000, DEFAULT_MODULE_BUILD_LENGTH)
            module0_type = int.from_bytes(module_block[0:4], "big", signed=False)
            module0_mode = int.from_bytes(module_block[8:12], "big", signed=False)
            module0_serial = int.from_bytes(module_block[12:16], "big", signed=False)
            module_info = {
                "module0_type": module0_type,
                "module0_type_name": module_type_name(module0_type),
                "module0_firmware_mode": module0_mode,
                "module0_firmware_mode_text": chr(module0_mode) if 0 <= module0_mode <= 255 else str(module0_mode),
                "module0_serial": module0_serial,
                "module_build_read_ok": True,
            }
        except Exception as exc:
            module_info = {"module_build_read_ok": False, "error": str(exc)}
        result["module_build"] = module_info

        # Quality mask and per-channel quality codes.
        quality: Dict[str, object] = {}
        try:
            mask_reg = regs["quality_mask"]
            qmask = read_named(client, mask_reg)
            assert isinstance(qmask, int)
            channels = quality_channels_from_mask(qmask)
            quality["mask"] = f"0x{qmask:016X}"
            quality["channels_with_quality_bits"] = channels
            codes = read_quality_codes(client, regs["quality_code_0_0"].address, 64)
            nonzero = [{
                "channel": idx,
                "code": code,
                "meaning": QUALITY_CODES.get(code, "see guide page 213"),
            } for idx, code in enumerate(codes) if code != 0]
            quality["nonzero_codes"] = nonzero
        except Exception as exc:
            quality["error"] = str(exc)
        result["quality"] = quality

    finally:
        client.close()

    return result


def format_text_report(scan: Dict[str, object]) -> str:
    lines: List[str] = []
    lines.append(f"[{scan.get('timestamp')}] host={scan.get('host')} port={scan.get('port')}")
    if "ping_ok" in scan:
        lines.append(f"ping_ok={scan.get('ping_ok')}")
    if scan.get("heal_attempted"):
        lines.append(f"heal_attempted=True heal_succeeded={scan.get('heal_succeeded')}")
        if "ping_ok_after_heal" in scan:
            lines.append(f"ping_ok_after_heal={scan.get('ping_ok_after_heal')}")

    status = scan.get("status", {})
    if isinstance(status, dict):
        lines.append("status:")
        for key in [
            "unit_type_name", "unit_type", "mem_map_version", "powerup_clear_flag", "busy_flag",
            "last_error_code", "error_address", "loader_revision_text", "firmware_revision_text",
            "hardware_revision_date", "installed_ram", "mac_enet1", "ip_enet1", "subnet_enet1",
            "gateway_enet1", "dns_enet1", "bootp_dhcp_status", "uptime_ms", "uptime_s",
        ]:
            if key in status:
                lines.append(f"  {key}: {status[key]}")

    module_build = scan.get("module_build", {})
    if isinstance(module_build, dict):
        lines.append("module_build:")
        for key in [
            "module_build_read_ok", "module0_type_name", "module0_type",
            "module0_firmware_mode_text", "module0_firmware_mode", "module0_serial", "error",
        ]:
            if key in module_build:
                lines.append(f"  {key}: {module_build[key]}")

    quality = scan.get("quality", {})
    if isinstance(quality, dict):
        lines.append("quality:")
        if "mask" in quality:
            lines.append(f"  mask: {quality['mask']}")
        channels = quality.get("channels_with_quality_bits", [])
        if isinstance(channels, list):
            lines.append(f"  channels_with_quality_bits: {channels}")
        nonzero = quality.get("nonzero_codes", [])
        if isinstance(nonzero, list) and nonzero:
            lines.append("  nonzero_codes:")
            for item in nonzero:
                lines.append(
                    f"    ch {item['channel']:02d}: code={item['code']} meaning={item['meaning']}"
                )
        elif "error" in quality:
            lines.append(f"  error: {quality['error']}")
        else:
            lines.append("  nonzero_codes: []")

    return "\n".join(lines)


def main() -> int:
    args = parse_args()
    regs = load_register_defs(args.csv)

    while True:
        try:
            scan = perform_scan(
                host=args.host,
                port=args.port,
                timeout=args.timeout,
                regs=regs,
                do_ping=not args.no_ping,
                heal_timeout=args.heal_timeout,
                quiet_heal=args.quiet_heal,
            )
            if args.json:
                print(json.dumps(scan, indent=2, sort_keys=False))
            else:
                print(format_text_report(scan))
        except KeyboardInterrupt:
            return 130
        except Exception as exc:
            stamp = dt.datetime.now().isoformat(timespec="seconds")
            msg = {"timestamp": stamp, "host": args.host, "error": str(exc)}
            if args.json:
                print(json.dumps(msg, indent=2))
            else:
                print(f"[{stamp}] ERROR host={args.host}: {exc}", file=sys.stderr)

        if args.mode == "once":
            return 0
        time.sleep(max(0.1, args.interval))


if __name__ == "__main__":
    raise SystemExit(main())