From 1c64aeb07185480b0f9ba6ad9a83d53611299058 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Wed, 7 Oct 2026 10:38:21 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=89=20Initial=20commit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 16 + .vscode/launch.json | 140 +++++++++ .vscode/settings.json | 6 + README.md | 198 ++++++++++++ config.example.yaml | 42 +++ docs/IMPLEMENTATION_PLAN.md | 211 +++++++++++++ docs/SPECIFICATION.md | 263 ++++++++++++++++ pyproject.toml | 34 ++ src/opcda_diagnozer/__init__.py | 13 + src/opcda_diagnozer/__main__.py | 6 + src/opcda_diagnozer/config.py | 255 +++++++++++++++ src/opcda_diagnozer/events.py | 87 ++++++ src/opcda_diagnozer/main.py | 377 +++++++++++++++++++++++ src/opcda_diagnozer/monitor/__init__.py | 1 + src/opcda_diagnozer/monitor/detectors.py | 24 ++ src/opcda_diagnozer/monitor/poller.py | 25 ++ src/opcda_diagnozer/monitor/state.py | 20 ++ src/opcda_diagnozer/opc/__init__.py | 1 + src/opcda_diagnozer/opc/automation.py | 222 +++++++++++++ src/opcda_diagnozer/opc/base.py | 164 ++++++++++ src/opcda_diagnozer/opc/browse.py | 194 ++++++++++++ src/opcda_diagnozer/opc/discovery.py | 224 ++++++++++++++ src/opcda_diagnozer/opc/fake.py | 130 ++++++++ src/opcda_diagnozer/quality.py | 69 +++++ src/opcda_diagnozer/report/__init__.py | 1 + src/opcda_diagnozer/report/base.py | 13 + src/opcda_diagnozer/report/eventlog.py | 22 ++ src/opcda_diagnozer/report/http.py | 23 ++ src/opcda_diagnozer/report/jsonl.py | 25 ++ src/opcda_diagnozer/service.py | 19 ++ test.py | 0 tests/test_browse.py | 310 +++++++++++++++++++ tests/test_config.py | 85 +++++ tests/test_detectors.py | 20 ++ tests/test_discovery.py | 66 ++++ tests/test_poller_fake.py | 33 ++ tests/test_quality.py | 41 +++ tests/test_state.py | 11 + 38 files changed, 3391 insertions(+) create mode 100644 .gitignore create mode 100644 .vscode/launch.json create mode 100644 .vscode/settings.json create mode 100644 README.md create mode 100644 config.example.yaml create mode 100644 docs/IMPLEMENTATION_PLAN.md create mode 100644 docs/SPECIFICATION.md create mode 100644 pyproject.toml create mode 100644 src/opcda_diagnozer/__init__.py create mode 100644 src/opcda_diagnozer/__main__.py create mode 100644 src/opcda_diagnozer/config.py create mode 100644 src/opcda_diagnozer/events.py create mode 100644 src/opcda_diagnozer/main.py create mode 100644 src/opcda_diagnozer/monitor/__init__.py create mode 100644 src/opcda_diagnozer/monitor/detectors.py create mode 100644 src/opcda_diagnozer/monitor/poller.py create mode 100644 src/opcda_diagnozer/monitor/state.py create mode 100644 src/opcda_diagnozer/opc/__init__.py create mode 100644 src/opcda_diagnozer/opc/automation.py create mode 100644 src/opcda_diagnozer/opc/base.py create mode 100644 src/opcda_diagnozer/opc/browse.py create mode 100644 src/opcda_diagnozer/opc/discovery.py create mode 100644 src/opcda_diagnozer/opc/fake.py create mode 100644 src/opcda_diagnozer/quality.py create mode 100644 src/opcda_diagnozer/report/__init__.py create mode 100644 src/opcda_diagnozer/report/base.py create mode 100644 src/opcda_diagnozer/report/eventlog.py create mode 100644 src/opcda_diagnozer/report/http.py create mode 100644 src/opcda_diagnozer/report/jsonl.py create mode 100644 src/opcda_diagnozer/service.py create mode 100644 test.py create mode 100644 tests/test_browse.py create mode 100644 tests/test_config.py create mode 100644 tests/test_detectors.py create mode 100644 tests/test_discovery.py create mode 100644 tests/test_poller_fake.py create mode 100644 tests/test_quality.py create mode 100644 tests/test_state.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a8750a --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# Python bytecode / build +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ + +# Virtual environments +.venv/ + +# Test / tool caches +.pytest_cache/ + +# Local runtime artifacts (copy config.example.yaml -> config.yaml) +config.yaml +logs/ \ No newline at end of file diff --git a/.vscode/launch.json b/.vscode/launch.json new file mode 100644 index 0000000..28b73b2 --- /dev/null +++ b/.vscode/launch.json @@ -0,0 +1,140 @@ +{ + // Debug configurations for the OPC DA Diagnozer CLI. + // All configs run `python -m opcda_diagnozer` with the 32-bit venv + // interpreter — the bitness that matches the 32-bit OPC DA wrapper DLLs + // (SPECIFICATION §11). Pick one from the Run and Debug view (Ctrl+Shift+D). + // + // NOTE: `run` / `probe` are Phase 1 stubs until the core engine lands + // (docs/IMPLEMENTATION_PLAN.md §4) — the configs are ready for that day. + "version": "0.2.0", + "configurations": [ + { + // Basic use: foreground monitoring run (the normal way to run the tool). + // Currently prints the Phase 1 stub message; becomes live with the + // core engine. Set breakpoints in main.py / monitor/ to debug it. + "name": "opcda-diagnozer: run (foreground)", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["run", "--config", "config.example.yaml"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, + { + // Basic use: validate the example config without connecting. + "name": "opcda-diagnozer: check-config", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["check-config", "--config", "config.example.yaml"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, + { + // Show all available OPC DA servers (auto: Automation Wrapper over + // COM, falls back to the local registry scan). + "name": "opcda-diagnozer: list-servers", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["list-servers"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, + { + // Show all available OPC DA servers via the no-COM registry scan, + // with full metadata (CLSID, description, DA versions) as JSON. + "name": "opcda-diagnozer: list-servers (registry, JSON)", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["list-servers", "--source", "registry", "--json"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, + { + // Show OPC DA servers on a remote DCOM host — prompts for the host + // name at launch. Needs Launch+Activation permissions for your user + // on the target host (README "DCOM troubleshooting"). + "name": "opcda-diagnozer: list-servers (remote host…)", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["list-servers", "--host", "${input:dcomHost}"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, + { + // List the devices (address-space branches) of the configured + // server — one level, with per-device tag counts. + "name": "opcda-diagnozer: list-devices", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": ["list-devices", "--config", "config.example.yaml"], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + }, + { + // List the tags of one device (whole subtree) as JSON — the + // item_id values are what go into config `tags:`. Adjust the + // device path to the server your config points at. + "name": "opcda-diagnozer: list-tags (recursive, JSON)", + "type": "debugpy", + "request": "launch", + "python": "${workspaceFolder}/.venv/Scripts/python.exe", + "module": "opcda_diagnozer", + "args": [ + "list-tags", + "--config", "config.example.yaml", + "Simulation Items.Bucket Brigade", + "--recursive", + "--json" + ], + "cwd": "${workspaceFolder}", + "console": "integratedTerminal", + "justMyCode": false, + "env": { + "PYTHONUNBUFFERED": "1" + } + } + ], + "inputs": [ + { + "id": "dcomHost", + "type": "promptString", + "description": "Remote host to enumerate OPC DA servers from (via DCOM)", + "default": "CONTROL-HOST01" + } + ] +} \ No newline at end of file diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..093bd59 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,6 @@ +{ + // OPC DA wrappers (OPCAutomation.dll / GBDaCtl.dll) are 32-bit in-proc COM + // DLLs — the interpreter bitness must match (SPECIFICATION §11). + "python.defaultInterpreterPath": "${workspaceFolder}/.venv/Scripts/python.exe", + "python-envs.defaultEnvManager": "ms-python.python:venv" +} \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..1022e9c --- /dev/null +++ b/README.md @@ -0,0 +1,198 @@ +# OPC DA Diagnozer + +A read-only Windows service that acts as an independent OPC DA reference client to +diagnose **disconnections**, **communication latency**, and **data-quality problems** +between a SCADA system and a control system communicating over OPC DA. + +It continuously connects to the OPC DA server, polls a pre-defined list of tags, and +reports every anomaly with precise timestamps — unnatural read latencies, DCOM +disconnects, hung RPC calls, tags marked BAD/UNCERTAIN, stale timestamps, frozen values, +and flapping quality — as structured events (JSONL logs, Windows Event Log, optional +HTTP webhook). + +> **Status:** planning. Read the +> [specification](docs/SPECIFICATION.md) and the +> [implementation plan](docs/IMPLEMENTATION_PLAN.md) first. + +## Why + +OPC DA failures are notoriously hard to pin down because they hide behind generic COM +errors, calls that hang with no timeout, or silently degraded item quality. This tool +answers the triage question *where is the problem?* with evidence: + +| Observation from the Diagnozer | Most likely problem location | +|--------------------------------|------------------------------| +| Clean reads while SCADA complains | SCADA host itself (DCOM perms, CPU, its own load) | +| Diagnozer also sees disconnects/RPC errors | Shared path: network, firewall, DCOM, OPC server host | +| Connection stable but many tags `BAD_COMM_FAILURE` | Link between OPC server and the control devices | +| Hangs without errors | Silent network drop (firewall/MTU) — classic DCOM pitfall | + +See the full interpretation matrix in the [specification](docs/SPECIFICATION.md). + +## Planned features + +- Windows service (auto-start, SCM crash recovery) with a foreground `run` mode for testing +- Local or remote (DCOM) connections to any OPC DA server +- Server browsing: `list-servers` (discovery), `list-devices` and `list-tags` + (address-space walk) — live now +- Latency monitoring: spikes, streaks, and p95 trends against configurable thresholds +- Hang watchdog: flags RPC calls that never return — the silent killer +- Quality monitoring: BAD/UNCERTAIN transitions with decoded subcodes, persistent-bad + and flapping detection, mass-degradation alarm +- Staleness & frozen-value detection while quality still reports GOOD +- Structured events: rotating JSONL, Windows Event Log, optional HTTP webhook +- Strictly read-only; single group, modest poll rate — does not perturb the measured system + +## Quick start + +``` +# list the OPC DA servers available locally (or on a DCOM host with --host) +opcda-diagnozer list-servers +opcda-diagnozer list-servers --host CONTROL-HOST01 # remote, via DCOM +opcda-diagnozer list-servers --source registry --json # no-COM local scan + +# list the devices (address-space branches) of the configured server +opcda-diagnozer list-devices +opcda-diagnozer list-devices --depth 2 # include sub-devices +opcda-diagnozer list-devices --json # script-friendly output + +# list the tags of one device (the item_id values go into config `tags:`) +opcda-diagnozer list-tags "Simulation Items.Bucket Brigade" +opcda-diagnozer list-tags "Simulation Items" "Bucket Brigade" # equivalent +opcda-diagnozer list-tags --recursive "Simulation Items" # whole subtree +opcda-diagnozer list-tags # every tag on the server + +# validate config +opcda-diagnozer check-config + +# connect once and read every configured tag, print results +opcda-diagnozer probe + +# foreground run (console) — best for first deployment validation +opcda-diagnozer run + +# install as Windows service (dedicated account required for DCOM auth) +opcda-diagnozer install --account DOMAIN\svc-opcdiag +``` +`list-servers` uses two discovery mechanisms (`--source`): the OPC DA Automation +Wrapper's `GetOPCServers` over COM (works remotely via DCOM; needs Launch+Activation +permissions on the target host), and a local COM-registration scan that reads the OPC +DA component categories directly from the registry — no COM, useful on locked-down +hosts or when the wrapper DLL is not registered. `auto` tries the first, falls back to +the second. Note the bitness rule still applies: a 32-bit process enumerates the +32-bit registry view — exactly the servers this tool can load. + +`list-devices` / `list-tags` browse the **connected server's** address space through +the Automation Wrapper's browser (`CreateBrowser`): devices are *branches*, tags are +*leaves*. They work locally and over DCOM (same Launch+Activation requirements). A +device path is either one dotted string (`"Simulation Items.Bucket Brigade"`) or +several quoted branch names — use the latter when a branch name itself contains a +dot. On flat-namespace servers (`organization: flat`) every tag hangs directly off +the root; `list-tags` without a device lists them all. +### Configuration example (`config.example.yaml`) + +```yaml +opc: + server: "Matrikon.OPC.Simulation.1" # server ProgID + host: "CONTROL-HOST01" # DCOM host; omit for local + connect_timeout_s: 10 + +poll: + interval_s: 1.0 + timeout_ms: 5000 # beyond this the call is flagged as a hang + +tags: + - { name: "Channel1.Device1.Counter", expect_change: true } + - { name: "Channel1.Device1.Level" } + +latency: + warn_ms: 1000 + critical_ms: 5000 + slow_poll_streak: 5 + window_polls: 60 + +quality: + persistent_bad_s: 30 + flapping_transitions_per_window: 5 + +stale: + enabled: true + max_age_s: 120 + +report: + heartbeat_s: 300 + jsonl: { dir: "C:/ProgramData/OPCDA-Diagnozer/logs", max_mb: 50, backups: 10 } + eventlog: { source: "OPCDA-Diagnozer", min_severity: WARN } +``` + +## DCOM troubleshooting quick reference + +**Setup checklist for the service account (both hosts):** +1. `dcomcnfg` → Component Services → DCOM Config → the OPC server → Properties: + - Security → Launch & Activation: allow **Remote Activation** for the service account. + - Security → Access Permissions: allow Local + Remote access. + - Identity: prefer a dedicated account over "Launching user". +2. Firewall on the OPC server host: TCP 135 (RPC endpoint mapper) + the RPC dynamic + port range (optionally restricted via `netsh int ipv4 set dynamicport tcp ...`). +3. NTP time sync on both hosts — OPC DA timestamps and staleness detection depend on it. +4. Be aware of post-2022 Windows **DCOM hardening** (KB5004442 family): DCOM activation + requires explicit authentication and may break legacy OPC setups with + `E_ACCESSDENIED`-type failures. + +**Common HRESULTs the tool decodes and reports:** + +| HRESULT | Meaning | +|---------|---------| +| `0x800706BA` | RPC server unavailable — server process down, firewall, or DCOM activation refused | +| `0x800706BE` | RPC call failed mid-flight — server crashed or connection reset | +| `0x80010108` | Object disconnected from clients — OPC server dropped us (crash/restart) | +| `0x80070005` | Access denied — DCOM security misconfiguration | +| `0x80040154` | Class not registered — wrong ProgID, or wrapper/server registered under a different bitness | +| `0xC0040007` | Invalid item ID — tag name wrong in config (surfaced by `probe`) | + +**Bitness warning:** the OPC DA Automation Wrapper (`OPCAutomation.dll`) and the Graybox +wrapper are typically **32-bit in-proc DLLs** — the Python interpreter must match +(32-bit). This is decided and verified in Phase 0 of the implementation plan. + +## Safety + +- The tool never writes to the OPC server — there is no write path in the code. +- It uses one OPC group and a static, small tag list at a modest poll rate (default 1 s) + so it does not load the system it is measuring. + +## Development setup + +The OPC DA wrappers (`OPCAutomation.dll`, `GBDaCtl.dll`) are **32-bit in-proc COM DLLs**, +so the Python interpreter **must be 32-bit** (SPECIFICATION §11 / Phase 0 decision). + +```powershell +# create the 32-bit venv (requires the 32-bit Python installed via the py launcher) +& "C:\Users\\AppData\Local\Programs\Python\Python314-32\python.exe" -m venv .venv +& ".\.venv\Scripts\python.exe" -m pip install -e ".[dev]" + +# verify bitness — must print 32 +& ".\.venv\Scripts\python.exe" -c "import struct; print(struct.calcsize('P') * 8)" + +# run the (COM-free) unit + fake-integration tests +& ".\.venv\Scripts\python.exe" -m pytest +``` + +Notes: +- Dependencies are chosen to be installable without a C compiler: **ruamel.yaml** + (pure Python) is used instead of PyYAML, which publishes no 32-bit CPython 3.14 + wheels and whose sdist requires MSVC. +- COM-dependent modules (Automation Wrapper, Event Log, service) are import-guarded; + the unit test suite runs on any OS via the in-process fake client + (`opcda_diagnozer.opc.fake`). +- 32-bit CPython can be installed with the Python Install Manager + (`py install 3.14 -a x86`) or the installer from python.org ("Download Windows + installer (32-bit)"). + +## Documentation + +- [Specification](docs/SPECIFICATION.md) — detection rules, event model, operational behavior, acceptance criteria +- [Implementation plan](docs/IMPLEMENTATION_PLAN.md) — phases, milestones, testing, risks + +## License + +TBD \ No newline at end of file diff --git a/config.example.yaml b/config.example.yaml new file mode 100644 index 0000000..92eb161 --- /dev/null +++ b/config.example.yaml @@ -0,0 +1,42 @@ +# OPC DA Diagnozer — example configuration. +# Copy to config.yaml and adjust. Every threshold below is configurable +# (defaults live in src/opcda_diagnozer/config.py; rules in docs/SPECIFICATION.md §6). + +opc: + server: "Matrikon.OPC.Simulation.1" # OPC DA server ProgID + #host: "CONTROL-HOST01" # DCOM host; omit for a local server + connect_timeout_s: 10 + +poll: + interval_s: 1.0 # poll cadence; keep modest (read-only tool) + timeout_ms: 5000 # beyond this the call is flagged as a hang (D7) + hard_hang_s: 120 # COM thread abandoned + rebuilt after this (§10.4) + +tags: + - { name: "Channel1.Device1.Counter", expect_change: true } # D13 frozen-value detection applies + - { name: "Channel1.Device1.Level" } + +latency: + warn_ms: 1000 # D4 single-spike threshold + critical_ms: 5000 # D5 sustained-slow severity + slow_poll_streak: 5 # D5 consecutive slow polls + window_polls: 60 # D6 rolling window for the p95 trend + +quality: + persistent_bad_s: 30 # D10 single tag BAD longer than this + flapping_transitions_per_window: 5 # D11 quality transitions per window... + flapping_window_s: 600 # ...within this window (10 min) + +stale: + enabled: true + max_age_s: 120 # D12 device timestamp older than this while GOOD + +report: + heartbeat_s: 300 # D15 periodic stats snapshot + jsonl: + dir: "C:/ProgramData/OPCDA-Diagnozer/logs" + max_mb: 50 # rotate after this size + backups: 10 # rotated files kept + eventlog: + source: "OPCDA-Diagnozer" + min_severity: WARN # INFO / WARN / CRITICAL \ No newline at end of file diff --git a/docs/IMPLEMENTATION_PLAN.md b/docs/IMPLEMENTATION_PLAN.md new file mode 100644 index 0000000..cfaacf1 --- /dev/null +++ b/docs/IMPLEMENTATION_PLAN.md @@ -0,0 +1,211 @@ +# OPC DA Diagnozer — Implementation Plan + +Companion to [`docs/SPECIFICATION.md`](SPECIFICATION.md). Targets **v1.0** as the first +production-deployable release. + +--- + +## 1. Milestones + +| Milestone | Content | Exit condition | +|-----------|---------|----------------| +| **v0.1 — Foreground proof** | CLI reads configured tags once per second in a console, prints events | Successfully reads from a real (or simulation) OPC DA server | +| **v0.2 — Detection engine** | All detectors (D1–D15), event model, JSONL logging | Unit tests green; chaos scenarios produce correct events | +| **v0.5 — Windows service** | pywin32 service, watchdog/hang recovery, Event Log, CLI install/run | Runs unattended through reboot + server-kill test | +| **v1.0 — Production** | Packaging, runbook, 24 h soak, optional webhook | Acceptance criteria AC1–AC10 (spec §13) met | + +## 2. Project layout + +``` +opcda-diagnozer/ +├── README.md +├── pyproject.toml +├── config.example.yaml +├── docs/ +│ ├── SPECIFICATION.md +│ └── IMPLEMENTATION_PLAN.md +├── src/opcda_diagnozer/ +│ ├── __init__.py +│ ├── __main__.py # python -m opcda_diagnozer +│ ├── main.py # CLI: run / install / uninstall / check-config / probe / list-devices / list-tags +│ ├── config.py # pydantic-free typed config (dataclasses) + YAML load + validation +│ ├── events.py # DiagEvent, Severity, EventKind +│ ├── quality.py # OPC DA quality decode tables +│ ├── opc/ +│ │ ├── base.py # OpcClient + Browser protocols, ReadResult, OpcError(hresult) +│ │ ├── discovery.py # list-servers: Automation GetOPCServers / registry scan +│ │ ├── browse.py # list-devices/list-tags: address-space walk (branches/leaves) +│ │ ├── automation.py # OPC DA Automation Wrapper adapter (OPCAutomation.dll) +│ │ └── fake.py # in-process fake client + fake address space (tests, demo) +│ ├── monitor/ +│ │ ├── state.py # connection state machine + backoff +│ │ ├── poller.py # COM thread: connect → poll loop → feed detectors +│ │ └── detectors.py # D1–D15 rule implementations +│ ├── report/ +│ │ ├── base.py # Reporter protocol +│ │ ├── jsonl.py # rotating JSONL file +│ │ ├── eventlog.py # Windows Event Log (WARN+) +│ │ └── http.py # optional webhook +│ └── service.py # pywin32 ServiceFramework wrapper +└── tests/ + ├── test_quality.py + ├── test_detectors.py + ├── test_state.py + ├── test_config.py + └── test_poller_fake.py # end-to-end against opc.fake, no COM needed +``` + +## 3. Phase 0 — Feasibility & environment (≈0.5–1 day) + +**Goal:** validate the COM access strategy against reality before building anything. + +Tasks: +1. Identify target OPC DA server: ProgID, host, DA version, bitness (32/64-bit EXE?). +2. Choose wrapper backend: + - **A. OPC DA Automation Wrapper** (`OPCAutomation.dll`) — ubiquitous in industry, + ships with many SCADA installs; likely default. + - **B. OpenOPC / Graybox wrapper** (`GBDaCtl.dll`) — fallback. + - Note: both are typically **32-bit in-proc DLLs** → client Python must be 32-bit + (see spec §11). Decide interpreter bitness here. +3. Stand up a **test server** (Matrikon OPC Simulation Server, KepServerEX demo, or the + actual control-system server in a maintenance window) and successfully read one tag + from Python (`win32com.client.Dispatch`). +4. Verify DCOM configuration for the service account (`dcomcnfg`: Launch/Activation → + Remote Activation; Access → Local+Remote). Record required firewall ports + (TCP 135 + RPC dynamic range). +5. Record baseline behavior: normal read RTT, quality values, what error the wrapper + surfaces when the server is stopped (feeds D1 implementation). + +**Exit:** a throwaway script reads ≥ 3 tags from the remote server with measured RTT. + +## 4. Phase 1 — Core engine (≈2–3 days) + +1. `config.py` — YAML config (see `config.example.yaml` sketch in README), dataclass + model, strict validation, `check-config` messages that explain exactly what is wrong. +2. `quality.py` — decode tables (spec §7), `decode_quality(raw: int) -> QualityInfo`. +3. `opc/base.py` — client abstraction: + ```python + class OpcClient(Protocol): + def connect(self) -> None: ... + def read(self, tags: Sequence[str]) -> list[ReadResult]: ... + def close(self) -> None: ... + ``` + `ReadResult`: `tag, value, quality_raw, device_time, error` — plus per-call RTT + measured by the poller. COM errors are normalized to `OpcError` carrying the raw + `HRESULT` + decoded name. +4. `opc/automation.py` — Automation Wrapper adapter (single group, static items, + `SyncRead`); every COM exception mapped to `OpcError` with `win32com` scode. +5. `monitor/state.py` — state machine (spec §10.2), backoff, reconnect accounting. +6. `monitor/detectors.py` — D1–D15 as small pure functions/classes over + `PollOutcome` (list[ReadResult] + rtt_ms + error), using an injected clock for + testability. Rolling window (deque, size `window_polls`) for p95/streak logic. +7. `monitor/poller.py` — the single COM thread; watchdog logic for hang detection + (spec §10.4): a second thread timestamps poll starts/ends; on `hard_hang_s` the COM + thread is abandoned and rebuilt. +8. `events.py` — event dataclass + JSON serialization. +9. `opc/fake.py` — scriptable fake client (sequences of outcomes: good reads, errors, + hangs, quality flips) → powers the entire unit-test suite without COM/hardware. + +**Exit:** `opcda-diagnozer run` (foreground console) produces correct events against the +fake client; unit tests for all detectors. + +## 5. Phase 2 — Reporters (≈1 day) + +1. `report/jsonl.py` — rotation (size + backups), atomic-ish append, header line with + config digest for provenance. +2. `report/eventlog.py` — pywin32 `win32evtlog` usage; map severities; source + registration via `install` step. +3. `report/http.py` — optional webhook with timeout + simple retry/drop on failure + (never blocks the poll loop; queue with bound). +4. Heartbeat assembly (D15): uptime, reconnects, RTT percentiles, tags-bad counts. + +**Exit:** events visible in file + Event Log; heartbeat cadence verified. + +## 6. Phase 3 — Windows service (≈1 day) + +1. `service.py` — `win32serviceutil.ServiceFramework`; service host thread spawns the + COM worker; clean `stop` handling (`SetServiceStatus STOP_PENDING`, join with + timeout, abandon hung COM thread on shutdown). +2. CLI verbs: `install /uninstall /start /stop` (wraps `sc.exe` semantics), + `run` (foreground), `check-config`, `probe` (connect once, print one read), + `list-items` (browse validation tool). +3. SCM recovery config (`sc failure ... restart`), auto-start, service account + documentation. +4. Config/log default dir: `%ProgramData%\OPCDA-Diagnozer\`. + +**Exit:** survives reboot; survives OPC-server kill (produces D1/D2 events, reconnects); +`probe` and `check-config` give usable diagnostics to an ops engineer. + +## 7. Phase 4 — Integration & chaos testing (≈1–2 days) + +Scenario matrix against a simulation server (and, when available, the real server): + +| Scenario | How to inject | Expected events | +|----------|---------------|-----------------| +| Server process killed | `taskkill` | D1 (HRESULT), D2 with outage duration | +| Server service stopped/restarted | `net stop/start` | D1, D2 | +| Firewall drop (silent) | block outbound RPC ports / pull NIC | D7 hang, then D1 | +| Latency injection | Clumsy / `netsh` delay | D4 spikes, D5 streak, D6 trend | +| Device comm failure | simulation server "communication disturbance" items | D8 with `BAD_COMM_FAILURE`, D10 persistence | +| Quality oscillation | toggling sim items | D11 flapping | +| Stale/frozen tag | simulation server static item marked `expect_change` | D12 / D13 | +| Mass bad quality | all sim items to bad | D14 | +| Config errors | bad ProgID, unknown tag | fast fail + actionable message; `OPC_E_INVALIDITEMID` surfaced at startup | +| Soak | 24 h healthy run | no leaks, heartbeats every 5 min, zero false positives | + +**Exit:** each row verified manually against the event log; AC2/AC3/AC4/AC5 confirmed. + +## 8. Phase 5 — Packaging & docs (≈0.5–1 day) + +- `pyproject.toml` wheel installable with pip into the chosen (32-bit) Python; + document the on-host install procedure (PyInstaller packaging for pywin32 services is + fragile — deferred). +- `README.md` finalized (install, config, DCOM checklist, HRESULT table). +- Ops runbook: how to read events, what each kind means (link to spec §12 matrix), + how to reconfigure thresholds, where evidence lives. +- Version 1.0.0 tag. + +## 9. Testing strategy + +- **Unit** (no COM): detectors against scripted `PollOutcome` sequences with a fake + clock; quality decode; config validation; backoff/state transitions; JSONL rotation. +- **Integration:** `test_poller_fake.py` runs the full poller+reporters against + `opc.fake` in-process — the whole event pipeline without Windows COM. +- **Manual/chaos:** Phase 4 matrix (necessarily on Windows against a real/simulated + OPC DA server — cannot be automated in CI without the DCOM environment). +- CI: unit + fake-integration tests run on any OS (COM adapters are import-guarded); + chaos tests are a documented manual procedure. + +## 10. Risk register + +| Risk | Impact | Mitigation | +|------|--------|------------| +| Wrapper DLL bitness mismatch (32-bit only) | Cannot load wrapper | Phase 0 decision; use 32-bit Python; keep client abstraction so backend can be swapped | +| Automation Wrapper not present on diag host | No COM access | Phase 0 check; ship/distribute wrapper install in runbook (vendor redistribution rules apply) | +| DCOM hardening (KB5004442) blocks activation | Connect fails with `E_ACCESSDENIED`/activation errors | Runbook covers `dcomcnfg` steps; events surface raw HRESULT | +| COM calls hang indefinitely | Service appears alive but blind | Explicit watchdog + thread recycling (spec §10.4) | +| COM apartment violations (cross-thread calls) | Random `RPC_E_WRONG_THREAD` crashes | Single dedicated COM thread; all COM handles stay on it | +| Our client perturbs the system | Skewed diagnosis | One group, static small tag list, 1 s default poll, read-only | +| pywin32 service + COM quirks (GIL, STA init) | Service instability | Phase 3 spike; foreground `run` mode always available as fallback | +| Tag list too large / wrong item IDs | Noise, misleading RTT | `probe` + `check-config` validate up-front; per-poll error accounting | +| SCADA/host clock skew | Misleading staleness events | Local receive time always recorded; `max_age_s` configurable | + +## 11. Effort estimate + +| Phase | Estimate | +|-------|----------| +| 0 Feasibility | 0.5–1 d | +| 1 Core engine | 2–3 d | +| 2 Reporters | 1 d | +| 3 Service | 1 d | +| 4 Chaos testing | 1–2 d | +| 5 Packaging/docs | 0.5–1 d | +| **Total v1.0** | **≈ 6–9 days** | + +## 12. Definition of done (v1.0) + +- All Phase 4 scenarios pass; AC1–AC10 (spec §13) verified. +- 24 h soak clean. +- README + runbook complete enough for an ops engineer to deploy unaided. +- Code reviewed; no write path to OPC anywhere in the codebase. +- Config example covers every configurable threshold. \ No newline at end of file diff --git a/docs/SPECIFICATION.md b/docs/SPECIFICATION.md new file mode 100644 index 0000000..a54a274 --- /dev/null +++ b/docs/SPECIFICATION.md @@ -0,0 +1,263 @@ +# OPC DA Diagnozer — Project Specification + +**Version:** 0.1 (draft) +**Status:** Planning +**Target platform:** Windows only (OPC DA is a COM/DCOM technology) + +--- + +## 1. Problem statement + +We experience intermittent **disconnections** and/or **high communication latency** between a +SCADA system and a control system that communicate over **OPC DA** (Data Access, the +COM/DCOM-based predecessor of OPC UA). Today we have no objective evidence to answer the +key triage question: + +> *Is the problem in the SCADA host, the network/DCOM layer, the OPC DA server, or the +> link between the OPC server and the control devices?* + +OPC DA failure modes are notoriously hard to diagnose because failures surface as generic +`HRESULT` errors, hung calls with **no client-side timeout**, or silently degraded item +quality — long before the SCADA operator sees anything. + +## 2. Approach + +Deploy an **independent, read-only OPC DA reference client** that continuously exercises +the same path the SCADA uses: + +- Connect to the OPC DA server (local or remote via DCOM) from a Windows machine. +- Poll a small, pre-defined list of tags at a fixed interval. +- Measure **read round-trip time (RTT)**, track **item quality** (GOOD / UNCERTAIN / BAD + and subcodes), **device timestamps**, and **connection state**. +- Emit structured, timestamped **diagnostic events** whenever anything deviates from + normal — disconnects, slow reads, hung calls, quality degradation, frozen/stale values. + +Because the client is independent of the SCADA, its observations provide a **reference +baseline** against which the SCADA's complaints can be triangulated (see §12). + +## 3. Deployment architecture + +```mermaid +flowchart LR + subgraph SCADAzone["SCADA zone (or dedicated diag host)"] + DIAG["OPC DA Diagnozer
(Windows service)"] + end + subgraph CTL["Control system / OPC DA server host"] + OPC["OPC DA server
(COM/DCOM out-of-proc)"] + DEV["Control devices / PLC link"] + end + DIAG -- "DCOM reads (RPC 135 + dynamic ports)" --> OPC + OPC -.-> DEV + DIAG -- "Events (JSONL / Event Log / HTTP)" --> OPS["Engineers & monitoring"] +``` + +Placement note: run the tool **as close to the SCADA machine as possible** (same host or +same network segment) so its measured latencies are comparable to what the SCADA +experiences. Optionally run a second instance next to the OPC server host for differential +diagnosis. + +## 4. Goals + +1. Run unattended as a **Windows service** (auto-start, SCM crash recovery). +2. Connect to a configured OPC DA server (local or remote DCOM) and read a configured tag list. +3. Detect and report, with precise timestamps: + - disconnections and reconnects (including DCOM error codes), + - hung / timed-out RPC calls, + - read latencies exceeding thresholds (single spikes, streaks, and trends), + - tags whose quality is BAD or UNCERTAIN (with decoded subcode), + - persistent-bad and flapping (oscillating) quality per tag, + - stale device timestamps / frozen values while quality reports GOOD, + - mass quality degradation (many tags BAD at once → control-link problem). +4. Produce durable, machine-readable evidence (structured logs + Windows Event Log + + optional HTTP webhook) suitable for incident review and correlation. +5. Be **strictly read-only** and lightweight — must not perturb the system it measures. + +## 5. Non-goals (v1) + +- No writes to the OPC server (no `Write` code path at all). +- No OPC UA, OPC HDA, or Alarms & Events support. +- No data historian / storage of every sample (only events + periodic summaries). +- No UI/dashboard in v1 (JSONL logs + webhook are enough for Loki/Grafana ingestion). +- No automatic remediation (diagnose only). + +## 6. Detection rules + +| ID | Condition | Default rule | Severity | +|----|-----------|--------------|----------| +| D1 | Connection lost | Any read/connect COM failure | CRITICAL (event incl. decoded HRESULT) | +| D2 | Reconnect succeeded | After D1, reconnect works | INFO (event with outage duration) | +| D3 | Reconnect flapping | > 3 disconnects in 10 min | CRITICAL | +| D4 | Slow poll (spike) | Single read RTT > `warn_ms` (default 1000) | WARN | +| D5 | Sustained slow polls | RTT > `warn_ms` for `slow_poll_streak` (5) consecutive polls | CRITICAL | +| D6 | Latency trend | p95 RTT over last `window_polls` (60) > `warn_ms` | WARN | +| D7 | Poll hang | Read call does not return within `timeout_ms` (5000) — see §10.4 | CRITICAL (immediate), see recovery behavior | +| D8 | Tag quality → BAD | Any transition into BAD, decoded subcode | WARN (per tag) | +| D9 | Tag quality → UNCERTAIN | Any transition into UNCERTAIN | INFO | +| D10 | Persistent bad tag | Single tag BAD longer than `persistent_bad_s` (30) | WARN (escalated from D8) | +| D11 | Quality flapping | > 5 quality transitions for one tag within 10 min | WARN | +| D12 | Stale timestamp | Device timestamp older than `max_age_s` (120) while quality GOOD | WARN | +| D13 | Frozen value | `expect_change: true` tag with unchanged value+timestamp for `max_age_s` | WARN | +| D14 | Mass quality degradation | ≥ 50% of monitored tags BAD simultaneously while connection OK | CRITICAL (`CONTROL_LINK_DEGRADED` — points at OPC-server↔device link, not DCOM) | +| D15 | Heartbeat / summary | Periodic stats snapshot every `heartbeat_s` (300) | INFO | + +All thresholds are configurable; severities are INFO / WARN / CRITICAL. + +## 7. OPC DA quality codes (reference) + +OPC DA quality is a 16-bit field. Bits 7–6 of the low byte give the class: +`0xC0` GOOD, `0x40` UNCERTAIN, `0x00` BAD. Bits 5–0 are the substatus. Values the tool +decodes and names: + +| Code | Meaning | +|------|---------| +| `0xC0` | GOOD | +| `0xD8` | GOOD / local override | +| `0x44` | UNCERTAIN / last usable value | +| `0x40` | UNCERTAIN (unspecified) | +| `0x00` | BAD (unspecified) | +| `0x04` | BAD / configuration error | +| `0x08` | BAD / not connected | +| `0x0C` | BAD / device failure | +| `0x10` | BAD / sensor failure | +| `0x14` | BAD / last known value | +| `0x18` | BAD / communication failure | +| `0x1C` | BAD / out of service | + +Remaining UNCERTAIN subcodes (DA 2.05a) are decoded by the same table in `quality.py`. +Every quality event carries the raw hex code **and** the decoded name. + +## 8. Event model + +Every event is a structured record: + +```json +{ + "time_utc": "2026-10-07T12:34:56.789Z", + "kind": "QUALITY_BAD", + "severity": "WARN", + "detail": { + "tag": "Channel1.Device1.Counter", + "quality_raw": "0x18", + "quality": "BAD_COMM_FAILURE", + "previous_quality": "GOOD", + "value": 1234.0, + "device_time_utc": "2026-10-07T12:34:55.100Z", + "rtt_ms": 42.1 + }, + "context": { + "uptime_s": 3600.5, + "reconnects_total": 2, + "rtt_p95_ms": 55.2, + "tags_bad": 3, + "tags_total": 25 + } +} +``` + +`time_utc` is always the **local client receive time** (wall clock), so events remain +comparable even if the remote host's clock is wrong. Device timestamps are reported +alongside and clearly labeled. + +## 9. Outputs (reporters) + +| Reporter | Purpose | Notes | +|----------|---------|-------| +| JSONL rotating file | Primary evidence trail | One JSON event per line; rotation by size; default dir `%ProgramData%\OPCDA-Diagnozer\logs` | +| Windows Event Log | Ops visibility | WARN+ only; source `OPCDA-Diagnozer` | +| Console | Foreground/debug runs | Human-readable | +| HTTP webhook (optional) | Alert integration | POST one JSON event per request, optional bearer token | + +Reporters are pluggable (`Reporter.emit(event)`); a heartbeat event every 5 minutes +proves liveness and enables baseline graphs even when everything is healthy. + +## 10. Operational behavior + +### 10.1 Run as a Windows service +- Registered via pywin32 (`opcda-diagnozer install`), auto-start, runs under a + **dedicated service account** (required for DCOM authentication, see §11). +- SCM recovery: restart on failure (first/second failure), plus internal watchdog. + +### 10.2 Connection state machine +```mermaid +stateDiagram-v2 + [*] --> Connecting + Connecting --> Connected : connect OK + Connecting --> ReconnectWait : COM error + Connected --> ReconnectWait : read error / hang + ReconnectWait --> Connecting : backoff elapsed +``` +Reconnect backoff: exponential 1 s → 2 s → … capped at 60 s. Every disconnect/reconnect +pair is an event; outage duration is computed and reported. + +### 10.3 COM apartment handling +All OPC/COM work runs on one dedicated STA thread inside the service process +(`pythoncom.CoInitialize()` on that thread only). Detectors run on the same thread (pure +Python, cheap); event reporting happens off-thread via a queue so slow I/O never blocks +polling. + +### 10.4 The DCOM hang problem (important) +DCOM/RPC calls have **no client-side timeout**. If the network path breaks silently +(e.g., firewall drops), a read call can block for tens of seconds to minutes without +erroring. The tool handles this explicitly: + +1. A watchdog flags a **poll hang** (D7) as soon as a call exceeds `timeout_ms`. +2. If the hang exceeds `hard_hang_s` (default 120 s), the service abandons the stuck + thread and creates a fresh connection on a new thread (stale `RPC` objects are + released by COM). +3. Hang events are distinguished from clean disconnects — this is often the smoking gun + for silent network drops. + +### 10.5 Safety constraints +- Read-only by construction: no write API is called; the code has no write path. +- Single OPC group, static item list, modest poll rate (default 1 s) to keep added + server load negligible. +- Configuration errors fail fast at startup with actionable messages + (`opcda-diagnozer check-config` validates without connecting… and `--probe` connects + once and reports). + +## 11. Environment & constraints + +| Constraint | Consequence | +|------------|-------------| +| OPC DA is Windows/COM only | Service runs on Windows only (Server 2012 R2 … 2022, Win 10/11) | +| OPC DA wrappers are often **32-bit** DLLs (OPC DA Automation Wrapper `OPCAutomation.dll`, Graybox `GBDaCtl.dll`) | Python interpreter bitness must match the wrapper DLL — likely **32-bit Python** for v1; decision in Phase 0 | +| DCOM security | The service account needs Launch+Activation and Access permissions on the OPC server (via `dcomcnfg`); account must authenticate across the network (not `LocalSystem`) | +| Post-2022 Windows DCOM hardening (KB5004442 family) | DCOM activation now requires explicit authentication; legacy OPC setups may fail activation — errors must be reported clearly | +| RPC over TCP | Firewall: TCP 135 + RPC dynamic port range on the server host; misconfig shows up as `RPC_S_SERVER_UNAVAILABLE` | +| Clock skew | Device timestamps may be misleading; every event also carries local receive time; recommend NTP on both hosts | + +## 12. Diagnostic interpretation matrix (why this tool exists) + +| Observation from the Diagnozer | Most likely problem location | +|--------------------------------|------------------------------| +| Clean reads, low RTT, all GOOD — while SCADA still complains | SCADA host itself: per-machine DCOM permissions, CPU/memory saturation, its own group/subscription load | +| Diagnozer also sees disconnects / RPC errors | Shared path: network, firewall, DCOM layer, or OPC server host | +| Connection stable, but `BAD_COMM_FAILURE` on many device tags | Link between the OPC server and the control devices (downstream of OPC) — **not** a DCOM/network problem | +| Slow RTT spikes at consistent times of day | Server load pattern; correlate with heartbeat summaries | +| Hangs without errors returned (D7) | Silent network drop (firewall/MTU/NIC) — classic DCOM pitfall | +| One instance (SCADA zone) slow, other (server zone) fast | Network segment between the zones | + +## 13. Acceptance criteria (v1.0) + +- **AC1** Runs as a service, survives reboot, auto-starts, recovers from a killed process. +- **AC2** Detects OPC server process stop within one poll interval (or one hang window for + silent drops). +- **AC3** Detects injected latency above thresholds within 1–2 polls; event contains + measured RTT. +- **AC4** Quality transitions produce events with raw code and decoded name. +- **AC5** Staleness/freeze detection within tolerance of `max_age_s`. +- **AC6** No OPC writes are ever issued (code review + no write path exists). +- **AC7** Events land in JSONL and Event Log; heartbeats appear even when all is healthy. +- **AC8** Invalid config fails fast with actionable messages. +- **AC9** Footprint ≤ ~50 MB RAM, ~0% CPU while idle-polling on a healthy link. +- **AC10** 24 h soak against a simulation server: no leaks, correct reconnect accounting. + +## 14. Future extensions (post-v1) + +- Async subscription monitoring (`IOPCAsyncIO2` DataChange) to compare push vs. poll + latency — validates the SCADA's actual callback path. +- Per-item latency bisection (read items individually to isolate a slow tag). +- SQLite time-series of samples for post-incident graphing. +- Prometheus exporter / Grafana dashboard. +- Email / Teams / webhook alerting with deduplication. +- Multi-server monitoring in one service. \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9052a3c --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,34 @@ +[build-system] +requires = ["setuptools>=69"] +build-backend = "setuptools.build_meta" + +[project] +name = "opcda-diagnozer" +version = "0.1.0" +description = "Read-only OPC DA reference client that diagnoses disconnections, latency and data-quality problems between SCADA and control systems" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + # ruamel.yaml is pure Python: PyYAML publishes no win32 (32-bit) wheels for + # CPython 3.14 and its sdist build requires MSVC, which diag hosts lack + # (the interpreter itself must be 32-bit — SPECIFICATION §11 bitness rule). + "ruamel.yaml>=0.18", + # OPC DA is Windows/COM-only; every pywin32 import in the code is guarded + # so tests can run on any OS (IMPLEMENTATION_PLAN §9). + "pywin32>=306; sys_platform == 'win32'", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", +] + +[project.scripts] +opcda-diagnozer = "opcda_diagnozer.main:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +testpaths = ["tests"] +addopts = "-ra" \ No newline at end of file diff --git a/src/opcda_diagnozer/__init__.py b/src/opcda_diagnozer/__init__.py new file mode 100644 index 0000000..4935da2 --- /dev/null +++ b/src/opcda_diagnozer/__init__.py @@ -0,0 +1,13 @@ +"""OPC DA Diagnozer — a read-only OPC DA reference diagnostic client. + +Continuously polls a configured tag list over OPC DA and reports anomalies — +disconnections, latency, hung calls, quality degradation, stale/frozen values — +as structured events. See docs/SPECIFICATION.md. + +Strictly read-only by construction: there is no write path to the OPC server +anywhere in this codebase (SPECIFICATION §10.5). +""" + +__version__ = "0.1.0" + +__all__ = ["__version__"] \ No newline at end of file diff --git a/src/opcda_diagnozer/__main__.py b/src/opcda_diagnozer/__main__.py new file mode 100644 index 0000000..3c37a61 --- /dev/null +++ b/src/opcda_diagnozer/__main__.py @@ -0,0 +1,6 @@ +"""Support for ``python -m opcda_diagnozer``.""" + +from opcda_diagnozer.main import main + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/src/opcda_diagnozer/config.py b/src/opcda_diagnozer/config.py new file mode 100644 index 0000000..a06888f --- /dev/null +++ b/src/opcda_diagnozer/config.py @@ -0,0 +1,255 @@ +"""Typed configuration: dataclasses, YAML loading and strict validation. + +Pydantic-free by design (IMPLEMENTATION_PLAN §4.1). Every validation error is +actionable: it names the offending key and explains what is wrong, so +``opcda-diagnozer check-config`` output lets an ops engineer fix the file +without reading the code. +""" + +from __future__ import annotations + +from dataclasses import MISSING, dataclass, field, fields, replace +from pathlib import Path +from typing import Any + +from ruamel.yaml import YAML +from ruamel.yaml.error import YAMLError + +_yaml = YAML(typ="safe") # pure-Python safe loader; config files are small + + +class ConfigError(Exception): + """Invalid configuration; the message is user-facing (check-config output).""" + + +@dataclass(frozen=True) +class OpcConfig: + # server: OPC DA server ProgID, e.g. "Matrikon.OPC.Simulation.1" + # host: DCOM host; None means a local (same-machine) server + server: str + host: str | None = None + connect_timeout_s: float = 10.0 + + +@dataclass(frozen=True) +class TagEntry: + # expect_change: True -> frozen-value detection (D13) applies to this tag + name: str + expect_change: bool = False + + +@dataclass(frozen=True) +class PollConfig: + interval_s: float = 1.0 + timeout_ms: int = 5000 # D7: beyond this the call is flagged as a hang + hard_hang_s: float = 120.0 # SPECIFICATION §10.4: abandon + rebuild the COM thread + + +@dataclass(frozen=True) +class LatencyConfig: + warn_ms: float = 1000.0 # D4 spike threshold + critical_ms: float = 5000.0 # D5 sustained-slow severity + slow_poll_streak: int = 5 # D5 consecutive slow polls + window_polls: int = 60 # D6 rolling window for the p95 trend + + +@dataclass(frozen=True) +class QualityConfig: + persistent_bad_s: float = 30.0 # D10 + flapping_transitions_per_window: int = 5 # D11 + flapping_window_s: float = 600.0 # D11 window (10 min) + + +@dataclass(frozen=True) +class StaleConfig: + enabled: bool = True + max_age_s: float = 120.0 # D12 / D13 + + +@dataclass(frozen=True) +class JsonlConfig: + dir: str = "C:/ProgramData/OPCDA-Diagnozer/logs" + max_mb: int = 50 + backups: int = 10 + + +@dataclass(frozen=True) +class EventLogConfig: + source: str = "OPCDA-Diagnozer" + min_severity: str = "WARN" # INFO / WARN / CRITICAL + + +@dataclass(frozen=True) +class ReportConfig: + heartbeat_s: float = 300.0 # D15 + jsonl: JsonlConfig = field(default_factory=JsonlConfig) + eventlog: EventLogConfig = field(default_factory=EventLogConfig) + + +@dataclass(frozen=True) +class Config: + """Fully validated tool configuration (see config.example.yaml).""" + + opc: OpcConfig + poll: PollConfig = field(default_factory=PollConfig) + tags: tuple[TagEntry, ...] = () + latency: LatencyConfig = field(default_factory=LatencyConfig) + quality: QualityConfig = field(default_factory=QualityConfig) + stale: StaleConfig = field(default_factory=StaleConfig) + report: ReportConfig = field(default_factory=ReportConfig) + + +_ALLOWED_TOP_LEVEL = {"opc", "poll", "tags", "latency", "quality", "stale", "report"} +_TAG_KEYS = {"name", "expect_change"} + + +def _build_section(cls: type[Any], data: Any, where: str) -> Any: + """Instantiate a frozen dataclass section, rejecting unknown/missing keys.""" + data = {} if data is None else data + if not isinstance(data, dict): + raise ConfigError(f"{where}: must be a mapping, got {type(data).__name__}") + known = {f.name for f in fields(cls)} + unknown = sorted(set(data) - known) + if unknown: + raise ConfigError(f"{where}: unknown key(s) {unknown}; accepted keys: {sorted(known)}") + for f in fields(cls): + if f.name not in data and f.default is MISSING and f.default_factory is MISSING: + raise ConfigError(f"{where}: missing required key '{f.name}'") + try: + return cls(**data) + except TypeError as exc: + raise ConfigError(f"{where}: {exc}") from None + + +def _build_report(data: Any) -> ReportConfig: + data = {} if data is None else dict(data) + jsonl = _build_section(JsonlConfig, data.pop("jsonl", None), "report.jsonl") + eventlog = _build_section(EventLogConfig, data.pop("eventlog", None), "report.eventlog") + rest = _build_section(ReportConfig, data, "report") + return replace(rest, jsonl=jsonl, eventlog=eventlog) + + +def _build_tags(data: Any) -> tuple[TagEntry, ...]: + if data is None: + return () + if not isinstance(data, list): + raise ConfigError(f"tags: must be a list, got {type(data).__name__}") + entries: list[TagEntry] = [] + for i, item in enumerate(data): + if isinstance(item, str): + if not item.strip(): + raise ConfigError(f"tags[{i}]: name must be a non-empty string") + entries.append(TagEntry(name=item)) + continue + if not isinstance(item, dict): + raise ConfigError( + f"tags[{i}]: must be a tag name string or a mapping with 'name'" + " (and optional 'expect_change')" + ) + name = item.get("name") + if not isinstance(name, str) or not name.strip(): + raise ConfigError(f"tags[{i}]: 'name' is required and must be a non-empty string") + unknown = sorted(set(item) - _TAG_KEYS) + if unknown: + raise ConfigError(f"tags[{i}]: unknown key(s) {unknown}; accepted keys: {sorted(_TAG_KEYS)}") + expect = item.get("expect_change", False) + if not isinstance(expect, bool): + raise ConfigError( + f"tags[{i}]: 'expect_change' must be a boolean, got {type(expect).__name__}" + ) + entries.append(TagEntry(name=name, expect_change=expect)) + return tuple(entries) + + +def _validate(config: Config) -> None: + """Cross-field validation; each message names the offending key.""" + if not config.opc.server or not config.opc.server.strip(): + raise ConfigError( + "opc.server: required — the OPC DA server ProgID, " + "e.g. 'Matrikon.OPC.Simulation.1'" + ) + if config.opc.connect_timeout_s <= 0: + raise ConfigError(f"opc.connect_timeout_s: must be > 0, got {config.opc.connect_timeout_s}") + if not config.tags: + raise ConfigError( + "tags: at least one tag is required — without tags there is nothing to monitor" + ) + names = [t.name for t in config.tags] + duplicates = sorted({n for n in names if names.count(n) > 1}) + if duplicates: + raise ConfigError(f"tags: duplicate tag name(s) {duplicates}") + if config.poll.interval_s <= 0: + raise ConfigError(f"poll.interval_s: must be > 0, got {config.poll.interval_s}") + if config.poll.timeout_ms <= 0: + raise ConfigError(f"poll.timeout_ms: must be > 0, got {config.poll.timeout_ms}") + if config.poll.hard_hang_s * 1000 <= config.poll.timeout_ms: + raise ConfigError( + f"poll.hard_hang_s: must exceed poll.timeout_ms " + f"({config.poll.timeout_ms} ms); got {config.poll.hard_hang_s} s" + ) + if not 0 < config.latency.warn_ms <= config.latency.critical_ms: + raise ConfigError( + "latency.warn_ms: must satisfy 0 < warn_ms <= critical_ms " + f"(got warn_ms={config.latency.warn_ms}, critical_ms={config.latency.critical_ms})" + ) + if config.latency.slow_poll_streak < 1: + raise ConfigError(f"latency.slow_poll_streak: must be >= 1, got {config.latency.slow_poll_streak}") + if config.latency.window_polls < 1: + raise ConfigError(f"latency.window_polls: must be >= 1, got {config.latency.window_polls}") + if config.quality.persistent_bad_s <= 0: + raise ConfigError(f"quality.persistent_bad_s: must be > 0, got {config.quality.persistent_bad_s}") + if config.quality.flapping_transitions_per_window < 1: + raise ConfigError( + f"quality.flapping_transitions_per_window: must be >= 1, " + f"got {config.quality.flapping_transitions_per_window}" + ) + if config.quality.flapping_window_s <= 0: + raise ConfigError(f"quality.flapping_window_s: must be > 0, got {config.quality.flapping_window_s}") + if config.stale.enabled and config.stale.max_age_s <= 0: + raise ConfigError(f"stale.max_age_s: must be > 0, got {config.stale.max_age_s}") + if config.report.heartbeat_s <= 0: + raise ConfigError(f"report.heartbeat_s: must be > 0, got {config.report.heartbeat_s}") + if config.report.eventlog.min_severity not in ("INFO", "WARN", "CRITICAL"): + raise ConfigError( + "report.eventlog.min_severity: must be one of INFO/WARN/CRITICAL, " + f"got {config.report.eventlog.min_severity!r}" + ) + if config.report.jsonl.max_mb < 1: + raise ConfigError(f"report.jsonl.max_mb: must be >= 1, got {config.report.jsonl.max_mb}") + if config.report.jsonl.backups < 0: + raise ConfigError(f"report.jsonl.backups: must be >= 0, got {config.report.jsonl.backups}") + + +def load_config(path: str | Path) -> Config: + """Load and strictly validate a YAML configuration file.""" + p = Path(path) + if not p.is_file(): + raise ConfigError(f"config file not found: {p}") + try: + data = _yaml.load(p.read_text(encoding="utf-8")) + except YAMLError as exc: + raise ConfigError(f"{p}: invalid YAML: {exc}") from None + if data is None: + data = {} + if not isinstance(data, dict): + raise ConfigError(f"{p}: top level must be a mapping") + unknown_top = sorted(set(data) - _ALLOWED_TOP_LEVEL) + if unknown_top: + raise ConfigError( + f"{p}: unknown section(s) {unknown_top}; " + f"accepted sections: {sorted(_ALLOWED_TOP_LEVEL)}" + ) + config = Config( + opc=_build_section(OpcConfig, data.get("opc"), "opc"), + poll=_build_section(PollConfig, data.get("poll"), "poll"), + tags=_build_tags(data.get("tags")), + latency=_build_section(LatencyConfig, data.get("latency"), "latency"), + quality=_build_section(QualityConfig, data.get("quality"), "quality"), + stale=_build_section(StaleConfig, data.get("stale"), "stale"), + report=_build_report(data.get("report")), + ) + try: + _validate(config) + except TypeError as exc: + raise ConfigError(f"invalid value type in config: {exc}") from None + return config \ No newline at end of file diff --git a/src/opcda_diagnozer/events.py b/src/opcda_diagnozer/events.py new file mode 100644 index 0000000..8abb1ab --- /dev/null +++ b/src/opcda_diagnozer/events.py @@ -0,0 +1,87 @@ +"""Event model: DiagEvent, Severity, EventKind (SPECIFICATION §8). + +Every anomaly report is one :class:`DiagEvent`. ``time_utc`` is always the +**local client receive time** (wall clock), so events stay comparable even when +the remote host clock is wrong; device timestamps travel inside ``detail``, +clearly labeled (SPECIFICATION §8). +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from enum import Enum +from typing import Any + + +def utc_now() -> datetime: + """Timezone-aware current UTC time (the local receive clock).""" + return datetime.now(timezone.utc) + + +class Severity(str, Enum): + INFO = "INFO" + WARN = "WARN" + CRITICAL = "CRITICAL" + + +class EventKind(str, Enum): + """Event kinds — one per detection rule D1–D15 (SPECIFICATION §6).""" + + CONNECTED = "CONNECTED" + DISCONNECTED = "DISCONNECTED" # D1 + RECONNECTED = "RECONNECTED" # D2 + RECONNECT_FLAPPING = "RECONNECT_FLAPPING" # D3 + SLOW_POLL = "SLOW_POLL" # D4 + SLOW_POLL_STREAK = "SLOW_POLL_STREAK" # D5 + LATENCY_TREND = "LATENCY_TREND" # D6 + POLL_HANG = "POLL_HANG" # D7 + QUALITY_BAD = "QUALITY_BAD" # D8 + QUALITY_UNCERTAIN = "QUALITY_UNCERTAIN" # D9 + QUALITY_BAD_PERSISTENT = "QUALITY_BAD_PERSISTENT" # D10 + QUALITY_FLAPPING = "QUALITY_FLAPPING" # D11 + STALE_TIMESTAMP = "STALE_TIMESTAMP" # D12 + FROZEN_VALUE = "FROZEN_VALUE" # D13 + CONTROL_LINK_DEGRADED = "CONTROL_LINK_DEGRADED" # D14 + HEARTBEAT = "HEARTBEAT" # D15 + + +@dataclass(frozen=True) +class EventContext: + """Snapshot counters attached to every event (SPECIFICATION §8).""" + + uptime_s: float + reconnects_total: int + rtt_p95_ms: float | None = None + tags_bad: int = 0 + tags_total: int = 0 + + +@dataclass(frozen=True) +class DiagEvent: + """One structured diagnostic record — JSONL-ready via :meth:`to_json`.""" + + time_utc: datetime + kind: EventKind + severity: Severity + detail: dict[str, Any] = field(default_factory=dict) + context: EventContext | None = None + + def to_json(self) -> str: + """Serialize as one JSON line (SPECIFICATION §8, JSONL append-safe).""" + if self.time_utc.tzinfo is None: + raise ValueError("time_utc must be timezone-aware (use events.utc_now())") + payload: dict[str, Any] = { + "time_utc": _utc_z(self.time_utc), + "kind": self.kind.value, + "severity": self.severity.value, + "detail": self.detail, + "context": asdict(self.context) if self.context is not None else None, + } + return json.dumps(payload, ensure_ascii=False) + + +def _utc_z(dt: datetime) -> str: + """ISO-8601 UTC with millisecond precision and a Z suffix.""" + return dt.astimezone(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z") \ No newline at end of file diff --git a/src/opcda_diagnozer/main.py b/src/opcda_diagnozer/main.py new file mode 100644 index 0000000..d0f6c4d --- /dev/null +++ b/src/opcda_diagnozer/main.py @@ -0,0 +1,377 @@ +"""Command-line interface (IMPLEMENTATION_PLAN §6.2). + +Verbs: ``run``, ``probe``, ``check-config``, ``list-items`` plus the service +verbs ``install`` / ``uninstall`` / ``start`` / ``stop``. Currently functional: +``check-config``, ``list-servers`` (OPC DA server discovery) and the browse +verbs ``list-devices`` / ``list-tags`` (devices = branches, tags = leaves of +the connected server's address space); the remaining verbs land with Phases 1–3. +""" + +from __future__ import annotations + +import argparse +import sys +from collections.abc import Sequence + +from opcda_diagnozer import __version__ +from opcda_diagnozer.config import ConfigError, load_config +from opcda_diagnozer.opc.base import OpcError +from opcda_diagnozer.opc.discovery import ServerInfo, list_servers + +DEFAULT_CONFIG = "config.yaml" + +_NOT_IMPLEMENTED: dict[str, str] = { + "run": "Phase 1 (core engine)", + "probe": "Phase 1 (core engine)", + "list-items": "superseded by list-devices / list-tags", + "install": "Phase 3 (Windows service)", + "uninstall": "Phase 3 (Windows service)", + "start": "Phase 3 (Windows service)", + "stop": "Phase 3 (Windows service)", +} + + +def _add_config_option(parser: argparse.ArgumentParser) -> None: + parser.add_argument( + "--config", "-c", + default=DEFAULT_CONFIG, + help="path to the YAML config (default: %(default)s)", + ) + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="opcda-diagnozer", + description=( + "Read-only OPC DA reference client that diagnoses disconnections, " + "latency and data-quality problems (see docs/SPECIFICATION.md)." + ), + ) + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") + sub = parser.add_subparsers(dest="verb", metavar="verb", required=True) + + p = sub.add_parser("run", help="run in the foreground (console)") + _add_config_option(p) + + p = sub.add_parser("probe", help="connect once and print one read of every configured tag") + _add_config_option(p) + + p = sub.add_parser("check-config", help="validate the configuration and exit") + _add_config_option(p) + + p = sub.add_parser("list-items", help="browse the server item space (validation tool)") + _add_config_option(p) + + p = sub.add_parser( + "list-servers", + help="enumerate the OPC DA servers available locally or on a DCOM host", + ) + p.add_argument( + "--host", + metavar="HOST", + help="remote host to enumerate via DCOM (default: the local machine)", + ) + p.add_argument( + "--source", + choices=("auto", "automation", "registry"), + default="auto", + help="discovery mechanism: automation = OPC DA Automation Wrapper over COM; " + "registry = local COM-registration scan (no COM); auto = try automation, " + "fall back to registry (default: %(default)s)", + ) + p.add_argument( + "--json", + action="store_true", + help="emit one JSON object per server (for scripts) instead of a table", + ) + + p = sub.add_parser( + "list-devices", + help="list the devices (address-space branches) exposed by the configured server", + ) + _add_config_option(p) + p.add_argument( + "--depth", + type=int, + default=1, + metavar="N", + help="how many branch levels to descend (default: %(default)s)", + ) + p.add_argument( + "--no-counts", + action="store_true", + help="skip fetching per-device tag counts (faster on huge servers)", + ) + p.add_argument( + "--json", + action="store_true", + help="emit one JSON object per device (for scripts) instead of a table", + ) + + p = sub.add_parser( + "list-tags", + help="list the tags (leaves) of one device, or every tag on the server", + ) + _add_config_option(p) + p.add_argument( + "device", + nargs="*", + metavar="BRANCH", + help="device path: one or more branch names — either a single dotted string " + "in quotes ('\"Simulation Items.Bucket Brigade\"') or separate quoted " + "segments ('\"Simulation Items\" \"Bucket Brigade\"'). Default: the server root", + ) + p.add_argument( + "--recursive", "-r", + action="store_true", + help="walk the whole subtree, not just the device's direct tags", + ) + p.add_argument( + "--json", + action="store_true", + help="emit one JSON object per tag (for scripts) instead of a table", + ) + + p = sub.add_parser("install", help="install as a Windows service (requires admin)") + _add_config_option(p) + p.add_argument( + "--account", + metavar="DOMAIN\\USER", + help="dedicated service account (required for DCOM authentication, SPECIFICATION §11)", + ) + + for verb, help_text in ( + ("uninstall", "remove the Windows service"), + ("start", "start the Windows service"), + ("stop", "stop the Windows service"), + ): + sub.add_parser(verb, help=help_text) + + return parser + + +def _cmd_check_config(config_path: str) -> int: + try: + config = load_config(config_path) + except ConfigError as exc: + print(f"check-config: FAILED\n {exc}", file=sys.stderr) + return 2 + host = f" on host {config.opc.host!r}" if config.opc.host else " (local)" + expect_change = sum(1 for t in config.tags if t.expect_change) + print("check-config: OK") + print(f" server: {config.opc.server!r}{host}") + print(f" tags: {len(config.tags)} ({expect_change} with expect_change)") + print(f" poll: every {config.poll.interval_s} s, hang flagged after {config.poll.timeout_ms} ms") + return 0 + + +def _print_servers(servers: list[ServerInfo], host: str | None, as_json: bool) -> None: + where = f" on {host!r}" if host else " (local machine)" + if as_json: + import json + + for server in servers: + payload = { + "prog_id": server.prog_id, + "source": server.source, + "clsid": server.clsid, + "description": server.description, + "da_versions": list(server.da_versions), + } + print(json.dumps(payload, ensure_ascii=False)) + return + if not servers: + print(f"no OPC DA servers found{where}") + return + print(f"OPC DA servers available{where}:") + width = max(len(s.prog_id) for s in servers) + for server in servers: + notes = [f"via {server.source}"] + if server.da_versions: + notes.append(", ".join(server.da_versions)) + if server.description: + notes.append(server.description) + print(f" {server.prog_id:<{width}} ({'; '.join(notes)})") + + +def _cmd_list_servers(host: str | None, source: str, as_json: bool) -> int: + try: + servers = list_servers(host=host, source=source) # type: ignore[arg-type] + except OpcError as exc: + print(f"list-servers: FAILED\n {exc}", file=sys.stderr) + return 2 + _print_servers(servers, host, as_json) + if not as_json and not servers and host is None and source in ("auto", "registry"): + print( + "hint: no servers found — is an OPC DA server installed on this host? " + "(servers that skip the OPC component categories are invisible to " + "this scan; try --source automation)", + file=sys.stderr, + ) + return 0 + + +def _connect_from_config(config_path: str): + """Load the config and connect to its configured server (Automation Wrapper). + + Returns the connected client; raises :class:`ConfigError` / :class:`OpcError` + with actionable messages for the caller to print. + """ + from opcda_diagnozer.opc.automation import AutomationOpcClient + + config = load_config(config_path) + client = AutomationOpcClient( + server=config.opc.server, + host=config.opc.host, + connect_timeout_s=config.opc.connect_timeout_s, + ) + client.connect() + return client + + +def _print_devices(space, as_json: bool) -> None: + """Human table / JSON-lines output for the list-devices verb.""" + if as_json: + import json + + for device in space.devices: + payload = { + "name": device.name, + "path": list(device.path), + "qualified_name": device.qualified_name, + "sub_devices": list(device.sub_devices), + "tags": list(device.tag_names), + } + print(json.dumps(payload, ensure_ascii=False)) + return + if not space.devices: + print(f"no devices (branches) found — organization: {space.organization}") + return + print(f"address space organization: {space.organization}") + print("devices:") + width = max(len(d.qualified_name) for d in space.devices) + for device in space.devices: + notes = [f"{len(device.tag_names)} tags"] + if device.sub_devices: + notes.append(f"{len(device.sub_devices)} sub-devices") + print(f" {device.qualified_name:<{width}} ({', '.join(notes)})") + + +def _cmd_list_devices(config_path: str, depth: int, include_counts: bool, as_json: bool) -> int: + from opcda_diagnozer.opc.browse import list_devices + + try: + client = _connect_from_config(config_path) + browser = None + try: + browser = client.create_browser() + space = list_devices(browser, depth=depth, include_tag_counts=include_counts) + finally: + # Release the browser's COM reference *before* close() tears down the + # apartment — releasing an RCW after CoUninitialize crashes the process. + if browser is not None: + browser.close() + client.close() + except ConfigError as exc: + print(f"list-devices: FAILED\n {exc}", file=sys.stderr) + return 2 + except (OpcError, ValueError) as exc: + print(f"list-devices: FAILED\n {exc}", file=sys.stderr) + return 2 + _print_devices(space, as_json) + if not as_json and not space.devices: + hint = ( + "hint: the server address space is flat — every tag hangs directly off " + "the root; use: opcda-diagnozer list-tags" + if space.organization == "flat" + else "hint: no branches at the root — the server may expose only tags; " + "try: opcda-diagnozer list-tags" + ) + print(hint, file=sys.stderr) + return 0 + + +def _print_tags(tags, device_path: tuple[str, ...], recursive: bool, as_json: bool) -> None: + """Human table / JSON-lines output for the list-tags verb.""" + where = ".".join(device_path) or "(server root)" + scope = " (recursive)" if recursive else "" + if as_json: + import json + + for tag in tags: + payload = { + "name": tag.name, + "item_id": tag.item_id, + "path": list(tag.path), + } + print(json.dumps(payload, ensure_ascii=False)) + return + if not tags: + print(f"no tags found under {where}{scope}") + return + print(f"tags under {where}{scope}: {len(tags)}") + width = max(len(t.item_id) for t in tags) + for tag in tags: + print(f" {tag.item_id:<{width}} (leaf: {tag.name})") + + +def _cmd_list_tags( + config_path: str, + device: Sequence[str], + recursive: bool, + as_json: bool, +) -> int: + from opcda_diagnozer.opc.browse import list_tags, split_device_path + + # A single CLI argument is a *dotted* branch path; several arguments are + # explicit branch names (names may contain dots then). + device_arg = device[0] if len(device) == 1 else list(device) + try: + client = _connect_from_config(config_path) + browser = None + try: + browser = client.create_browser() + tags = list_tags(browser, device=device_arg or None, recursive=recursive) + finally: + if browser is not None: + browser.close() + client.close() + except ConfigError as exc: + print(f"list-tags: FAILED\n {exc}", file=sys.stderr) + return 2 + except (OpcError, ValueError) as exc: + print(f"list-tags: FAILED\n {exc}", file=sys.stderr) + return 2 + device_path = split_device_path(device_arg or None) + _print_tags(tags, device_path, recursive, as_json) + if not as_json and not tags and not recursive: + print( + "hint: the branch exists but exposes no direct leaves — try --recursive " + "to include the whole subtree", + file=sys.stderr, + ) + return 0 + + +def main(argv: Sequence[str] | None = None) -> int: + """CLI entry point (console script + ``python -m opcda_diagnozer``).""" + args = _build_parser().parse_args(argv) + if args.verb == "check-config": + return _cmd_check_config(args.config) + if args.verb == "list-servers": + return _cmd_list_servers(args.host, args.source, args.json) + if args.verb == "list-devices": + return _cmd_list_devices(args.config, args.depth, not args.no_counts, args.json) + if args.verb == "list-tags": + return _cmd_list_tags(args.config, args.device, args.recursive, args.json) + phase = _NOT_IMPLEMENTED[args.verb] + print( + f"{args.verb}: not implemented yet — planned for {phase} " + "(see docs/IMPLEMENTATION_PLAN.md)", + file=sys.stderr, + ) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/src/opcda_diagnozer/monitor/__init__.py b/src/opcda_diagnozer/monitor/__init__.py new file mode 100644 index 0000000..306adf3 --- /dev/null +++ b/src/opcda_diagnozer/monitor/__init__.py @@ -0,0 +1 @@ +"""Monitoring core: connection state machine, poller, detectors D1–D15.""" \ No newline at end of file diff --git a/src/opcda_diagnozer/monitor/detectors.py b/src/opcda_diagnozer/monitor/detectors.py new file mode 100644 index 0000000..b37da7b --- /dev/null +++ b/src/opcda_diagnozer/monitor/detectors.py @@ -0,0 +1,24 @@ +"""Detector rules D1–D15 (SPECIFICATION §6). + +Phase 1 (IMPLEMENTATION_PLAN §4.6): each detector becomes a small pure class +over :class:`PollOutcome` sequences with an injected clock; a rolling window +(deque, size ``window_polls``) powers p95/streak/flapping logic. Severity +mapping lives in SPECIFICATION §6. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime + +from opcda_diagnozer.opc.base import ReadResult + + +@dataclass(frozen=True) +class PollOutcome: + """One completed poll cycle, as fed to the detectors.""" + + started_at: datetime # local receive time when the read call started (wall clock) + rtt_ms: float # measured round-trip time of the read call + results: tuple[ReadResult, ...] = () # per-tag results; empty if the call itself failed + error: str | None = None # normalized OpcError message when the whole poll failed (D1/D2/D7) \ No newline at end of file diff --git a/src/opcda_diagnozer/monitor/poller.py b/src/opcda_diagnozer/monitor/poller.py new file mode 100644 index 0000000..f4118c9 --- /dev/null +++ b/src/opcda_diagnozer/monitor/poller.py @@ -0,0 +1,25 @@ +"""The poll loop on the single dedicated COM thread (SPECIFICATION §10.3). + +Phase 1 (IMPLEMENTATION_PLAN §4.7): connect → poll loop → feed detectors, plus +the hang watchdog (SPECIFICATION §10.4): a second thread timestamps poll +starts/ends; when a call exceeds ``hard_hang_s`` the stuck COM thread is +abandoned and a fresh connection is built on a new thread. +""" + +from __future__ import annotations + +from opcda_diagnozer.config import Config +from opcda_diagnozer.opc.base import OpcClient + + +class Poller: + """Drives one :class:`OpcClient` on its own thread — implemented in Phase 1.""" + + def __init__(self, client: OpcClient, config: Config) -> None: + self._client = client + self._config = config + + def run_forever(self) -> None: + raise NotImplementedError( + "Phase 1: poller + hang watchdog (docs/IMPLEMENTATION_PLAN.md §4.7)" + ) \ No newline at end of file diff --git a/src/opcda_diagnozer/monitor/state.py b/src/opcda_diagnozer/monitor/state.py new file mode 100644 index 0000000..702d1dc --- /dev/null +++ b/src/opcda_diagnozer/monitor/state.py @@ -0,0 +1,20 @@ +"""Connection state machine + reconnect backoff (SPECIFICATION §10.2). + +States: Connecting → Connected → ReconnectWait → Connecting … +Backoff is exponential 1 s → 2 s → … capped at 60 s; every disconnect/reconnect +pair is an event (D1/D2) with the outage duration computed and reported. + +The transition/backoff engine lands in Phase 1 (IMPLEMENTATION_PLAN §4.5). +""" + +from __future__ import annotations + +from enum import Enum + + +class ConnectionState(Enum): + """States of the connection state machine (SPECIFICATION §10.2).""" + + CONNECTING = "connecting" + CONNECTED = "connected" + RECONNECT_WAIT = "reconnect_wait" \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/__init__.py b/src/opcda_diagnozer/opc/__init__.py new file mode 100644 index 0000000..613d882 --- /dev/null +++ b/src/opcda_diagnozer/opc/__init__.py @@ -0,0 +1 @@ +"""OPC DA client backends: shared protocol, Automation Wrapper adapter, fake.""" \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/automation.py b/src/opcda_diagnozer/opc/automation.py new file mode 100644 index 0000000..29578ed --- /dev/null +++ b/src/opcda_diagnozer/opc/automation.py @@ -0,0 +1,222 @@ +"""OPC DA Automation Wrapper adapter (``OPCAutomation.dll``) — Phase 1. + +The Automation Wrapper is a ubiquitous **32-bit in-proc** COM DLL: the Python +process must be 32-bit to load it (SPECIFICATION §11 — the Phase 0 bitness +decision). All COM work happens on one dedicated STA thread owned by the +poller (SPECIFICATION §10.3); COM exceptions map to :class:`OpcError` with the +``win32com`` scode (IMPLEMENTATION_PLAN §4.4). + +pywin32 imports are deferred into the methods so this module stays importable +on any platform — tests run cross-OS with the fake client (plan §9). + +Verified against the live ``OPC.Automation`` on this machine (2026-10-07, +Matrikon.OPC.Simulation.1): ``Connect(progId)`` / ``Connect(progId, node)``, +``CreateBrowser()`` returns a dispatch object with ``ShowBranches`` / +``ShowLeafs`` (sic) / ``Count`` / ``Item(i)`` (1-based) / ``GetItemID`` / +``MoveDown`` / ``MoveUp`` / ``MoveToRoot`` / ``MoveTo(list)`` / +``CurrentPosition``; ``Organization`` 1 = hierarchical, 2 = flat. Bad branch +names raise ``E_INVALIDARG``; ``MoveTo`` requires a *sequence* (a bare string +is a pywin32 SAFEARRAY ``TypeError``). +""" + +from __future__ import annotations + +from collections.abc import Sequence +from typing import Any + +from opcda_diagnozer.opc.base import ( + Browser, + OpcError, + ReadResult, + TagInfo, + as_com_error, +) + +_NOT_IMPL = "Phase 1: Automation Wrapper adapter (docs/IMPLEMENTATION_PLAN.md §4.4)" + +#: OPC_BROWSE organization constants (Automation Wrapper): 1 = hierarchical, +#: 2 = flat (tags addressed by bare ItemID, no branches to walk). +_ORGANIZATION_LABELS = {1: "hierarchical", 2: "flat"} + + +class _AutomationBrowser: + """Thin, read-only :class:`Browser` over one wrapper browser object. + + COM failures are re-raised as :class:`OpcError` with the raw HRESULT. The + underlying dispatch object is dropped in :meth:`close` **before** + ``CoUninitialize`` — releasing an RCW after apartment teardown crashes the + process ("Win32 exception releasing IUnknown" / access violation). + """ + + def __init__(self, raw_browser: Any) -> None: + self._browser = raw_browser + + @property + def organization(self) -> str: + try: + raw = int(self._browser.Organization) + except Exception as exc: # noqa: BLE001 + raise as_com_error("could not read the browser Organization property", exc) from None + return _ORGANIZATION_LABELS.get(raw, f"unknown ({raw})") + + def _move_to(self, path: Sequence[str]) -> None: + """Reposition the browser at ``path`` (root when empty).""" + try: + if path: + # A bare string is a pywin32 marshalling TypeError — always a list. + self._browser.MoveTo(list(path)) + else: + self._browser.MoveToRoot() + except Exception as exc: # noqa: BLE001 + dotted = ".".join(path) + raise as_com_error( + f"could not move the browser to {dotted or 'the root'} " + "(bad branch name?)", + exc, + ) from None + + def list_branches(self, path: Sequence[str] = ()) -> list[str]: + self._move_to(path) + try: + self._browser.ShowBranches() + count = int(self._browser.Count) + return [str(self._browser.Item(i)).strip() for i in range(1, count + 1)] + except Exception as exc: # noqa: BLE001 + raise as_com_error( + f"could not enumerate branches under {'.'.join(path) or 'the root'}", + exc, + ) from None + + def list_leaves(self, path: Sequence[str] = ()) -> list[TagInfo]: + self._move_to(path) + try: + self._browser.ShowLeafs() # sic — the wrapper spells it "Leafs" + count = int(self._browser.Count) + leaves: list[TagInfo] = [] + for i in range(1, count + 1): + name = str(self._browser.Item(i)).strip() + if not name: + continue + item_id = str(self._browser.GetItemID(name)).strip() + if not item_id: + # Some servers return an empty ItemID for filtered leaves; + # fall back to the browse name so config tags stay usable. + item_id = name + leaves.append(TagInfo(name=name, item_id=item_id, path=tuple(path))) + return leaves + except Exception as exc: # noqa: BLE001 + raise as_com_error( + f"could not enumerate tags under {'.'.join(path) or 'the root'}", + exc, + ) from None + + def move_to(self, path: Sequence[str]) -> None: + self._move_to(path) + + def close(self) -> None: + self._browser = None + + +class AutomationOpcClient: + """``win32com``-based :class:`OpcClient`. + + ``connect`` / ``create_browser`` / ``close`` are live (they power the + ``list-devices`` / ``list-tags`` validation verbs); ``read`` lands with the + Phase 1 core engine (IMPLEMENTATION_PLAN §4.4). + """ + + def __init__(self, server: str, host: str | None = None, connect_timeout_s: float = 10.0) -> None: + self._server = server + self._host = host + self._connect_timeout_s = connect_timeout_s + self._com_initialized = False + self._opc: Any = None # the wrapper's dispatch object when connected + self._browsers: list[_AutomationBrowser] = [] # live browsers of this client + + @staticmethod + def _require_pywin32() -> None: + try: + import pythoncom # noqa: F401 (import guard, Windows + pywin32 only) + except ImportError as exc: + raise OpcError( + "the Automation Wrapper backend requires pywin32 (Windows only); " + "install it into the 32-bit venv with: pip install pywin32" + ) from exc + + def connect(self) -> None: + self._require_pywin32() + import pythoncom + + from win32com.client import Dispatch + + try: + pythoncom.CoInitialize() + self._com_initialized = True + except Exception as exc: # noqa: BLE001 + raise as_com_error("could not initialize COM on this thread", exc) from None + try: + try: + opc = Dispatch("OPC.Automation") + if self._host: + opc.Connect(self._server, self._host) + else: + opc.Connect(self._server) + self._opc = opc + except Exception as exc: # noqa: BLE001 + where = f"host {self._host!r}" if self._host else "the local machine" + raise as_com_error( + f"could not connect to OPC DA server {self._server!r} on {where}", + exc, + ) from None + except BaseException: + # The apartment must be torn down whenever the connection is not up. + self._release_and_uninit() + raise + + def create_browser(self) -> Browser: + if self._opc is None: + raise OpcError("AutomationOpcClient: create_browser() before connect()") + try: + raw = self._opc.CreateBrowser() + except Exception as exc: # noqa: BLE001 + raise as_com_error( + f"server {self._server!r} does not support browsing (CreateBrowser failed)", + exc, + ) from None + browser = _AutomationBrowser(raw) + self._browsers.append(browser) + return browser + + def _release_and_uninit(self) -> None: + """Drop COM references **inside** the apartment, then CoUninitialize. + + Live browsers created by this client are released first: releasing an + RCW after the apartment is torn down causes "Win32 exception releasing + IUnknown" / access violations at exit (observed: 0x80010108 fatal + exception under pytest). + """ + for browser in self._browsers: + browser.close() + self._browsers.clear() + if self._opc is not None: + try: + # Graceful server-side disconnect before dropping our references: + # the wrapper tears down its server connection (and the RPC + # channel) instead of the RPC runtime discovering the drop at + # CoUninitialize time. + self._opc.Disconnect() + except Exception: # noqa: BLE001 — best-effort teardown, never fatal + pass + self._opc = None + if self._com_initialized: + import pythoncom + + self._com_initialized = False + pythoncom.CoUninitialize() + + def read(self, tags: Sequence[str]) -> list[ReadResult]: + self._require_pywin32() + raise NotImplementedError(_NOT_IMPL) + + def close(self) -> None: + self._release_and_uninit() \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/base.py b/src/opcda_diagnozer/opc/base.py new file mode 100644 index 0000000..7af126a --- /dev/null +++ b/src/opcda_diagnozer/opc/base.py @@ -0,0 +1,164 @@ +"""OPC client abstraction (IMPLEMENTATION_PLAN §4.3). + +Backends implement :class:`OpcClient` (Automation Wrapper, fake). COM errors +are normalized to :class:`OpcError` carrying the raw HRESULT and a decoded name. +There is deliberately **no write path** — the tool is read-only by construction +(SPECIFICATION §10.5). + +Browsing (devices = branches, tags = leaves) is modelled by :class:`Browser`; +the ``list-devices`` / ``list-tags`` CLI verbs walk the address space through +it without any recursion into the monitored poll path. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from typing import Any, Protocol, runtime_checkable + + +@dataclass(frozen=True) +class ReadResult: + """One tag read: value, raw quality, device timestamp — or a per-tag error.""" + + tag: str + value: Any + quality_raw: int + device_time: datetime | None + error: str | None = None + + +# Common HRESULTs the tool decodes and reports (README "DCOM troubleshooting"). +_HRESULT_NAMES: dict[int, str] = { + 0x800706BA: "RPC_S_SERVER_UNAVAILABLE — server process down, firewall, or DCOM activation refused", + 0x800706BE: "RPC_S_CALL_FAILED — server crashed or connection reset mid-flight", + 0x80010108: "RPC_E_DISCONNECTED — OPC server dropped us (crash/restart)", + 0x80070005: "E_ACCESSDENIED — DCOM security misconfiguration", + 0x80040154: "REGDB_E_CLASSNOTREG — wrong ProgID, or wrapper/server registered under a different bitness", + 0x80020009: "DISP_E_EXCEPTION — the server rejected the call (e.g. no such branch/tag)", + 0x80070057: "E_INVALIDARG — invalid argument (e.g. wrong branch or item name)", + 0xC0040007: "OPC_E_INVALIDITEMID — tag name wrong in config (surfaced by probe)", +} + + +def decode_hresult(hresult: int) -> str: + """Explain a raw HRESULT in ops terms; unknown codes get a stable label.""" + return _HRESULT_NAMES.get(hresult & 0xFFFFFFFF, "unrecognized HRESULT") + + +def as_com_error(message: str, exc: BaseException) -> OpcError: + """Map a pywin32 error to :class:`OpcError`, keeping the raw HRESULT when present. + + The source exception's traceback is **cleared** before it is discarded: a + propagating exception pins its frames — including locals holding raw COM + dispatch pointers (``CDispatch``) — until a later garbage collection. When + that GC runs after ``CoUninitialize`` the RCW release hits a dead apartment + and Windows prints "Win32 exception occurred releasing IUnknown" (observed + with the Automation Wrapper). Clearing the traceback keeps the HRESULT + (already extracted) while letting the RCWs die inside the apartment. + """ + hresult = getattr(exc, "hresult", None) + if hresult is None and getattr(exc, "args", None): + first = exc.args[0] + if isinstance(first, int): + hresult = first + exc.__traceback__ = None + return OpcError(message, hresult=hresult) + + +class OpcError(RuntimeError): + """Normalized COM/OPC error; carries the raw HRESULT when available.""" + + def __init__(self, message: str, hresult: int | None = None) -> None: + super().__init__(message) + self.hresult = hresult + + def __str__(self) -> str: + base = super().__str__() + if self.hresult is None: + return base + code = self.hresult & 0xFFFFFFFF + return f"{base} (HRESULT 0x{code:08X}: {decode_hresult(code)})" + + +#: Access rights mask bits (OPCEnum / Automation Wrapper OPC_ACCESS-rights). +#: The wrapper's ``AccessRights`` property combines READ (1) and WRITE (2); +#: this tool is read-only by construction and never sets a WRITE-only view. +ACCESS_READ = 0x01 +ACCESS_WRITE = 0x02 + + +@dataclass(frozen=True) +class DeviceInfo: + """One browsable device: a branch in the server's hierarchical address space. + + The Automation Wrapper exposes no per-branch metadata beyond the name, so a + "device" here is simply the branch path plus the counts the CLI prints. + """ + + name: str + #: Branch path from the server root, e.g. ``("Simulation Items", "Bucket Brigade")``. + path: tuple[str, ...] + #: Sub-branches directly under this device (usually channels/groups). + sub_devices: tuple[str, ...] = () + #: Leaf (tag) names directly under this device. + tag_names: tuple[str, ...] = () + + @property + def qualified_name(self) -> str: + """Human-friendly dotted rendering: ``Simulation Items.Bucket Brigade``.""" + return ".".join(self.path) if self.path else self.name + + +@dataclass(frozen=True) +class TagInfo: + """One browsable tag (leaf): its ItemID is what goes into config ``tags:``.""" + + name: str + #: Fully qualified ItemID as accepted by read calls, e.g. ``Bucket Brigade.Int1``. + item_id: str + #: Branch path of the device the tag lives under; ``()`` for flat servers. + path: tuple[str, ...] = () + + +@runtime_checkable +class Browser(Protocol): + """Read-only traversal of a connected server's address space. + + Mirrors the Automation Wrapper's ``IOPCBrowse`` surface (SPECIFICATION §8): + branches are devices/sub-devices, leaves are tags. Implementations walk on + demand; the browse verbs fetch one level at a time so a huge server never + stalls the CLI on an unbounded recursion. + """ + + @property + def organization(self) -> str: + """``"hierarchical"`` or ``"flat"`` (OPC_BROWSE constants).""" + ... + + def list_branches(self, path: Sequence[str] = ()) -> list[str]: + """Sub-branch names directly under ``path`` (empty path = server root).""" + ... + + def list_leaves(self, path: Sequence[str] = ()) -> list[TagInfo]: + """Tags directly under ``path``, with their fully qualified ItemIDs.""" + ... + + def move_to(self, path: Sequence[str]) -> None: + """Reposition at ``path`` (root when empty); raises OpcError when it does not exist.""" + ... + + def close(self) -> None: + """Release the underlying COM/browser resources (inside the apartment).""" + ... + + +@runtime_checkable +class OpcClient(Protocol): + """A read-only OPC DA client: connect → browse/read → close. No write path.""" + + def connect(self) -> None: ... + def create_browser(self) -> Browser: ... + def read(self, tags: Sequence[str]) -> list[ReadResult]: ... + def close(self) -> None: ... \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/browse.py b/src/opcda_diagnozer/opc/browse.py new file mode 100644 index 0000000..7e2e00b --- /dev/null +++ b/src/opcda_diagnozer/opc/browse.py @@ -0,0 +1,194 @@ +"""Browse the connected server's address space: devices (branches) and tags (leaves). + +OPC DA has no formal "device" concept — servers publish a *hierarchical +address space* whose top-level branches almost always represent devices (or +device families), with tags as the leaves. This module maps that tree onto the +two questions ops asks (SPECIFICATION §8, validation tooling): + +- *Which devices does this server expose?* → :func:`list_devices` +- *Which tags does device X expose?* → :func:`list_tags` + +The Automation Wrapper browser (``opc.CreateBrowser()``) is the only portable +traversal API — it works over DCOM for remote servers too, with the same +Launch+Activation requirements as ``list-servers``. Traversal is one level at +a time; the ``max_depth`` guard keeps a pathological (cyclic/AB-tree) server +from stalling the CLI. + +Flat-namespace servers report ``organization == "flat"``; there every leaf is +addressed by its bare ItemID, the root is treated as one implicit device. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass + +from opcda_diagnozer.opc.base import Browser, DeviceInfo, OpcError, TagInfo + +#: Walk guard: deep SCADA hierarchies exist (Channel → Device → Tag), but OPC +#: trees deeper than this are almost certainly a server-side loop. +MAX_DEPTH = 32 + + +@dataclass(frozen=True) +class Space: + """Result of a device-space walk: one entry per discovered branch node.""" + + organization: str + root: DeviceInfo + devices: tuple[DeviceInfo, ...] + + +def split_device_path(path: Sequence[str] | str | None) -> tuple[str, ...]: + """Normalize a user-supplied device path into branch segments. + + A **single** string is treated as dotted: ``"Simulation Items.Bucket Brigade"`` + → ``("Simulation Items", "Bucket Brigade")``. **Multiple** segments are + taken as explicit branch names — no dot splitting, so names that contain + a dot stay intact (they just have to be passed as separate arguments). + Empty segments and whitespace are dropped. + """ + if path is None: + return () + if isinstance(path, str): + segments: list[str] = [p.strip() for p in path.split(".")] + return tuple(p for p in segments if p) + parts: list[str] = [] + for segment in path: + piece = str(segment).strip() + if piece: + parts.append(piece) + return tuple(parts) + + +def list_devices( + browser: Browser, + *, + depth: int = 1, + include_tag_counts: bool = True, +) -> Space: + """Enumerate devices (branches) starting at the server root. + + ``depth=1`` lists top-level branches (the typical "device list"); greater + depths reveal sub-devices (channels/groups). Each :class:`DeviceInfo` + carries the direct sub-branches and — when ``include_tag_counts`` is on — + the leaf names under it, so the CLI can print ``Device (3 tags, 2 sub)``. + + Raises :class:`OpcError` when the server rejects the walk (COM failure or + unsupported browse interface). The walk is bounded by ``depth`` (≤ + :data:`MAX_DEPTH`), so a pathological server can never stall the CLI. + """ + if depth < 1: + raise ValueError("depth must be >= 1") + if depth > MAX_DEPTH: + raise ValueError(f"depth must be <= {MAX_DEPTH}") + organization = browser.organization + root_branches = browser.list_branches(()) + root_leaves = browser.list_leaves(()) if include_tag_counts else [] + root = DeviceInfo( + name="(root)", + path=(), + sub_devices=tuple(root_branches), + tag_names=tuple(t.name for t in root_leaves), + ) + devices: list[DeviceInfo] = [] + _walk_level(browser, (), root_branches, depth, devices, include_tag_counts) + return Space(organization=organization, root=root, devices=tuple(devices)) + + +def _walk_level( + browser: Browser, + path: tuple[str, ...], + branches: Sequence[str], + depth: int, + devices: list[DeviceInfo], + include_tag_counts: bool, +) -> None: + if depth <= 0: + return + for branch in branches: + sub_path = path + (branch,) + sub_branches = tuple(browser.list_branches(sub_path)) + tag_names = ( + tuple(t.name for t in browser.list_leaves(sub_path)) + if include_tag_counts + else () + ) + devices.append( + DeviceInfo( + name=branch, + path=sub_path, + sub_devices=sub_branches, + tag_names=tag_names, + ) + ) + _walk_level(browser, sub_path, sub_branches, depth - 1, devices, include_tag_counts) + + +def list_tags( + browser: Browser, + device: str | Sequence[str] | None = None, + *, + recursive: bool = False, +) -> list[TagInfo]: + """List the tags of one device (branch), or of the whole server. + + ``device`` accepts either a dotted string (``"Simulation Items.Bucket + Brigade"``) or a pre-split sequence of branch names. ``None`` means the + server root; with ``recursive=True`` every leaf under it is returned + (bounded by :data:`MAX_DEPTH`). The returned :class:`TagInfo.item_id` + values are exactly what goes into the config ``tags:`` list. An unknown + device path raises :class:`OpcError` pointing at ``list-devices``. + """ + path = _require_device(browser, device, what="list-tags") + leaves = browser.list_leaves(path) + tags: list[TagInfo] = list(leaves) + if not recursive: + return sorted(tags, key=lambda t: t.item_id.lower()) + sub_branches = browser.list_branches(path) + _collect_recursive(browser, path, sub_branches, tags) + return sorted(tags, key=lambda t: t.item_id.lower()) + + +def _collect_recursive( + browser: Browser, + path: tuple[str, ...], + branches: Sequence[str], + tags: list[TagInfo], +) -> None: + for branch in branches: + sub_path = path + (branch,) + if len(sub_path) > MAX_DEPTH: + raise OpcError( + f"address space deeper than {MAX_DEPTH} levels at " + f"{'.'.join(sub_path)} — refusing to walk further (possible " + "server-side cycle)" + ) + tags.extend(browser.list_leaves(sub_path)) + _collect_recursive(browser, sub_path, browser.list_branches(sub_path), tags) + + +def _require_device( + browser: Browser, + device: str | Sequence[str] | None, + *, + what: str, +) -> tuple[str, ...]: + """Resolve and validate a user-supplied device path (exists, is a branch).""" + path = split_device_path(device) + if not path: + return () + siblings = browser.list_branches(path[:-1]) + if path[-1] not in siblings: + raise OpcError( + f"{what}: device {path[-1]!r} not found under " + f"{'.'.join(path[:-1]) or '(root)'} — run list-devices to see the " + "available branches" + ) + return path + + +def tags_for_device(browser: Browser, device: str | Sequence[str]) -> list[TagInfo]: + """List tags of a **known** device path — validation helper (probe).""" + path = _require_device(browser, device, what="list-tags") + return sorted(browser.list_leaves(path), key=lambda t: t.item_id.lower()) \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/discovery.py b/src/opcda_diagnozer/opc/discovery.py new file mode 100644 index 0000000..08622b0 --- /dev/null +++ b/src/opcda_diagnozer/opc/discovery.py @@ -0,0 +1,224 @@ +"""Enumerate the OPC DA servers available to this process (local, or via DCOM). + +Two mechanisms, matching how OPC DA discovery actually works on Windows: + +- ``automation`` — the OPC DA Automation Wrapper (``OPC.Automation``, i.e. + ``OPCAutomation.dll``). ``GetOPCServers(node)`` returns the ProgIDs the host + publishes through the COM component categories; with a node name it + enumerates a remote host over DCOM (requires Launch+Activation permissions, + SPECIFICATION §11). Needs COM initialized; a 32-bit interpreter sees the + 32-bit registrations — the same servers this tool can load. +- ``registry`` — a direct COM-registration scan of + ``HKCR\\CLSID\\{...}\\Implemented Categories`` for the OPC DA component + categories. Local only, needs no COM at all — the fallback for locked-down + hosts or boxes where the wrapper DLL is not registered. + +OPC Foundation DA component categories (verified): + DA 1.0 ``{63D5F430-CFE4-11D1-B2C8-0060083BA1FB}`` + DA 2.0 ``{63D5F432-CFE4-11D1-B2C8-0060083BA1FB}`` + DA 3.0 ``{CC603E42-48D7-4E57-A010-7563B3B0A715}`` +""" + +from __future__ import annotations + +import sys +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from typing import Any, Literal + +from opcda_diagnozer.opc.base import OpcError, as_com_error + +AUTOMATION_PROG_ID = "OPC.Automation" # OPC DA Automation Wrapper (OPCAutomation.dll) + +#: Component-category CATID -> human label (OPC Foundation). +DA_CATEGORIES: dict[str, str] = { + "{63D5F430-CFE4-11D1-B2C8-0060083BA1FB}": "DA 1.0", + "{63D5F432-CFE4-11D1-B2C8-0060083BA1FB}": "DA 2.0", + "{CC603E42-48D7-4E57-A010-7563B3B0A715}": "DA 3.0", +} + +Source = Literal["auto", "automation", "registry"] + + +@dataclass(frozen=True) +class ServerInfo: + """One discoverable OPC DA server.""" + + prog_id: str + source: Literal["automation", "registry"] + clsid: str | None = None + description: str | None = None + da_versions: tuple[str, ...] = () + + +def normalize_string_array(raw: Any) -> list[str]: + """Coerce a COM VARIANT result (tuple of str / str / None) to a clean str list.""" + if raw is None: + items: tuple[Any, ...] = () + elif isinstance(raw, str): + items = (raw,) + elif isinstance(raw, (list, tuple)): + items = tuple(raw) + elif hasattr(raw, "__iter__"): + items = tuple(raw) + else: + items = () + cleaned = [item.strip() for item in items if isinstance(item, str) and item.strip()] + seen: set[str] = set() + deduped: list[str] = [] + for item in cleaned: + if item not in seen: + seen.add(item) + deduped.append(item) + return deduped + + +@contextmanager +def _com_apartment() -> Iterator[None]: + """Initialize COM for the duration of the enumeration (SPECIFICATION §10.3). + + Enumeration runs on the CLI thread, not the future poller's dedicated COM + thread, so it initializes/uninitializes its own apartment here. + """ + import pythoncom + + pythoncom.CoInitialize() + try: + yield + finally: + pythoncom.CoUninitialize() + + +def _via_automation(host: str | None) -> list[ServerInfo]: + """Enumerate via the Automation Wrapper — local machine, or remote via DCOM.""" + try: + import pywintypes + from win32com.client import Dispatch + except ImportError as exc: + raise OpcError( + "the Automation Wrapper source requires pywin32 (Windows only); " + "install it into the 32-bit venv with: pip install pywin32" + ) from exc + where = f"host {host!r}" if host else "the local machine" + with _com_apartment(): + try: + opc = Dispatch(AUTOMATION_PROG_ID) + try: + raw = opc.GetOPCServers(host) if host else opc.GetOPCServers() + # Marshal everything to plain Python data and release the COM + # reference *inside* the apartment: releasing an IUnknown after + # CoUninitialize crashes (observed: "Win32 exception releasing + # IUnknown" / access violation at interpreter shutdown). + prog_ids = normalize_string_array(raw) + finally: + opc = None # drop the COM reference while COM is still initialized + except pywintypes.com_error as exc: + raise as_com_error( + f"could not enumerate OPC DA servers on {where} via the " + "Automation Wrapper (is OPCAutomation.dll registered?)", + exc, + ) from None + except AttributeError as exc: + raise OpcError( + "the registered Automation Wrapper does not expose " + f"GetOPCServers ({exc}); try --source registry" + ) from None + servers = [ServerInfo(prog_id=prog, source="automation") for prog in prog_ids] + return sorted(servers, key=lambda s: s.prog_id.lower()) + + +def _registry_value(root: Any, subkey: str) -> str | None: + """Default string value of ``root\\subkey``; None when absent/not a string.""" + import winreg + + try: + with winreg.OpenKey(root, subkey) as key: + value, _ = winreg.QueryValueEx(key, "") + except OSError: + return None + return value if isinstance(value, str) else None + + +def _implemented_da_categories(root: Any, clsid: str) -> tuple[str, ...]: + """DA versions a CLSID implements, from its Implemented Categories key.""" + import winreg + + found: list[str] = [] + try: + with winreg.OpenKey(root, f"{clsid}\\Implemented Categories") as cats: + index = 0 + while True: + try: + category = winreg.EnumKey(cats, index) + except OSError: + break + index += 1 + label = DA_CATEGORIES.get(category) + if label is not None: + found.append(label) + except OSError: + pass # no Implemented Categories subkey -> not an OPC DA server + return tuple(sorted(found)) + + +def _via_registry(host: str | None) -> list[ServerInfo]: + """Scan COM registrations for DA component categories — local, no COM needed. + + A 32-bit process sees the 32-bit registry view, which is exactly the set + of servers loadable by this (32-bit) tool (SPECIFICATION §11). + """ + if host is not None: + raise OpcError( + "the registry source only enumerates the local machine; " + "remote enumeration requires the Automation wrapper over DCOM " + "(use --source auto or --source automation with --host)" + ) + if sys.platform != "win32": + raise OpcError("the registry source requires Windows") + import winreg + + servers: list[ServerInfo] = [] + with winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, "CLSID") as root: + index = 0 + while True: + try: + clsid = winreg.EnumKey(root, index) + except OSError: + break + index += 1 + if not (clsid.startswith("{") and clsid.endswith("}")): + continue + da_versions = _implemented_da_categories(root, clsid) + if not da_versions: + continue + prog_id = _registry_value(root, f"{clsid}\\ProgID") + servers.append( + ServerInfo( + prog_id=prog_id or clsid, + source="registry", + clsid=clsid, + description=_registry_value(root, clsid), + da_versions=da_versions, + ) + ) + return sorted(servers, key=lambda s: s.prog_id.lower()) + + +def list_servers(host: str | None = None, source: Source = "auto") -> list[ServerInfo]: + """Enumerate available OPC DA servers (see module docstring for mechanisms). + + ``auto`` tries the Automation Wrapper first and falls back to the registry + scan when COM enumeration fails on the local machine; with a remote + ``host`` only the Automation Wrapper can be used, so its error propagates. + """ + if source not in ("auto", "automation", "registry"): + raise ValueError(f"unknown source {source!r}") + if source in ("auto", "automation"): + try: + return _via_automation(host) + except OpcError: + if source == "automation" or host is not None: + raise + # local auto: fall back to the COM-registration scan + return _via_registry(host) \ No newline at end of file diff --git a/src/opcda_diagnozer/opc/fake.py b/src/opcda_diagnozer/opc/fake.py new file mode 100644 index 0000000..7461640 --- /dev/null +++ b/src/opcda_diagnozer/opc/fake.py @@ -0,0 +1,130 @@ +"""Scriptable in-process fake OPC client — tests and demo without COM/hardware. + +Phase 1 (IMPLEMENTATION_PLAN §4.9) turns this into the scenario engine +(scripted sequences of good reads, errors, hangs, quality flips) that powers +the entire unit-test suite. This placeholder always reads GOOD-quality zeros +so the skeleton's smoke tests have something deterministic to assert on. + +The fake carries a small **hierarchical address space** (Matrikon-like: +``Simulation Items → Bucket Brigade → Int1…``) so the browse verbs +(``list-devices`` / ``list-tags``) and their tests run cross-OS without COM. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from datetime import datetime, timezone + +from opcda_diagnozer.opc.base import Browser, OpcError, ReadResult, TagInfo + +#: The fake address space: branch path -> (sub-branches, leaf names). +#: Mirrors the shape of a real SCADA server (Channel → Device → Tags). +FAKE_SPACE: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]] = { + (): (("Simulation Items", "Configured Aliases"), ("#MonitorACLFile", "@ClientCount", "@Clients")), + ("Simulation Items",): ( + ( + "Bucket Brigade", + "Random", + "Read Error", + "Saw-toothed Waves", + "Square Waves", + "Triangle Waves", + "Write Error", + "Write Only", + ), + (), + ), + ("Simulation Items", "Bucket Brigade"): ( + (), + ("Int1", "Int2", "Int4", "Real4", "Real8", "String", "Boolean"), + ), + ("Simulation Items", "Random"): ((), ("Int1", "Int4", "Money", "Time")), + ("Simulation Items", "Read Error"): ((), ("Int1", "Int4")), + ("Simulation Items", "Saw-toothed Waves"): ((), ("Int1", "Real8")), + ("Simulation Items", "Square Waves"): ((), ("Int1", "Real8")), + ("Simulation Items", "Triangle Waves"): ((), ("Int1", "Real8")), + ("Simulation Items", "Write Error"): ((), ("Int1", "Int4")), + ("Simulation Items", "Write Only"): ((), ("Int1", "Int4")), + ("Configured Aliases",): ((), ("Alias1", "Alias2")), +} + + +def _item_id(path: tuple[str, ...], leaf: str) -> str: + """Fully qualified ItemID, mirroring the verified Matrikon sim convention: + + the immediate branch + leaf (``Bucket Brigade.Int1`` even though the branch + is nested under ``Simulation Items``); bare leaf at the root. This is what + the real server's ``GetItemID`` returns for the same tree. + """ + return f"{path[-1]}.{leaf}" if path else leaf + + +class _FakeBrowser: + """In-process :class:`Browser` over a fake space — pure data, no COM.""" + + def __init__(self, space: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]]) -> None: + self._space = space + + @property + def organization(self) -> str: + return "hierarchical" if self._space.get(()) and self._space[()][0] else "flat" + + def list_branches(self, path: Sequence[str] = ()) -> list[str]: + key = tuple(path) + entry = self._space.get(key) + if entry is None: + raise OpcError(f"no such branch: {'.'.join(key) or '(root)'}") + return list(entry[0]) + + def list_leaves(self, path: Sequence[str] = ()) -> list[TagInfo]: + key = tuple(path) + entry = self._space.get(key) + if entry is None: + raise OpcError(f"no such branch: {'.'.join(key) or '(root)'}") + return [ + TagInfo(name=leaf, item_id=_item_id(key, leaf), path=key) + for leaf in entry[1] + ] + + def move_to(self, path: Sequence[str]) -> None: + key = tuple(path) + if key and key not in self._space: + raise OpcError(f"no such branch: {'.'.join(key)}") + + def close(self) -> None: + self._space = {} + + +class FakeOpcClient: + """In-process :class:`OpcClient`; every read returns GOOD-quality zeros. + + ``space`` is injectable so tests can exercise flat namespaces, empty + servers, and deep trees; the default mirrors a Matrikon-like hierarchy. + """ + + def __init__( + self, + space: dict[tuple[str, ...], tuple[tuple[str, ...], tuple[str, ...]]] | None = None, + ) -> None: + self.connected = False + self._space = FAKE_SPACE if space is None else space + + def connect(self) -> None: + self.connected = True + + def create_browser(self) -> Browser: + if not self.connected: + raise OpcError("FakeOpcClient: create_browser() before connect()") + return _FakeBrowser(self._space) + + def read(self, tags: Sequence[str]) -> list[ReadResult]: + if not self.connected: + raise OpcError("FakeOpcClient: read() before connect()") + now = datetime.now(timezone.utc) + return [ + ReadResult(tag=tag, value=0, quality_raw=0xC0, device_time=now) + for tag in tags + ] + + def close(self) -> None: + self.connected = False \ No newline at end of file diff --git a/src/opcda_diagnozer/quality.py b/src/opcda_diagnozer/quality.py new file mode 100644 index 0000000..793daf9 --- /dev/null +++ b/src/opcda_diagnozer/quality.py @@ -0,0 +1,69 @@ +"""OPC DA quality decode tables (SPECIFICATION §7). + +Quality is a 16-bit field: bits 7–6 of the low byte give the class +(0xC0 GOOD / 0x40 UNCERTAIN / 0x00 BAD); bits 5–0 are the substatus. Every +quality event must carry the raw hex code **and** the decoded name. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class QualityClass(str, Enum): + GOOD = "GOOD" + UNCERTAIN = "UNCERTAIN" + BAD = "BAD" + + +@dataclass(frozen=True) +class QualityInfo: + """Decoded quality for one read (SPECIFICATION §7).""" + + raw: int + name: str + quality_class: QualityClass + + @property + def raw_hex(self) -> str: + """Raw code formatted for events, e.g. ``0x18``.""" + return f"0x{self.raw & 0xFF:02X}" + + +# Codes from SPECIFICATION §7. +# TODO(Phase 1): complete the remaining DA 2.05a UNCERTAIN subcodes in this +# table — they decode by the same scheme (spec §7 closing note). +_QUALITY_NAMES: dict[int, str] = { + 0xC0: "GOOD", + 0xD8: "GOOD_LOCAL_OVERRIDE", + 0x40: "UNCERTAIN", + 0x44: "UNCERTAIN_LAST_USABLE", + 0x00: "BAD", + 0x04: "BAD_CONFIG_ERROR", + 0x08: "BAD_NOT_CONNECTED", + 0x0C: "BAD_DEVICE_FAILURE", + 0x10: "BAD_SENSOR_FAILURE", + 0x14: "BAD_LAST_KNOWN_VALUE", + 0x18: "BAD_COMM_FAILURE", + 0x1C: "BAD_OUT_OF_SERVICE", +} + + +def _class_of(low: int) -> QualityClass: + """Class comes from bits 7–6 of the low byte.""" + bits = low & 0xC0 + if bits == 0xC0: + return QualityClass.GOOD + if bits == 0x40: + return QualityClass.UNCERTAIN + return QualityClass.BAD + + +def decode_quality(raw: int) -> QualityInfo: + """Decode a raw OPC DA quality word into a named :class:`QualityInfo`.""" + low = raw & 0xFF + name = _QUALITY_NAMES.get(low) + if name is None: + name = f"{_class_of(low).value}_0x{low:02X}" + return QualityInfo(raw=raw, name=name, quality_class=_class_of(low)) \ No newline at end of file diff --git a/src/opcda_diagnozer/report/__init__.py b/src/opcda_diagnozer/report/__init__.py new file mode 100644 index 0000000..9be1efe --- /dev/null +++ b/src/opcda_diagnozer/report/__init__.py @@ -0,0 +1 @@ +"""Event reporters: JSONL file, Windows Event Log, optional HTTP webhook.""" \ No newline at end of file diff --git a/src/opcda_diagnozer/report/base.py b/src/opcda_diagnozer/report/base.py new file mode 100644 index 0000000..fa4534c --- /dev/null +++ b/src/opcda_diagnozer/report/base.py @@ -0,0 +1,13 @@ +"""Reporter protocol — pluggable event sinks (SPECIFICATION §9).""" + +from __future__ import annotations + +from typing import Protocol + +from opcda_diagnozer.events import DiagEvent + + +class Reporter(Protocol): + """Anything that can receive events; reporting never blocks the poll loop.""" + + def emit(self, event: DiagEvent) -> None: ... \ No newline at end of file diff --git a/src/opcda_diagnozer/report/eventlog.py b/src/opcda_diagnozer/report/eventlog.py new file mode 100644 index 0000000..311e7fc --- /dev/null +++ b/src/opcda_diagnozer/report/eventlog.py @@ -0,0 +1,22 @@ +"""Windows Event Log reporter, WARN+ only (IMPLEMENTATION_PLAN §5.2). + +Uses ``win32evtlog``; the source is registered during ``install``. pywin32 is +import-guarded so tests can run on any OS (IMPLEMENTATION_PLAN §9). +""" + +from __future__ import annotations + +from opcda_diagnozer.events import DiagEvent, Severity + + +class EventLogReporter: + """Windows Event Log sink — implemented in Phase 2.""" + + def __init__(self, source: str = "OPCDA-Diagnozer", min_severity: str = "WARN") -> None: + self._source = source + self._min_severity = Severity(min_severity) + + def emit(self, event: DiagEvent) -> None: + raise NotImplementedError( + "Phase 2: Event Log reporter (docs/IMPLEMENTATION_PLAN.md §5.2)" + ) \ No newline at end of file diff --git a/src/opcda_diagnozer/report/http.py b/src/opcda_diagnozer/report/http.py new file mode 100644 index 0000000..acd11d2 --- /dev/null +++ b/src/opcda_diagnozer/report/http.py @@ -0,0 +1,23 @@ +"""Optional HTTP webhook reporter (IMPLEMENTATION_PLAN §5.3). + +Phase 2: POST one JSON event per request, optional bearer token, timeout plus +simple retry/drop on failure — must never block the poll loop (bounded queue). +""" + +from __future__ import annotations + +from opcda_diagnozer.events import DiagEvent + + +class HttpReporter: + """Webhook sink — implemented in Phase 2.""" + + def __init__(self, url: str, bearer_token: str | None = None, timeout_s: float = 5.0) -> None: + self._url = url + self._bearer_token = bearer_token + self._timeout_s = timeout_s + + def emit(self, event: DiagEvent) -> None: + raise NotImplementedError( + "Phase 2: HTTP webhook reporter (docs/IMPLEMENTATION_PLAN.md §5.3)" + ) \ No newline at end of file diff --git a/src/opcda_diagnozer/report/jsonl.py b/src/opcda_diagnozer/report/jsonl.py new file mode 100644 index 0000000..9a93b5b --- /dev/null +++ b/src/opcda_diagnozer/report/jsonl.py @@ -0,0 +1,25 @@ +"""Rotating JSONL reporter — the primary evidence trail (IMPLEMENTATION_PLAN §5.1). + +Phase 2 adds: size-based rotation with ``backups`` kept, atomic-ish append, +and a header line carrying the config digest for provenance. +""" + +from __future__ import annotations + +from pathlib import Path + +from opcda_diagnozer.events import DiagEvent + + +class JsonlReporter: + """One JSON event per line, rotated by size — implemented in Phase 2.""" + + def __init__(self, directory: str | Path, max_mb: int = 50, backups: int = 10) -> None: + self._directory = Path(directory) + self._max_mb = max_mb + self._backups = backups + + def emit(self, event: DiagEvent) -> None: + raise NotImplementedError( + "Phase 2: JSONL reporter (docs/IMPLEMENTATION_PLAN.md §5.1)" + ) \ No newline at end of file diff --git a/src/opcda_diagnozer/service.py b/src/opcda_diagnozer/service.py new file mode 100644 index 0000000..7344f93 --- /dev/null +++ b/src/opcda_diagnozer/service.py @@ -0,0 +1,19 @@ +"""Windows service host via pywin32 ``ServiceFramework`` — Phase 3. + +IMPLEMENTATION_PLAN §6.1: the service spawns the COM worker thread and must +stop cleanly (``STOP_PENDING``, join with timeout, abandon a hung COM thread +on shutdown). SCM recovery (restart on failure) is configured at install time. +""" + +from __future__ import annotations + +from collections.abc import Sequence + + +class OpcdaDiagnozerService: + """pywin32 service wrapper — implemented in Phase 3.""" + + def __init__(self, args: Sequence[str]) -> None: + raise NotImplementedError( + "Phase 3: Windows service host (docs/IMPLEMENTATION_PLAN.md §6)" + ) \ No newline at end of file diff --git a/test.py b/test.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_browse.py b/tests/test_browse.py new file mode 100644 index 0000000..6d154d9 --- /dev/null +++ b/tests/test_browse.py @@ -0,0 +1,310 @@ +"""Browse tests (opc/browse.py + fake browser + CLI verbs). + +Pure tests run on any OS against the fake client's address space; the COM +path is Windows-only and covered by the live smoke tests (IMPLEMENTATION_PLAN +§9: COM adapters are import-guarded). +""" + +import sys + +import pytest + +from opcda_diagnozer.main import main as cli_main +from opcda_diagnozer.opc.base import Browser, DeviceInfo, OpcError, TagInfo +from opcda_diagnozer.opc.browse import ( + MAX_DEPTH, + list_devices, + list_tags, + split_device_path, + tags_for_device, +) +from opcda_diagnozer.opc.fake import FAKE_SPACE, FakeOpcClient + + +@pytest.fixture +def browser() -> Browser: + client = FakeOpcClient() + client.connect() + return client.create_browser() + + +# --------------------------------------------------------------------------- # +# split_device_path +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (None, ()), + ((), ()), + ("", ()), + (" ", ()), + ("Simulation Items.Bucket Brigade", ("Simulation Items", "Bucket Brigade")), + ("Simulation Items", ("Simulation Items",)), + ("a..b", ("a", "b")), # empty segments dropped + (("Simulation Items", "Bucket Brigade"), ("Simulation Items", "Bucket Brigade")), + (("Simulation Items", " Bucket Brigade "), ("Simulation Items", "Bucket Brigade")), + (("Simulation Items", "", "Bucket Brigade"), ("Simulation Items", "Bucket Brigade")), + # explicit segments keep embedded dots intact (no split) + (("Bucket. Inc", "Device 1"), ("Bucket. Inc", "Device 1")), + ], +) +def test_split_device_path(raw, expected: tuple[str, ...]) -> None: + assert split_device_path(raw) == expected + + +# --------------------------------------------------------------------------- # +# list_devices +# --------------------------------------------------------------------------- # + + +def test_list_devices_top_level(browser: Browser) -> None: + space = list_devices(browser) + assert space.organization == "hierarchical" + assert space.root.sub_devices == ("Simulation Items", "Configured Aliases") + names = [d.name for d in space.devices] + assert names == ["Simulation Items", "Configured Aliases"] + assert all(d.path == (d.name,) for d in space.devices) + + +def test_list_devices_depth_two_includes_sub_devices(browser: Browser) -> None: + space = list_devices(browser, depth=2) + paths = [d.path for d in space.devices] + assert ("Simulation Items", "Bucket Brigade") in paths + assert ("Simulation Items", "Random") in paths + # depth 2: the root's direct children plus their direct children + assert len(paths) == 2 + 8 # 2 top-level + 8 under Simulation Items + + +def test_list_devices_reports_tag_counts(browser: Browser) -> None: + space = list_devices(browser) + by_name = {d.name: d for d in space.devices} + assert by_name["Simulation Items"].tag_names == () # no direct leaves + assert by_name["Configured Aliases"].tag_names == ("Alias1", "Alias2") + assert by_name["Simulation Items"].sub_devices[:4] == ( + "Bucket Brigade", "Random", "Read Error", "Saw-toothed Waves", + ) + + +def test_list_devices_no_counts_skips_leaves(browser: Browser) -> None: + space = list_devices(browser, include_tag_counts=False) + assert space.root.tag_names == () + assert all(d.tag_names == () for d in space.devices) + + +@pytest.mark.parametrize("depth", [0, -1, MAX_DEPTH + 1]) +def test_list_devices_rejects_bad_depth(browser: Browser, depth: int) -> None: + with pytest.raises(ValueError): + list_devices(browser, depth=depth) + + +def test_list_devices_cycle_guard(monkeypatch: pytest.MonkeyPatch) -> None: + """A pathological self-nesting branch must not hang the recursive walk.""" + + class _LoopBrowser: + @property + def organization(self) -> str: + return "hierarchical" + + def list_branches(self, path=()): + return ["Loop"] # "Loop" always contains another "Loop" — cycle + + def list_leaves(self, path=()): + return [] + + def move_to(self, path): + return None + + def close(self) -> None: + return None + + with pytest.raises(OpcError, match="deeper than"): + list_tags(_LoopBrowser(), recursive=True) # type: ignore[arg-type] + + +# --------------------------------------------------------------------------- # +# list_tags +# --------------------------------------------------------------------------- # + + +def test_list_tags_root_direct(browser: Browser) -> None: + tags = list_tags(browser) + assert [t.name for t in tags] == ["#MonitorACLFile", "@ClientCount", "@Clients"] + assert all(t.item_id == t.name for t in tags) # root leaves: bare ItemIDs + assert all(t.path == () for t in tags) + + +def test_list_tags_of_device_dotted_string(browser: Browser) -> None: + tags = list_tags(browser, device="Simulation Items.Bucket Brigade") + item_ids = [t.item_id for t in tags] + assert "Bucket Brigade.Int1" in item_ids + assert "Bucket Brigade.Real8" in item_ids + assert len(tags) == 7 + + +def test_list_tags_of_device_segments(browser: Browser) -> None: + tags = list_tags(browser, device=["Simulation Items", "Random"]) + assert {t.name for t in tags} == {"Int1", "Int4", "Money", "Time"} + assert all(t.path == ("Simulation Items", "Random") for t in tags) + + +def test_list_tags_recursive(browser: Browser) -> None: + tags = list_tags(browser, device="Simulation Items", recursive=True) + item_ids = {t.item_id for t in tags} + # direct children of "Simulation Items" have no leaves; recursion gathers all + assert "Bucket Brigade.Int1" in item_ids + assert "Random.Money" in item_ids + assert "Read Error.Int4" in item_ids + assert "Saw-toothed Waves.Real8" in item_ids + assert all(t.path for t in tags) + + + +def test_tags_are_sorted_case_insensitively(browser: Browser) -> None: + tags = list_tags(browser, device="Simulation Items.Bucket Brigade") + assert [t.item_id for t in tags] == sorted( + (t.item_id for t in tags), key=lambda s: s.lower() + ) + + +# --------------------------------------------------------------------------- # +# fake browser & client protocol conformance +# --------------------------------------------------------------------------- # + + +def test_fake_browser_matches_protocol(browser: Browser) -> None: + # runtime_checkable protocol: structural check + assert isinstance(browser, Browser) + + +def test_fake_client_browser_requires_connect() -> None: + client = FakeOpcClient() + with pytest.raises(OpcError, match="before connect"): + client.create_browser() + + +def test_fake_browser_unknown_branch_raises() -> None: + client = FakeOpcClient() + client.connect() + browser = client.create_browser() + with pytest.raises(OpcError, match="no such branch"): + browser.list_branches(["Simulation Items", "Missing"]) + with pytest.raises(OpcError, match="no such branch"): + browser.list_leaves(["Missing"]) + + +def test_fake_flat_space_reports_flat_organization() -> None: + client = FakeOpcClient(space={(): ((), ("Tag1", "Tag2"))}) + client.connect() + browser = client.create_browser() + assert browser.organization == "flat" + space = list_devices(browser) + assert space.devices == () + assert space.root.tag_names == ("Tag1", "Tag2") + + +# --------------------------------------------------------------------------- # +# CLI verbs (against the fake backend through a config file) +# --------------------------------------------------------------------------- # + + +def _run_cli(monkeypatch: pytest.MonkeyPatch, tmp_path, args: list[str]) -> int: + """Run the CLI with the fake client injected in place of the COM adapter.""" + import opcda_diagnozer.main as main_mod + from opcda_diagnozer.opc.fake import FakeOpcClient + + config = tmp_path / "config.yaml" + config.write_text( + "opc:\n server: Fake.Server.1\ntags:\n - name: Bucket Brigade.Int1\n", + encoding="utf-8", + ) + client = FakeOpcClient() + + def _fake_connect(config_path: str): + client.connect() + return client + + monkeypatch.setattr(main_mod, "_connect_from_config", _fake_connect) + return cli_main(args) + + +def test_cli_list_devices_table(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None: + code = _run_cli(monkeypatch, tmp_path, ["list-devices", "-c", str(tmp_path / "config.yaml")]) + out = capsys.readouterr().out + assert code == 0 + assert "Simulation Items" in out + assert "Configured Aliases" in out + assert "2 tags" in out + + +def test_cli_list_devices_json(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None: + code = _run_cli(monkeypatch, tmp_path, [ + "list-devices", "-c", str(tmp_path / "config.yaml"), "--json", + ]) + out = capsys.readouterr().out + assert code == 0 + import json + + lines = [json.loads(line) for line in out.splitlines() if line.strip()] + assert {d["name"] for d in lines} == {"Simulation Items", "Configured Aliases"} + assert lines[0]["qualified_name"] == "Simulation Items" # walk order = root order + + +def test_cli_list_tags_dotted(monkeypatch: pytest.MonkeyPatch, tmp_path, capsys) -> None: + code = _run_cli(monkeypatch, tmp_path, [ + "list-tags", "-c", str(tmp_path / "config.yaml"), + "Simulation Items.Bucket Brigade", + ]) + out = capsys.readouterr().out + assert code == 0 + assert "Bucket Brigade.Int1" in out + assert "Bucket Brigade.Real8" in out + + +def test_cli_list_tags_segments_and_recursive( + monkeypatch: pytest.MonkeyPatch, tmp_path, capsys +) -> None: + code = _run_cli(monkeypatch, tmp_path, [ + "list-tags", "-c", str(tmp_path / "config.yaml"), + "Simulation Items", "--recursive", "--json", + ]) + out = capsys.readouterr().out + assert code == 0 + import json + + item_ids = {json.loads(line)["item_id"] for line in out.splitlines() if line.strip()} + assert "Bucket Brigade.Int1" in item_ids + assert "Random.Money" in item_ids + + +def test_cli_list_tags_missing_device_fails_cleanly( + monkeypatch: pytest.MonkeyPatch, tmp_path, capsys +) -> None: + code = _run_cli(monkeypatch, tmp_path, [ + "list-tags", "-c", str(tmp_path / "config.yaml"), "No.Such.Device", + ]) + captured = capsys.readouterr() + assert code == 2 + assert "FAILED" in captured.err + + +@pytest.mark.skipif(sys.platform != "win32", reason="COM browse needs Windows + pywin32") +def test_live_automation_browse_smoke() -> None: + """Live smoke test — requires the Matrikon simulation server on this dev box.""" + pytest.importorskip("pythoncom") + from opcda_diagnozer.opc.automation import AutomationOpcClient + + client = AutomationOpcClient(server="Matrikon.OPC.Simulation.1") + try: + client.connect() + browser = client.create_browser() + devices = list_devices(browser) + assert devices.organization == "hierarchical" + assert any(d.name == "Simulation Items" for d in devices.devices) + tags = list_tags(browser, device="Simulation Items.Bucket Brigade") + assert any(t.item_id == "Bucket Brigade.Int1" for t in tags) + except OpcError as exc: + pytest.skip(f"simulation server not usable on this machine: {exc}") + finally: + client.close() \ No newline at end of file diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..1e7dc8b --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,85 @@ +"""Config loading + validation tests (IMPLEMENTATION_PLAN §4.1).""" + +from pathlib import Path + +import pytest + +from opcda_diagnozer.config import ConfigError, load_config + +EXAMPLE = Path(__file__).resolve().parents[1] / "config.example.yaml" + + +def _write(tmp_path: Path, text: str) -> Path: + p = tmp_path / "config.yaml" + p.write_text(text, encoding="utf-8") + return p + + +def test_example_config_loads() -> None: + config = load_config(EXAMPLE) + assert config.opc.server == "Matrikon.OPC.Simulation.1" + assert config.opc.host == "CONTROL-HOST01" + assert len(config.tags) == 2 + assert config.tags[0].name == "Channel1.Device1.Counter" + assert config.tags[0].expect_change is True + assert config.tags[1].expect_change is False + assert config.poll.interval_s == 1.0 + assert config.poll.timeout_ms == 5000 + assert config.latency.warn_ms == 1000 + assert config.report.eventlog.min_severity == "WARN" + assert config.report.jsonl.dir == "C:/ProgramData/OPCDA-Diagnozer/logs" + + +def test_missing_file_is_actionable() -> None: + with pytest.raises(ConfigError, match="not found"): + load_config("no-such-config.yaml") + + +def test_missing_opc_section_is_actionable(tmp_path: Path) -> None: + p = _write(tmp_path, "tags:\n - name: t\n") + with pytest.raises(ConfigError, match=r"opc: missing required key 'server'"): + load_config(p) + + +def test_unknown_key_rejected(tmp_path: Path) -> None: + p = _write(tmp_path, "opc:\n server: X\n typo_key: 1\ntags:\n - name: t\n") + with pytest.raises(ConfigError, match="typo_key"): + load_config(p) + + +def test_unknown_top_level_section_rejected(tmp_path: Path) -> None: + p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: t\nnonsense: {}\n") + with pytest.raises(ConfigError, match="nonsense"): + load_config(p) + + +def test_empty_tags_rejected(tmp_path: Path) -> None: + p = _write(tmp_path, "opc:\n server: X\ntags: []\n") + with pytest.raises(ConfigError, match="at least one tag"): + load_config(p) + + +def test_duplicate_tags_rejected(tmp_path: Path) -> None: + p = _write(tmp_path, "opc:\n server: X\ntags:\n - name: a\n - name: a\n") + with pytest.raises(ConfigError, match="duplicate"): + load_config(p) + + +def test_latency_threshold_order_rejected(tmp_path: Path) -> None: + p = _write( + tmp_path, + "opc:\n server: X\ntags:\n - name: t\n" + "latency:\n warn_ms: 5000\n critical_ms: 1000\n", + ) + with pytest.raises(ConfigError, match="warn_ms"): + load_config(p) + + +def test_hard_hang_must_exceed_timeout(tmp_path: Path) -> None: + p = _write( + tmp_path, + "opc:\n server: X\ntags:\n - name: t\n" + "poll:\n timeout_ms: 5000\n hard_hang_s: 4\n", + ) + with pytest.raises(ConfigError, match="hard_hang_s"): + load_config(p) \ No newline at end of file diff --git a/tests/test_detectors.py b/tests/test_detectors.py new file mode 100644 index 0000000..0df070f --- /dev/null +++ b/tests/test_detectors.py @@ -0,0 +1,20 @@ +"""Detector tests (SPECIFICATION §6). Detector implementations land in Phase 1.""" + +from datetime import datetime, timezone + +from opcda_diagnozer.monitor.detectors import PollOutcome +from opcda_diagnozer.opc.base import ReadResult + + +def test_poll_outcome_holds_one_cycle() -> None: + result = ReadResult( + tag="Channel1.Device1.Level", value=1.5, quality_raw=0xC0, device_time=None + ) + outcome = PollOutcome( + started_at=datetime.now(timezone.utc), + rtt_ms=12.5, + results=(result,), + ) + assert outcome.rtt_ms == 12.5 + assert outcome.error is None + assert outcome.results[0].tag == "Channel1.Device1.Level" \ No newline at end of file diff --git a/tests/test_discovery.py b/tests/test_discovery.py new file mode 100644 index 0000000..6d9a59a --- /dev/null +++ b/tests/test_discovery.py @@ -0,0 +1,66 @@ +"""Discovery tests (opc/discovery.py). + +Pure tests run on any OS; the registry/COM scans are Windows-only and skip +elsewhere (IMPLEMENTATION_PLAN §9: COM adapters are import-guarded). +""" + +import sys + +import pytest + +from opcda_diagnozer.opc.base import OpcError +from opcda_diagnozer.opc.discovery import ( + DA_CATEGORIES, + ServerInfo, + list_servers, + normalize_string_array, +) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + (None, []), + ("", []), + ("One.Server.1", ["One.Server.1"]), + (("A.Sim.1", "A.Sim.2"), ["A.Sim.1", "A.Sim.2"]), + (("A.Sim.1", " A.Sim.1 ", "A.Sim.2"), ["A.Sim.1", "A.Sim.2"]), # dedupe + trim + ((1, None, "A.Sim.1"), ["A.Sim.1"]), # non-strings dropped + ([], []), + ], +) +def test_normalize_string_array(raw: object, expected: list[str]) -> None: + assert normalize_string_array(raw) == expected + + +def test_registry_source_rejects_remote_host() -> None: + with pytest.raises(OpcError, match="local"): + list_servers(host="REMOTE-HOST", source="registry") + + +def test_unknown_source_rejected() -> None: + with pytest.raises(ValueError, match="unknown source"): + list_servers(source="nonsense") # type: ignore[arg-type] + + +@pytest.mark.skipif(sys.platform != "win32", reason="winreg scan is Windows-only") +def test_registry_scan_local_structural() -> None: + servers = list_servers(source="registry") + assert isinstance(servers, list) + for server in servers: + assert server.source == "registry" + assert server.prog_id # non-empty ProgID (or CLSID fallback) + assert server.clsid is not None + assert server.clsid.startswith("{") and server.clsid.endswith("}") + assert set(server.da_versions) <= set(DA_CATEGORIES.values()) + assert [s.prog_id.lower() for s in servers] == sorted(s.prog_id.lower() for s in servers) + + +@pytest.mark.skipif(sys.platform != "win32", reason="COM enumeration needs Windows + pywin32") +def test_automation_enumeration_smoke() -> None: + """Live smoke test — requires a registered Automation Wrapper (dev box).""" + try: + servers = list_servers(source="automation") + except OpcError as exc: + pytest.skip(f"Automation Wrapper not usable on this machine: {exc}") + assert all(s.source == "automation" and s.prog_id for s in servers) \ No newline at end of file diff --git a/tests/test_poller_fake.py b/tests/test_poller_fake.py new file mode 100644 index 0000000..e2f2a2c --- /dev/null +++ b/tests/test_poller_fake.py @@ -0,0 +1,33 @@ +"""End-to-end pipeline tests against opc.fake (IMPLEMENTATION_PLAN §9). + +The fake client lets the event pipeline run without COM/hardware; Phase 1 +grows this into the full scenario suite. +""" + +import pytest + +from opcda_diagnozer.opc.base import OpcError +from opcda_diagnozer.opc.fake import FakeOpcClient + + +def test_fake_reads_good_quality() -> None: + client = FakeOpcClient() + client.connect() + results = client.read(["Channel1.Device1.Counter", "Channel1.Device1.Level"]) + assert [r.tag for r in results] == ["Channel1.Device1.Counter", "Channel1.Device1.Level"] + assert all(r.quality_raw == 0xC0 for r in results) + assert all(r.device_time is not None for r in results) + + +def test_fake_requires_connect_first() -> None: + client = FakeOpcClient() + with pytest.raises(OpcError): + client.read(["Channel1.Device1.Counter"]) + + +def test_fake_close_disconnects() -> None: + client = FakeOpcClient() + client.connect() + client.close() + with pytest.raises(OpcError): + client.read(["Channel1.Device1.Counter"]) \ No newline at end of file diff --git a/tests/test_quality.py b/tests/test_quality.py new file mode 100644 index 0000000..173fb25 --- /dev/null +++ b/tests/test_quality.py @@ -0,0 +1,41 @@ +"""Quality decode tests (SPECIFICATION §7).""" + +import pytest + +from opcda_diagnozer.quality import QualityClass, decode_quality + + +@pytest.mark.parametrize( + ("raw", "name", "quality_class"), + [ + (0xC0, "GOOD", QualityClass.GOOD), + (0xD8, "GOOD_LOCAL_OVERRIDE", QualityClass.GOOD), + (0x44, "UNCERTAIN_LAST_USABLE", QualityClass.UNCERTAIN), + (0x40, "UNCERTAIN", QualityClass.UNCERTAIN), + (0x00, "BAD", QualityClass.BAD), + (0x04, "BAD_CONFIG_ERROR", QualityClass.BAD), + (0x08, "BAD_NOT_CONNECTED", QualityClass.BAD), + (0x0C, "BAD_DEVICE_FAILURE", QualityClass.BAD), + (0x10, "BAD_SENSOR_FAILURE", QualityClass.BAD), + (0x14, "BAD_LAST_KNOWN_VALUE", QualityClass.BAD), + (0x18, "BAD_COMM_FAILURE", QualityClass.BAD), + (0x1C, "BAD_OUT_OF_SERVICE", QualityClass.BAD), + ], +) +def test_spec_table_decodes(raw: int, name: str, quality_class: QualityClass) -> None: + info = decode_quality(raw) + assert info.name == name + assert info.quality_class is quality_class + assert info.raw_hex == f"0x{raw:02X}" + + +def test_unknown_code_falls_back_to_class() -> None: + info = decode_quality(0x20) + assert info.quality_class is QualityClass.BAD + assert info.name.startswith("BAD_") + + +def test_class_bits_come_from_low_byte() -> None: + # 16-bit quality word: the high byte must not affect the class (§7) + assert decode_quality(0x0118).quality_class is QualityClass.BAD + assert decode_quality(0xFFC0).quality_class is QualityClass.GOOD \ No newline at end of file diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 0000000..f3556bd --- /dev/null +++ b/tests/test_state.py @@ -0,0 +1,11 @@ +"""State machine tests (SPECIFICATION §10.2). Transitions land in Phase 1.""" + +from opcda_diagnozer.monitor.state import ConnectionState + + +def test_connection_states_match_spec() -> None: + assert {s.name for s in ConnectionState} == { + "CONNECTING", + "CONNECTED", + "RECONNECT_WAIT", + } \ No newline at end of file