diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d7e5898..f0ad4cd 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -188,6 +188,8 @@ Blanton_TTL_Script/ │ # eth0→TARGET_10G、eth1→TARGET_1G，兩張卡同時持續 ping ├── port_prbs_monitor.sh # 100G uplink PRBS 測試（bcmcmd phy diag） ├── blanton_ber.sh # 全機 SerDes BER 掃描（phydiag prbs / prbsstat） + ├── nfc_polling.sh # NFC tag 背景輪詢 + PASS/FAIL 統計 + ├── blantons_nfc_validate.py # NFC/NCI 驗證工具（廠商提供，未修改） ├── blanton_tr518.sh # TR518 封包測試，量功耗前的負載來源 ├── TH6_SWB0_power_readback.sh # SWB0 19 條軌 V×I + 總功耗 ├── TH6_SWB1_power_readback.sh # SWB1 同上 @@ -342,9 +344,9 @@ PROFILE_CHANNELS=( "0:high:8" "1:low:8" ... ) # "channel:operation:change_perc | 模組 | 進入點 | 功能 | |------|--------|------| -| **Script A（Pre-test）** | `1_Blanton_Script_A.ttl` V1.1.1 | root 登入 → `date -s` 對時 → 清 job log → **收掉前一輪殘留**（`hw-test-session finish` → `bgctl reset --yes` → `bgctl stop --all` → `bgctl list` 確認清空）→ `hw-test-session start` → **設定風扇轉速** → **等資料面就緒**（`wait_init.ttl`）→ DUT 清單（boot/version/fwutil/syseeprom/ssdhealth/TPM/NVMe）→ BMC enroll + version/status → source bash 工具 → 清/讀 7 組 PCIe AER + `ras-mc-ctl --summary` → `lspci` → margin → PMON → dmesg → **10G/1G 管理網路 ping** → 100G uplink 設定與狀態 → **一輪流量基線** → `show uptime` | -| **Script B（Stress）** | `2_Blanton_Script_B.ttl` | `mlucas-avx2 -cpu 0:15` + `bgctl run` 的 `memtester` / SSD / USB 壓力 + BMC DDR（`bmc-manager run memtester`）→ BMC USB net test（探測 cdc_ncm 介面，`systemd-run` 掛 4 小時 ping）→ 啟動 mgmt ping 監控 → `jobs` / `bgctl list` → traffic（依 `SWB_UNIT`）→ `while 1` 每輪完整 PMON + margin | -| **Script C（Post-test）** | `3_Blanton_Script_C.ttl` | `kill $(jobs -p)` + `bgctl stop --all` / `reset` + 停 BMC USB unit 與 mgmt ping → PMON + dmesg → 7 組 AER + `ras-mc-ctl` → BMC USB `journalctl` → `ip -s link show eth0/eth1` → `cat` mlucas log → traffic `stop` + `report` → **BER 掃描**（依 `SWB_UNIT`）→ NVMe 健康 → `show reboot-cause` / `uptime` → **job log 複製到 USB（帶時戳）** → `hw-test-session finish` | +| **Script A（Pre-test）** | `1_Blanton_Script_A.ttl` V1.1.2 | root 登入 → `date -s` 對時 → 清 job log → **收掉前一輪殘留**（`hw-test-session finish` → `bgctl reset --yes` → `bgctl stop --all` → `bgctl list` 確認清空）→ `hw-test-session start` → **設定風扇轉速** → **等資料面就緒**（`wait_init.ttl`）→ DUT 清單（boot/version/fwutil/syseeprom/ssdhealth/TPM/NVMe）→ BMC enroll + version/status → source bash 工具 → 清/讀 7 組 PCIe AER + `ras-mc-ctl --summary` → `lspci` → margin → PMON → dmesg → **10G/1G 管理網路 ping** → **NFC 基線**（start → 30s → stop → report）→ 100G uplink 設定與狀態 → **一輪流量基線** → `show uptime` | +| **Script B（Stress）** | `2_Blanton_Script_B.ttl` | `mlucas-avx2 -cpu 0:15` + `bgctl run` 的 `memtester` / SSD / USB 壓力 + BMC DDR（`bmc-manager run memtester`）→ BMC USB net test（探測 cdc_ncm 介面，`systemd-run` 掛 4 小時 ping）→ 啟動 mgmt ping 監控 → **啟動 NFC 輪詢** → `jobs` / `bgctl list` → traffic（依 `SWB_UNIT`）→ `while 1` 每輪完整 PMON + margin | +| **Script C（Post-test）** | `3_Blanton_Script_C.ttl` | `kill $(jobs -p)` + `bgctl stop --all` / `reset` + 停 BMC USB unit、mgmt ping 與 **NFC 輪詢** → PMON + dmesg → 7 組 AER + `ras-mc-ctl` → BMC USB `journalctl` → `ip -s link show eth0/eth1` → `cat` mlucas log → **NFC `report`** → traffic `stop` + `report` → **BER 掃描**（依 `SWB_UNIT`）→ NVMe 健康 → `show reboot-cause` / `uptime` → **job log 複製到 USB（帶時戳）** → `hw-test-session finish` | | **SWB loopback 流量** | `blanton_traffic_linespeed [-u 0\|1\|all]` | `init`（VLAN 30..137，成對 `cdN`/`cdN+32`）→ `clear` → `start`（`tx 100 length=512`）→ `stop` → `report`（per-pair TX/RX 交叉比對 PASS/FAIL）；另有 `ps`（link/speed）與 `run`（一條龍）。**V0.6.0 起含 100G uplink `ce0`**（VLAN 138）：它沒有 loopback 對象，靠 **ingress mirror 打回自己**維持循環，`report` 最後一列就是它，底下再接一份跨 unit summary | | **FPGA 暫存器存取** | `cb_fpga [data]`、`fpga_rescan` | pcimem 打 `resource0` + BAR 相對 offset，32-bit word；BDF 於 source 時自動偵測；`pmc/icb/swb0/swb1_fpga` 走 VSPI 窗口 | | **CB I2C / PMBus** | `cb_i2c_init/scan/read/write`、`cb_pmbus_read/write/linear16` | FPGA F3 OpenCores master，**讀用 Repeated START**（SMBus/PMBus 裝置必需） | @@ -360,6 +362,7 @@ PROFILE_CHANNELS=( "0:high:8" "1:low:8" ... ) # "channel:operation:change_perc | **SerDes BER 掃描** | `blanton_ber.sh {init\|clear\|start\|stop\|report} [-u 0\|1\|all]` | 全機每個對接 port 的 PRBS BER：`prbs set p=3` + `prbs get`（確認每條 lane 都鎖上）→ `prbsstat STArt Interval=30` →（等一個 interval）→ `prbsstat Ber` → `STOp` → `prbs clear`。`report` 一列一條 lane，**BER < 1e-6 才 PASS**，最後跨 unit 的 summary 帶最差 lane。⚠️ PRBS 期間 link 顯示 DOWN、不過流量 | | **TR518 負載** | `blanton_tr518.sh {start\|stop} [-u 0\|1\|all]` | `port cd lb=mac` → `l2 learn off` → `test mode nr=yes` → `tr 518 ... testphase=1`，然後 settle `TR_SETTLE_SEC`（120s）再讓人量功耗。`stop` 只送 `port cd lb=none`（照 `TR518.txt`），L2 學習與 test mode **不會還原**，`TR_RESTORE_*=1` 才會 | | **SWB 軌功耗** | `TH6_SWB{0,1}_power_readback.sh` | 讀 `show platform voltage/current` **各一次**，19 條軌 V×I 與總功耗。舊版每軌各讀一次（38 次呼叫、每軌不同時刻），總和是不同瞬間的加總 | +| **NFC 讀取輪詢** | `nfc_polling.sh {start\|stop\|status\|tail\|report\|clear\|fg}` | 背景重複跑 `blantons_nfc_validate.py`，每輪追加一行機器可讀的 `RESULT=`（PASS/FAIL 取自工具自己的摘要行，兩者皆無則 **ERROR**，與 FAIL 分開計數）。`start` 會先清空 log；`report` **只讀 log，polling 中隨時可跑**，並標示 `RUNNING` 免得快照被當最終結果。`tail` **不 follow** | | **USB 目標偵測** | `usb_target.sh [-o dev\|mnt\|both\|id] [-n] [-r sec]` | 找出插入的 USB 儲存裝置，排除 `/` 與 `/host` 的底層碟（本平台可能從 USB DOM 開機），多顆時拒絕猜；掛載時會驗證真的可寫 | | **打包交付** | `publish/publish.sh [-v Ver] [-f] [-z] [-n]` | 版號讀 Script A 檔頭，複製成 `publish/Script_ABC_Blanton_/`，排除 `*.log`，可另出 zip | diff --git a/CLAUDE.md b/CLAUDE.md index aa4547f..8cb1b11 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -9,7 +9,7 @@ DUT 端的取數能力由一組 **bash 工具庫** 提供（CB FPGA `pcimem` → 完整架構見 [ARCHITECTURE.md](ARCHITECTURE.md)。 -**目前狀態（2026-08-28）**：Script A 為 **V1.1.1**（A 集中記錄 A/B/C 三支的變更）。涵蓋 +**目前狀態（2026-08-28）**：Script A 為 **V1.1.2**（A 集中記錄 A/B/C 三支的變更）。涵蓋 PCIe AER / DDR EDAC / PMON+TPM / 電壓 margin（9 顆 LTC2980）/ SWB loopback 線速流量 / 10G+1G 管理網路 ping / BMC DDR 與 USB / NVMe 健康，壓力層走平台的 `bgctl`。 switch unit 數量由 `config.ttl` 的 `SWB_UNIT0` / `SWB_UNIT1` 宣告，風扇轉速由 `FAN_SPEED` 指定 @@ -22,7 +22,8 @@ TR518 負載（`blanton_tr518.sh`，**TTL 尚未接**）、SWB 軌功耗（`TH6_ 線速流量也納入 100G `ce0`（VLAN 138，靠 ingress mirror 維持循環）。 Script A 開頭會先收掉前一輪殘留（`hw-test-session finish` + `bgctl reset/stop --all`）， 再 `bgctl list` 確認沒有 job 留著佔資源。 -最後一次發布是 **V1.1.0**；SWB 的 I2C 通道與位址已於 2026-08-19 上機驗證。 +NFC tag 讀取由 `nfc_polling.sh` 背景輪詢（A 做基線、B 起、C 停並 report）。 +最後一次發布是 **V1.1.1**；SWB 的 I2C 通道與位址已於 2026-08-19 上機驗證。 交付用 `./publish/publish.sh` 打包。 ## 技術棧 @@ -129,6 +130,16 @@ sleep 30 # 至少等一個 interval 再 rep ./Blanton_Script/blanton_tr518.sh stop # 只送 port cd lb=none # ⚠️ stop 不還原 l2 learn / test mode（TR_RESTORE_LEARN=1 / _TESTMODE=1 才會） +# ── NFC tag 輪詢 ──────────────────────────────────── +~/Blanton_Script/nfc_polling.sh start # 背景，start 會先清空 log +~/Blanton_Script/nfc_polling.sh status # pid、已跑幾輪、最後一筆結果 +~/Blanton_Script/nfc_polling.sh report # PASS/FAIL/ERROR 統計，polling 中也能跑 +~/Blanton_Script/nfc_polling.sh tail 50 # 最後 50 行（不 follow，不會卡住） +~/Blanton_Script/nfc_polling.sh stop +# log: ~/Blanton_Script/log/nfc_poll.log +# ERROR = 工具沒印出 [PASS]/[FAIL]（或逾時），跟「tag 讀不到」分開算 +# NFC_INTERVAL_SEC / NFC_TIMEOUT_SEC / NFC_ARGS 可調（要每個子指令都帶） + # ── USB 目標偵測 ──────────────────────────────────── bash ~/Blanton_Script/usb_target.sh -o dev # /dev/sdX1 bash ~/Blanton_Script/usb_target.sh -o mnt # 掛好並回傳掛載點 diff --git a/docs/release-notes/v1.1.2.md b/docs/release-notes/v1.1.2.md new file mode 100644 index 0000000..4405d5d --- /dev/null +++ b/docs/release-notes/v1.1.2.md @@ -0,0 +1,63 @@ +# v1.1.2 — NFC tag reads are now watched for the whole soak, not sampled once + +## ✨ New features + +**Background NFC polling** +* New `nfc_polling.sh` runs `blantons_nfc_validate.py` over and over in the background and keeps score. `start` / `stop` / `status` / `tail` / `report` / `clear` / `fg`. +* **`report` works while the poll is still running.** It only reads the log, so it never disturbs the run — and it labels itself `RUNNING` so a mid-soak snapshot is not mistaken for the final answer: + + ``` + state : RUNNING (pid=2240323) -- this is a snapshot, the count is still moving + rounds : 240 + PASS : 238 + FAIL : 2 + ERROR : 0 (tool did not produce a verdict) + RESULT : FAIL + + --- tags seen --- + 046F7BD2142290 238 + ``` +* **`ERROR` is counted apart from `FAIL`.** A round where the tool printed neither `[PASS]` nor `[FAIL]` — it crashed, the file was missing, it hung past the timeout — did not measure anything. That is a different problem from a tag that would not read, and rolling the two together hides whichever one you are not looking for. +* `report` also lists which NFCID1s were seen and how often, and the last ten failures with their timestamps and reasons. +* `start` clears the log first, so a report always describes the run you just did rather than everything since the DUT was built. +* Script A takes a 30-second NFC baseline, Script B starts the poll for the soak, Script C stops it with the other monitors and reports. + +## 🐛 Bug fixes + +**The NFC log never reached the USB stick** +* Script C copied `nfc_polling.log`; the file is `nfc_poll.log`. The `cp` failed, nothing checked it, and the run finished looking fine — you would only notice back at the desk with a job archive that had no NFC log in it. + +## 📦 Downloads + +| File | Contents | +|---|---| +| `Script_ABC_Blanton_V1.1.2.zip` | The full Tera Term working directory, including `Blanton_Script/` to copy onto the DUT | + +**After copying to the DUT:** + +```bash +chmod +x ~/Blanton_Script/*.sh # see below +grep -rlU $'\r' ~/Blanton_Script # expect no output +``` + +`blanton_ber.sh` and `nfc_polling.sh` are invoked by the macros through `./` or `~/`, so they need the execute bit along with `bmc_monitor.sh` and `usb_target.sh` — neither git nor a Windows/USB copy carries it reliably. `blantons_nfc_validate.py` does **not** need it: `nfc_polling.sh` calls it through `python3` precisely so one more `chmod` cannot be forgotten. + +## ⚠️ Before you run + +* **`nfc_polling.sh start` clears the log.** Script B starts a fresh poll for the soak, which wipes Script A's baseline log — the baseline numbers are already in the master console log by then, but the file is gone. +* **NFC log path is `~/Blanton_Script/log/nfc_poll.log`** and it is copied into the job directory by Script C for the USB archive. +* **These three tests cannot share the switch.** `blanton_ber.sh`, `blanton_tr518.sh` and `blanton_traffic_linespeed` all take over port loopback and L2 learning. Script C already orders BER after the traffic report. +* **PRBS takes the links down** while `blanton_ber.sh` is armed. That is the test working, not a fault. +* **`ce0` keeps running until you stop it.** An interrupted run leaves the 100G port saturated with nothing in `show interfaces status` to hint at it. +* **Two numbers must stay in step**: Script C's `sleep 30` and `blanton_ber.sh`'s `BER_INTERVAL`. +* **`FAN_SPEED` is 100** — the fans run flat out unless you change `config.ttl`. +* **Declare the switch population** with `SWB_UNIT0` / `SWB_UNIT1`. +* **PRBS on the two 100G uplinks (`port_prbs_monitor.sh`) is still not wired into the macros.** +* **The three items listed as unverified in v1.1.0 are still unverified**: the `ce0` ingress mirror, the 120s settle in `blanton_tr518.sh`, and `test mode nr=no` on stop. See [v1.1.0](https://etgit.et-wen.com/etwen/Blanton_TTL_Script/releases/tag/V1.1.0) for the one-command check on each. + +## 🔗 Links + +* [ARCHITECTURE.md](https://etgit.et-wen.com/etwen/Blanton_TTL_Script/src/branch/main/ARCHITECTURE.md) — test flow, data models, constraints +* [CLAUDE.md](https://etgit.et-wen.com/etwen/Blanton_TTL_Script/src/branch/main/CLAUDE.md) — commands and the bench gotchas + +**Full changelog:** [V1.1.1...V1.1.2](https://etgit.et-wen.com/etwen/Blanton_TTL_Script/compare/V1.1.1...V1.1.2) diff --git a/src/Script_ABC_Blanton/1_Blanton_Script_A.ttl b/src/Script_ABC_Blanton/1_Blanton_Script_A.ttl index 3efa741..35f9bce 100644 --- a/src/Script_ABC_Blanton/1_Blanton_Script_A.ttl +++ b/src/Script_ABC_Blanton/1_Blanton_Script_A.ttl @@ -1,6 +1,6 @@ ; ============================================================================= ; Script A for Blanton -; Version : V1.1.1 +; Version : V1.1.2 ; Date : 2026-08-28 ; Author : ETWen ; ============================================================================= @@ -138,6 +138,29 @@ ; A run that ended badly leaves jobs holding CPU and a ; session open; the next run used to inherit both, and ; the stress numbers were quietly wrong because of it +; V1.1.2 2026-08-28 Blanton_Script/nfc_polling.sh Add background NFC tag polling +; start / stop / status / tail / report / clear / fg +; wraps blantons_nfc_validate.py; start clears the log +; first, so a report covers this run only +; report works WHILE polling - it only reads the log, +; and says RUNNING so a snapshot is not read as final +; Each round appends a machine-readable RESULT line, so +; the tally never depends on the tool's wording +; PASS / FAIL come from the tool's own summary; ERROR +; means it produced neither, which is a different fault +; from a tag that would not read - counted separately +; tail does NOT follow (tail -f never returns, and a +; macro waiting on the prompt would hang there) +; Runs the .py through python3, not ./ - the file is +; mode 100644 and a USB copy carries no execute bit +; Blanton_Script/blantons_nfc_validate.py Vendor NFC/NCI validation tool (unmodified) +; ScriptA NFC baseline: start -> status -> 30s -> stop -> report +; ScriptB Start NFC polling for the soak, then status +; ScriptC Stop NFC polling with the other monitors, report it +; next to the mgmt ping log, and copy nfc_poll.log to +; the job dir for the USB archive +; Fix: that cp said nfc_polling.log - the file is +; nfc_poll.log, so the NFC log never reached the USB ; ============================================================================= include "config.ttl" @@ -306,6 +329,18 @@ sendln "cat ~/Blanton_Script/log/mgmt_ping.log" wait prompt_sonic_root sendln "~/Blanton_Script/mgmt_ping_monitor.sh clear" +; ========== NFC Test ========== +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh start" +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh status" +wait prompt_sonic_root +sendln "sleep 30" +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh stop" +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh report" + ; ========== 100G Port Status ========== wait prompt_sonic_root diff --git a/src/Script_ABC_Blanton/2_Blanton_Script_B.ttl b/src/Script_ABC_Blanton/2_Blanton_Script_B.ttl index a62d545..4332cb0 100644 --- a/src/Script_ABC_Blanton/2_Blanton_Script_B.ttl +++ b/src/Script_ABC_Blanton/2_Blanton_Script_B.ttl @@ -100,6 +100,12 @@ wait prompt_sonic_root pause 3 sendln "./Blanton_Script/mgmt_ping_monitor.sh status" +; ========== NFC Test ========== +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh start" +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh status" + ; ========== 100G Port ========== wait prompt_sonic_root diff --git a/src/Script_ABC_Blanton/3_Blanton_Script_C.ttl b/src/Script_ABC_Blanton/3_Blanton_Script_C.ttl index 02d5838..7ead29b 100644 --- a/src/Script_ABC_Blanton/3_Blanton_Script_C.ttl +++ b/src/Script_ABC_Blanton/3_Blanton_Script_C.ttl @@ -25,6 +25,8 @@ wait prompt_sonic_root sendln "~/Blanton_Script/mgmt_ping_monitor.sh status" wait prompt_sonic_root sendln "~/Blanton_Script/mgmt_ping_monitor.sh stop" +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh stop" ;============================================================================================================= ; 100G PRBS STOP @@ -92,6 +94,10 @@ sendln "cat /host/hw-eval/jobs/qfx5252-stress-usb.log" wait prompt_sonic_root sendln "cat ~/Blanton_Script/log/mgmt_ping.log" +; ========== NFC Test result ========== +wait prompt_sonic_root +sendln "~/Blanton_Script/nfc_polling.sh report" + ; ========== 100G Port ========== wait prompt_sonic_root sendln "show interfaces status Ethernet513,Ethernet514" @@ -201,6 +207,8 @@ wait prompt_sonic_root sendln "cp ./Blanton_Script/log/bmc_poll.log /host/hw-eval/jobs/" wait prompt_sonic_root sendln "cp ./Blanton_Script/log/mgmt_ping.log /host/hw-eval/jobs/" +wait prompt_sonic_root +sendln "cp ./Blanton_Script/log/nfc_poll.log /host/hw-eval/jobs/" wait prompt_sonic_root getdate ts_date "%Y%m%d" diff --git a/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py b/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py new file mode 100644 index 0000000..7e01703 --- /dev/null +++ b/src/Script_ABC_Blanton/Blanton_Script/blantons_nfc_validate.py @@ -0,0 +1,1338 @@ +#!/usr/bin/env python3 +import argparse +import datetime +import errno +import fcntl +import glob +import json +import mmap +import os +from pathlib import Path +import re +import socket +import struct +import sys +import time + + +TOOL_VERSION = "2.3" +EVIDENCE_SCHEMA_VERSION = 1 + +FPGA_VENDOR_ID = 0x1590 +FPGA_FUNCTION0_DEVICE_ID = 0x0180 +FPGA_ID = 0x2001 +NFC_I2C_CHANNEL = 13 +NFC_I2C_ADDRESS = 0x28 + +NFC_CTRL_REG = 0x680 +NFC_STAT_REG = 0x684 +NFC_STAT_PLRTY_REG = 0x688 +NFC_STAT_EDG_LVL_REG = 0x68C +NFC_STAT_INT_EN_REG = 0x690 +NFC_STAT_INT_STAT_REG = 0x694 +NFC_STAT_INT_CLR_REG = 0x698 +NFC_SEM_REG = 0x69C +FPGA_REV_ID_REG = 0x000 +FUNC_CAP_2_REG = 0x048 + +NFC_ENABLE_VALUE = 0x6 +NFC_DISABLE_VALUE = 0x9 +NFC_PRESENT_MASK = 1 << 0 +NFC_INTERRUPT_MASK = 1 << 2 + +I2C_SLAVE = 0x0703 +# The target ocores driver maps a read-address NACK to Linux ECOMM (70). +RETRYABLE_READ_ERRNOS = { + errno.ENXIO, + errno.EAGAIN, + getattr(errno, "ECOMM", 70), +} +if hasattr(errno, "EREMOTEIO"): + RETRYABLE_READ_ERRNOS.add(errno.EREMOTEIO) + +NCI_MT_COMMAND = 1 +NCI_MT_RESPONSE = 2 +NCI_MT_NOTIFICATION = 3 +NCI_GID_RF_MANAGEMENT = 0x01 +NCI_OID_RF_INTF_ACTIVATED = 0x05 + +NCI_INIT_COMMANDS = ( + ("core_reset", bytes.fromhex("20 00 01 01")), + ("core_init", bytes.fromhex("20 01 00")), + ("system_enable", bytes.fromhex("2F 02 00")), + ("discover_map", bytes.fromhex("21 00 04 01 02 01 01")), +) +NCI_RF_DISCOVER = bytes.fromhex("21 03 09 04 00 01 01 01 02 01 02 01") +NCI_RF_DEACTIVATE = bytes.fromhex("21 06 01 00") +NCI_COMMANDS = NCI_INIT_COMMANDS + (("rf_discover", NCI_RF_DISCOVER),) + +POLL_A_MODES = {0x00, 0x03} +POLL_B_MODES = {0x01} +POLL_F_MODES = {0x02, 0x05} + +TECHNOLOGY_NAMES = { + "A": "Type A", + "B": "Type B", + "F": "Type F", +} + + +class ValidationError(Exception): + """Raised when a validation requirement is not met.""" + + +class NciTimeout(ValidationError): + """Raised when no complete NCI frame arrives before a deadline.""" + + +class Evidence: + """Collect timestamped evidence while also emitting concise console output.""" + + def __init__(self, mode, debug=False): + self.start_monotonic = time.monotonic() + self.debug = debug + self.current_section = None + self.data = { + "schema_version": EVIDENCE_SCHEMA_VERSION, + "tool_version": TOOL_VERSION, + "mode": mode, + "started_utc": utc_now(), + "host": socket.gethostname(), + "events": [], + "sections": [], + "nci_frames": [], + "timing": [], + "result": "RUNNING", + } + + def section(self, title, debug_only=False): + if title == self.current_section: + return + self.current_section = title + self.data["sections"].append( + { + "elapsed_ms": round( + (time.monotonic() - self.start_monotonic) * 1000, 3 + ), + "title": title, + "debug_only": debug_only, + } + ) + if not debug_only or self.debug: + print("\n== {} ==".format(title), flush=True) + + def event(self, level, message, **details): + entry = { + "elapsed_ms": round((time.monotonic() - self.start_monotonic) * 1000, 3), + "level": level, + "message": message, + } + if details: + entry["details"] = details + self.data["events"].append(entry) + if level != "DEBUG" or self.debug: + indent = " " if level == "DEBUG" else "" + print("{}[{}] {}".format(indent, level, message), flush=True) + + def frame(self, direction, raw, purpose, retries=0): + entry = { + "elapsed_ms": round((time.monotonic() - self.start_monotonic) * 1000, 3), + "direction": direction, + "purpose": purpose, + "bytes": format_hex(raw), + } + if retries: + entry["read_retries"] = retries + self.data["nci_frames"].append(entry) + if self.debug: + print( + " [DEBUG] {:<4} {:<18} {}".format( + direction, purpose, entry["bytes"] + ), + flush=True, + ) + + +class FpgaWindow: + """Page-sized MMIO view of CB FPGA function 0 BAR0.""" + + def __init__(self, resource_path, writable=False): + self.resource_path = str(resource_path) + self.writable = writable + self.stream = None + self.mapping = None + + def __enter__(self): + mode = "r+b" if self.writable else "rb" + access = mmap.ACCESS_WRITE if self.writable else mmap.ACCESS_READ + self.stream = open(self.resource_path, mode, buffering=0) + try: + self.mapping = mmap.mmap( + self.stream.fileno(), mmap.PAGESIZE, access=access, offset=0 + ) + except Exception: + self.stream.close() + self.stream = None + raise + return self + + def __exit__(self, exc_type, exc_value, traceback): + if self.mapping is not None: + self.mapping.close() + if self.stream is not None: + self.stream.close() + + def read_u32(self, offset): + self._check_offset(offset) + return struct.unpack_from(" mmap.PAGESIZE or offset % 4: + raise ValidationError("invalid FPGA register offset 0x{:x}".format(offset)) + + +class NciFrame: + """Decoded unfragmented NCI control frame.""" + + def __init__(self, raw): + if len(raw) < 3: + raise ValidationError("NCI frame is shorter than its 3-byte header") + declared = raw[2] + if len(raw) != declared + 3: + raise ValidationError( + "NCI length mismatch: declared {} payload bytes, received {}".format( + declared, len(raw) - 3 + ) + ) + if raw[0] & 0x10: + raise ValidationError("fragmented NCI control packets are not supported") + if raw[1] & 0xC0: + raise ValidationError("NCI header has nonzero reserved OID bits") + self.raw = raw + self.mt = (raw[0] >> 5) & 0x07 + self.gid = raw[0] & 0x0F + self.oid = raw[1] & 0x3F + self.payload = raw[3:] + + def is_response_to(self, command): + request = NciFrame(command) + return ( + request.mt == NCI_MT_COMMAND + and self.mt == NCI_MT_RESPONSE + and self.gid == request.gid + and self.oid == request.oid + ) + + def is_activation_notification(self): + return ( + self.mt == NCI_MT_NOTIFICATION + and self.gid == NCI_GID_RF_MANAGEMENT + and self.oid == NCI_OID_RF_INTF_ACTIVATED + ) + + +class NciTransport: + """Polling-mode NCI transport over Linux i2c-dev.""" + + def __init__(self, device_path, address, poll_interval_s, evidence): + self.device_path = str(device_path) + self.address = address + self.poll_interval_s = poll_interval_s + self.evidence = evidence + self.fd = None + + def __enter__(self): + try: + self.fd = os.open(self.device_path, os.O_RDWR | os.O_CLOEXEC) + fcntl.ioctl(self.fd, I2C_SLAVE, self.address) + except OSError as exc: + if self.fd is not None: + os.close(self.fd) + self.fd = None + if exc.errno == errno.EBUSY: + raise ValidationError( + "I2C address 0x{:02x} is busy; another kernel client owns it".format( + self.address + ) + ) + raise ValidationError( + "cannot open {} at address 0x{:02x}: {}".format( + self.device_path, self.address, exc + ) + ) + return self + + def __exit__(self, exc_type, exc_value, traceback): + if self.fd is not None: + os.close(self.fd) + self.fd = None + + def send(self, raw, purpose): + if self.fd is None: + raise ValidationError("NCI transport is not open") + validate_command_frame(raw) + try: + written = os.write(self.fd, raw) + except OSError as exc: + raise ValidationError("NCI write failed for {}: {}".format(purpose, exc)) + if written != len(raw): + raise ValidationError( + "short NCI write for {}: {} of {} bytes".format( + purpose, written, len(raw) + ) + ) + self.evidence.frame("SEND", raw, purpose) + + def receive(self, timeout_s, purpose, retry_interval_s=None): + if self.fd is None: + raise ValidationError("NCI transport is not open") + if retry_interval_s is None: + retry_interval_s = self.poll_interval_s + deadline = time.monotonic() + timeout_s + retries = [0] + header = self._read_exact(3, deadline, retries, retry_interval_s) + payload = self._read_exact( + header[2], deadline, retries, retry_interval_s + ) + raw = header + payload + frame = NciFrame(raw) + self.evidence.frame("RECV", raw, purpose, retries=retries[0]) + return frame, time.monotonic() + + def _read_exact(self, count, deadline, retries, retry_interval_s): + data = bytearray() + while len(data) < count: + if time.monotonic() >= deadline: + raise NciTimeout( + "timed out while reading {} of {} NCI bytes".format( + len(data), count + ) + ) + try: + chunk = os.read(self.fd, count - len(data)) + except OSError as exc: + if exc.errno not in RETRYABLE_READ_ERRNOS: + raise ValidationError("NCI I2C read failed: {}".format(exc)) + retries[0] += 1 + time.sleep(retry_interval_s) + continue + if not chunk: + raise ValidationError("NCI I2C read returned end-of-file") + data.extend(chunk) + return bytes(data) + + +class NciSession: + """Run command/response sequencing and enforce the vendor timing rule.""" + + def __init__( + self, + transport, + evidence, + response_timeout_s, + max_gap_s, + command_settle_s, + activation_poll_interval_s, + ): + self.transport = transport + self.evidence = evidence + self.response_timeout_s = response_timeout_s + self.max_gap_s = max_gap_s + self.command_settle_s = command_settle_s + self.activation_poll_interval_s = activation_poll_interval_s + self.last_response_time = None + self.rf_started = False + + def command(self, name, raw, enforce_gap=True): + send_time = time.monotonic() + if self.last_response_time is not None: + gap = send_time - self.last_response_time + self.evidence.data["timing"].append( + { + "from": "previous_response", + "to": name, + "gap_ms": round(gap * 1000, 3), + "limit_ms": round(self.max_gap_s * 1000, 3), + } + ) + if enforce_gap and gap >= self.max_gap_s: + raise ValidationError( + "response-to-command gap before {} was {:.3f} ms; vendor limit is less than {:.3f} ms".format( + name, gap * 1000, self.max_gap_s * 1000 + ) + ) + self.transport.send(raw, name) + time.sleep(self.command_settle_s) + if raw == NCI_RF_DISCOVER: + self.rf_started = True + deadline = time.monotonic() + self.response_timeout_s + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise NciTimeout("no matching response for {}".format(name)) + frame, received_time = self.transport.receive(remaining, name) + if frame.mt == NCI_MT_NOTIFICATION: + self.evidence.event( + "DEBUG", + "notification received while waiting for {} response".format(name), + frame=format_hex(frame.raw), + ) + continue + if frame.mt != NCI_MT_RESPONSE: + raise ValidationError( + "unexpected NCI message type {} while waiting for {}".format( + frame.mt, name + ) + ) + if not frame.is_response_to(raw): + raise ValidationError( + "response {} does not match command {}".format( + format_hex(frame.raw), name + ) + ) + if not frame.payload: + raise ValidationError("{} response has no status byte".format(name)) + if frame.payload[0] != 0: + raise ValidationError( + "{} response status is 0x{:02X}, expected 0x00".format( + name, frame.payload[0] + ) + ) + self.last_response_time = received_time + return frame + + def wait_for_activation(self, timeout_s): + deadline = time.monotonic() + timeout_s + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: + raise NciTimeout("no RF interface activation notification received") + frame, _ = self.transport.receive( + remaining, + "activation_wait", + retry_interval_s=self.activation_poll_interval_s, + ) + if frame.is_activation_notification(): + return frame + if frame.mt == NCI_MT_NOTIFICATION: + self.evidence.event( + "DEBUG", + "non-activation notification received", + frame=format_hex(frame.raw), + ) + continue + raise ValidationError( + "unexpected non-notification while waiting for activation: {}".format( + format_hex(frame.raw) + ) + ) + + def deactivate(self, cleanup=False): + response = self.command( + "rf_deactivate", NCI_RF_DEACTIVATE, enforce_gap=not cleanup + ) + self.rf_started = False + return response + + +class Cursor: + """Bounds-checked byte cursor used by the activation parser.""" + + def __init__(self, data, label): + self.data = data + self.label = label + self.offset = 0 + + def byte(self, field): + if self.offset >= len(self.data): + raise ValidationError( + "{} is truncated before {}".format(self.label, field) + ) + value = self.data[self.offset] + self.offset += 1 + return value + + def take(self, count, field): + if count < 0 or self.offset + count > len(self.data): + raise ValidationError( + "{} is truncated in {}".format(self.label, field) + ) + value = self.data[self.offset : self.offset + count] + self.offset += count + return value + + def remaining(self): + return len(self.data) - self.offset + + +class ActiveTestState: + """Track resources that require cleanup after active testing.""" + + def __init__(self): + self.fpga = None + self.transport = None + self.session = None + + +def utc_now(): + return datetime.datetime.now(datetime.timezone.utc).isoformat() + + +def format_hex(raw): + return " ".join("{:02X}".format(value) for value in raw) + + +def parse_int_file(path): + return int(Path(path).read_text(encoding="ascii").strip(), 0) + + +def read_optional(path): + try: + return Path(path).read_text(encoding="utf-8", errors="replace").strip() + except OSError: + return None + + +def symlink_target_name(path): + try: + return Path(path).resolve(strict=True).name + except OSError: + return None + + +def validate_command_frame(raw): + frame = NciFrame(raw) + if frame.mt != NCI_MT_COMMAND: + raise ValidationError("frame is not an NCI command: {}".format(format_hex(raw))) + + +def discover_fpga_function0(pci_root="/sys/bus/pci/devices"): + candidates = [] + for device_path in sorted(Path(pci_root).glob("*")): + try: + vendor = parse_int_file(device_path / "vendor") + device = parse_int_file(device_path / "device") + except (OSError, ValueError): + continue + if vendor != FPGA_VENDOR_ID or device != FPGA_FUNCTION0_DEVICE_ID: + continue + resource = device_path / "resource0" + if not resource.exists(): + raise ValidationError("{} has no resource0".format(device_path)) + candidates.append( + { + "bdf": device_path.name, + "vendor": "0x{:04x}".format(vendor), + "device": "0x{:04x}".format(device), + "driver": symlink_target_name(device_path / "driver"), + "resource0": str(resource), + } + ) + if len(candidates) != 1: + raise ValidationError( + "expected one CB FPGA function 0 [{}:{:04x}], found {}".format( + "{:04x}".format(FPGA_VENDOR_ID), + FPGA_FUNCTION0_DEVICE_ID, + len(candidates), + ) + ) + return candidates[0] + + +def discover_i2c_adapter( + adapter_root="/sys/class/i2c-adapter", dev_root="/dev" +): + candidates = [] + for adapter_path_text in sorted(glob.glob(os.path.join(adapter_root, "i2c-*"))): + adapter_path = Path(adapter_path_text) + try: + fpga_id = parse_int_file(adapter_path / "fpga_id") + channel = parse_int_file(adapter_path / "channel") + bus = int(adapter_path.name.split("-", 1)[1]) + except (OSError, ValueError, IndexError): + continue + if fpga_id != FPGA_ID or channel != NFC_I2C_CHANNEL: + continue + device_path = Path(dev_root) / "i2c-{}".format(bus) + candidates.append( + { + "bus": bus, + "name": read_optional(adapter_path / "name"), + "fpga_id": "0x{:04x}".format(fpga_id), + "channel": channel, + "sysfs_path": str(adapter_path), + "device_path": str(device_path), + "device_exists": device_path.exists(), + } + ) + if len(candidates) != 1: + raise ValidationError( + "expected one FPGA 0x{:04x} channel {} adapter, found {}".format( + FPGA_ID, NFC_I2C_CHANNEL, len(candidates) + ) + ) + if not candidates[0]["device_exists"]: + raise ValidationError( + "{} is missing".format(candidates[0]["device_path"]) + ) + return candidates[0] + + +def inspect_i2c_client(bus, address, i2c_devices_root="/sys/bus/i2c/devices"): + client_path = Path(i2c_devices_root) / "{}-{:04x}".format(bus, address) + exists = client_path.exists() + result = { + "address": "0x{:02x}".format(address), + "path": str(client_path), + "registered": exists, + "name": None, + "modalias": None, + "driver": None, + } + if exists: + result.update( + { + "name": read_optional(client_path / "name"), + "modalias": read_optional(client_path / "modalias"), + "driver": symlink_target_name(client_path / "driver"), + } + ) + return result + + +def find_open_device_users(device_path, proc_root="/proc"): + users = [] + target = os.path.realpath(device_path) + for process_path in sorted(Path(proc_root).glob("[0-9]*")): + if not process_path.name.isdigit(): + continue + for fd_path in (process_path / "fd").glob("*"): + try: + linked = os.path.realpath(fd_path) + except OSError: + continue + if linked != target: + continue + command_raw = read_optional(process_path / "cmdline") or "" + users.append( + { + "pid": int(process_path.name), + "fd": fd_path.name, + "command": command_raw.replace("\x00", " ").strip(), + } + ) + return users + + +def remove_unbound_i2c_client(adapter, client, evidence): + if not client["registered"]: + return client + if client["driver"] is not None: + raise ValidationError( + "refusing to remove bound I2C client at 0x28 (driver={})".format( + client["driver"] + ) + ) + delete_path = Path(adapter["sysfs_path"]) / "delete_device" + try: + with open(delete_path, "w", encoding="ascii") as stream: + stream.write("0x{:02x}\n".format(NFC_I2C_ADDRESS)) + except OSError as exc: + raise ValidationError( + "failed to remove unbound I2C client through {}: {}".format( + delete_path, exc + ) + ) + current = inspect_i2c_client(adapter["bus"], NFC_I2C_ADDRESS) + if current["registered"]: + raise ValidationError("I2C client at 0x28 remains registered after removal") + evidence.data["removed_unbound_client"] = client + evidence.data["reboot_required"] = True + evidence.event( + "WARN", + "removed unbound client at 0x28; reboot is required after testing", + removed_client=client, + ) + return current + + +def read_nfc_registers(resource_path): + definitions = ( + (FPGA_REV_ID_REG, "fpga_rev_id"), + (FUNC_CAP_2_REG, "func_cap_2"), + (NFC_CTRL_REG, "nfc_ctrl"), + (NFC_STAT_REG, "nfc_stat"), + (NFC_STAT_PLRTY_REG, "nfc_stat_polarity"), + (NFC_STAT_EDG_LVL_REG, "nfc_stat_edge_level"), + (NFC_STAT_INT_EN_REG, "nfc_stat_int_enable"), + (NFC_STAT_INT_STAT_REG, "nfc_stat_int_status"), + (NFC_STAT_INT_CLR_REG, "nfc_stat_int_clear"), + (NFC_SEM_REG, "nfc_semaphore"), + ) + with FpgaWindow(resource_path, writable=False) as fpga: + values = {name: fpga.read_u32(offset) for offset, name in definitions} + result = {name: "0x{:08x}".format(value) for name, value in values.items()} + result["present"] = bool(values["nfc_stat"] & NFC_PRESENT_MASK) + result["interrupt"] = bool(values["nfc_stat"] & NFC_INTERRUPT_MASK) + return result + + +def collect_inventory(): + fpga = discover_fpga_function0() + adapter = discover_i2c_adapter() + client = inspect_i2c_client(adapter["bus"], NFC_I2C_ADDRESS) + try: + registers = read_nfc_registers(fpga["resource0"]) + register_error = None + except (OSError, ValueError, ValidationError) as exc: + registers = None + register_error = str(exc) + return { + "system": { + "hostname": socket.gethostname(), + "kernel": os.uname().release, + "sonic_version_file": read_optional("/etc/sonic/sonic_version.yml"), + }, + "fpga": fpga, + "i2c_adapter": adapter, + "address_owner": client, + "open_i2c_users": find_open_device_users(adapter["device_path"]), + "registers": registers, + "register_error": register_error, + "traffic_generated": False, + "hardware_writes": False, + } + + +def parse_tag_id(frame): + if not frame.is_activation_notification(): + raise ValidationError("frame is not RF_INTF_ACTIVATED_NTF (61 05)") + cursor = Cursor(frame.payload, "RF_INTF_ACTIVATED_NTF") + discovery_id = cursor.byte("discovery ID") + rf_interface = cursor.byte("RF interface") + protocol = cursor.byte("RF protocol") + mode = cursor.byte("activation RF technology and mode") + max_payload = cursor.byte("maximum data payload") + credits = cursor.byte("initial credits") + tech_length = cursor.byte("technology parameter length") + tech_params = cursor.take(tech_length, "technology parameters") + data_mode = cursor.byte("data exchange RF technology and mode") + tx_bitrate = cursor.byte("TX bit rate") + rx_bitrate = cursor.byte("RX bit rate") + activation_length = cursor.byte("activation parameter length") + activation_params = cursor.take(activation_length, "activation parameters") + if cursor.remaining() != 0: + raise ValidationError( + "RF_INTF_ACTIVATED_NTF has {} trailing bytes".format(cursor.remaining()) + ) + + if mode in POLL_A_MODES: + technology, identifier_kind, identifier, details = parse_poll_a(tech_params) + elif mode in POLL_B_MODES: + technology, identifier_kind, identifier, details = parse_poll_b(tech_params) + elif mode in POLL_F_MODES: + technology, identifier_kind, identifier, details = parse_poll_f(tech_params) + else: + raise ValidationError( + "unsupported activation RF technology/mode 0x{:02X}; raw frame preserved".format( + mode + ) + ) + + return { + "discovery_id": discovery_id, + "rf_interface": "0x{:02x}".format(rf_interface), + "protocol": "0x{:02x}".format(protocol), + "activation_mode": "0x{:02x}".format(mode), + "technology": technology, + "identifier_kind": identifier_kind, + "identifier": identifier.hex().upper(), + "max_data_payload": max_payload, + "initial_credits": credits, + "data_mode": "0x{:02x}".format(data_mode), + "tx_bitrate_code": "0x{:02x}".format(tx_bitrate), + "rx_bitrate_code": "0x{:02x}".format(rx_bitrate), + "activation_parameters": format_hex(activation_params), + "technology_details": details, + } + + +def parse_poll_a(params): + cursor = Cursor(params, "Type A technology parameters") + sens_res = cursor.take(2, "SENS_RES") + identifier_length = cursor.byte("NFCID1 length") + if identifier_length not in (4, 7, 10): + raise ValidationError( + "invalid Type A NFCID1 length {}".format(identifier_length) + ) + identifier = cursor.take(identifier_length, "NFCID1") + sel_res_length = cursor.byte("SEL_RES length") + if sel_res_length not in (0, 1): + raise ValidationError( + "invalid Type A SEL_RES length {}".format(sel_res_length) + ) + sel_res = cursor.take(sel_res_length, "SEL_RES") + remaining = cursor.remaining() + hr = b"" + if remaining: + hr_length = cursor.byte("HR length") + if hr_length not in (0, 2): + raise ValidationError("invalid Type A HR length {}".format(hr_length)) + hr = cursor.take(hr_length, "HR") + if cursor.remaining(): + raise ValidationError("Type A technology parameters contain trailing bytes") + return ( + "A", + "NFCID1", + identifier, + { + "sens_res": format_hex(sens_res), + "sel_res": format_hex(sel_res), + "hr": format_hex(hr), + }, + ) + + +def parse_poll_b(params): + cursor = Cursor(params, "Type B technology parameters") + sensb_length = cursor.byte("SENSB_RES length") + if sensb_length not in (11, 12): + raise ValidationError( + "invalid Type B SENSB_RES length {}".format(sensb_length) + ) + sensb_res = cursor.take(sensb_length, "SENSB_RES") + if cursor.remaining(): + raise ValidationError("Type B technology parameters contain trailing bytes") + return ( + "B", + "NFCID0", + sensb_res[:4], + {"sensb_res": format_hex(sensb_res)}, + ) + + +def parse_poll_f(params): + cursor = Cursor(params, "Type F technology parameters") + bit_rate = cursor.byte("bit-rate code") + sensf_length = cursor.byte("SENSF_RES length") + if sensf_length not in (16, 18): + raise ValidationError( + "invalid Type F SENSF_RES length {}".format(sensf_length) + ) + sensf_res = cursor.take(sensf_length, "SENSF_RES") + if cursor.remaining(): + raise ValidationError("Type F technology parameters contain trailing bytes") + return ( + "F", + "NFCID2", + sensf_res[:8], + { + "bit_rate_code": "0x{:02x}".format(bit_rate), + "sensf_res": format_hex(sensf_res), + }, + ) + + +def normalize_identifier(value): + if re.search(r"[^0-9A-Fa-f:\-\s]", value): + raise argparse.ArgumentTypeError( + "identifier may contain only hex digits, spaces, colons, and hyphens" + ) + compact = re.sub(r"[:\-\s]", "", value) + if not compact or len(compact) % 2: + raise argparse.ArgumentTypeError("identifier must contain complete hex bytes") + try: + return bytes.fromhex(compact) + except ValueError as exc: + raise argparse.ArgumentTypeError(str(exc)) + + +def register_snapshot(fpga): + ctrl = fpga.read_u32(NFC_CTRL_REG) + stat = fpga.read_u32(NFC_STAT_REG) + return { + "ctrl": "0x{:08x}".format(ctrl), + "stat": "0x{:08x}".format(stat), + "present": bool(stat & NFC_PRESENT_MASK), + "interrupt": bool(stat & NFC_INTERRUPT_MASK), + } + + +def run_active_test(args, evidence, inventory): + if os.geteuid() != 0: + raise ValidationError("active test must run as root") + if inventory["open_i2c_users"]: + raise ValidationError( + "refusing active test: {} is already open by another process".format( + inventory["i2c_adapter"]["device_path"] + ) + ) + if inventory["register_error"]: + raise ValidationError( + "cannot read FPGA registers: {}".format(inventory["register_error"]) + ) + owner = inventory["address_owner"] + pending_owner_removal = None + if owner["registered"]: + if owner["driver"] is not None: + raise ValidationError( + "refusing to remove bound I2C client at 0x28 (driver={})".format( + owner["driver"] + ) + ) + pending_owner_removal = owner + + validation_requested = ( + args.expected_id is not None or args.expected_technology is not None + ) + evidence.section("ACTIVE TEST") + evidence.event( + "INFO", + "mode={} iterations={}".format( + "validate" if validation_requested else "read_display", + args.iterations, + ), + ) + + resource = inventory["fpga"]["resource0"] + state = ActiveTestState() + cleanup_errors = [] + activation = None + try: + evidence.section("HARDWARE SETUP", debug_only=True) + state.fpga = FpgaWindow(resource, writable=True) + fpga = state.fpga.__enter__() + evidence.data["active_test"] = { + "initial_registers": register_snapshot(fpga), + "ven_low_ms": args.ven_low_ms, + "post_enable_ms": args.post_enable_ms, + "response_timeout_s": args.response_timeout_s, + "command_settle_ms": args.command_settle_ms, + "activation_timeout_s": args.activation_timeout_s, + "activation_poll_interval_ms": args.activation_poll_interval_ms, + "max_response_gap_ms": args.max_response_gap_ms, + "requested_iterations": args.iterations, + "presence_policy": "warn_and_continue", + "presence_warning": False, + } + + fpga.write_u32(NFC_CTRL_REG, NFC_DISABLE_VALUE) + disabled = register_snapshot(fpga) + evidence.event("DEBUG", "VEN driven low", registers=disabled) + if (fpga.read_u32(NFC_CTRL_REG) & 0xF) != NFC_DISABLE_VALUE: + raise ValidationError("NFC disable value did not read back as 0x9") + time.sleep(args.ven_low_ms / 1000.0) + + fpga.write_u32(NFC_CTRL_REG, NFC_ENABLE_VALUE) + time.sleep(args.post_enable_ms / 1000.0) + enabled = register_snapshot(fpga) + evidence.event("DEBUG", "VEN driven high", registers=enabled) + if (fpga.read_u32(NFC_CTRL_REG) & 0xF) != NFC_ENABLE_VALUE: + raise ValidationError("NFC enable value did not read back as 0x6") + if not enabled["present"]: + evidence.data["active_test"]["presence_warning"] = True + evidence.event( + "WARN", + "FPGA NFC presence bit is 0 after enable", + ) + + adapter = inventory["i2c_adapter"] + if pending_owner_removal is not None: + owner = remove_unbound_i2c_client( + adapter, pending_owner_removal, evidence + ) + inventory["address_owner_after_removal"] = owner + state.transport = NciTransport( + adapter["device_path"], + NFC_I2C_ADDRESS, + args.poll_interval_ms / 1000.0, + evidence, + ) + transport = state.transport.__enter__() + state.session = NciSession( + transport, + evidence, + args.response_timeout_s, + args.max_response_gap_ms / 1000.0, + args.command_settle_ms / 1000.0, + args.activation_poll_interval_ms / 1000.0, + ) + evidence.section("NCI INITIALIZATION", debug_only=True) + for name, command in NCI_INIT_COMMANDS: + state.session.command(name, command) + + evidence.section("TAG READS") + activations = [] + rf_loop_start = time.monotonic() + for iteration in range(1, args.iterations + 1): + evidence.event( + "DEBUG", + "starting RF detection iteration {}/{}".format( + iteration, args.iterations + ), + ) + state.session.command("rf_discover", NCI_RF_DISCOVER) + activation_frame = state.session.wait_for_activation( + args.activation_timeout_s + ) + activation = parse_tag_id(activation_frame) + activation["iteration"] = iteration + activations.append(activation) + evidence.data["activations"] = activations + evidence.event( + "TAG", + ( + "#{:06d} technology={} protocol={} " + "{}={}" + ).format( + iteration, + activation["technology"], + activation["protocol"], + activation["identifier_kind"], + activation["identifier"], + ), + activation=activation, + ) + state.session.deactivate() + + actual_id = bytes.fromhex(activation["identifier"]) + if ( + args.expected_technology + and activation["technology"] != args.expected_technology + ): + raise ValidationError( + "iteration {} tag technology {} does not match expected {}".format( + iteration, + activation["technology"], + args.expected_technology, + ) + ) + if args.expected_id is not None and actual_id != args.expected_id: + raise ValidationError( + "iteration {} tag identifier {} does not match expected {}".format( + iteration, + activation["identifier"], + args.expected_id.hex().upper(), + ) + ) + rf_loop_elapsed = time.monotonic() - rf_loop_start + evidence.data["active_test"]["completed_iterations"] = len(activations) + evidence.data["active_test"]["rf_loop_elapsed_seconds"] = round( + rf_loop_elapsed, 6 + ) + evidence.data["active_test"]["tag_mode"] = ( + "validate" if validation_requested else "read_display" + ) + validated_fields = [] + if args.expected_technology is not None: + validated_fields.append("technology") + if args.expected_id is not None: + validated_fields.append("identifier") + evidence.data["active_test"]["validated_fields"] = validated_fields + if validation_requested: + evidence.event( + "INFO", + "validated tag field(s): {}".format(", ".join(validated_fields)), + ) + evidence.data["active_test"]["final_registers_before_cleanup"] = register_snapshot(fpga) + return activations + finally: + primary_exception = sys.exc_info()[0] is not None + if ( + state.session is not None + and state.session.rf_started + and state.transport is not None + and state.transport.fd is not None + ): + try: + evidence.event("WARN", "attempting RF deactivation during cleanup") + state.session.deactivate(cleanup=True) + except Exception as exc: + cleanup_errors.append("RF deactivation failed: {}".format(exc)) + if state.transport is not None: + state.transport.__exit__(None, None, None) + if state.fpga is not None and state.fpga.mapping is not None: + if not args.leave_enabled: + try: + state.fpga.write_u32(NFC_CTRL_REG, NFC_DISABLE_VALUE) + final_ctrl = state.fpga.read_u32(NFC_CTRL_REG) + if (final_ctrl & 0xF) != NFC_DISABLE_VALUE: + raise ValidationError("disable readback is not 0x9") + evidence.event("DEBUG", "VEN left disabled after test") + except Exception as exc: + cleanup_errors.append("VEN disable failed: {}".format(exc)) + else: + evidence.event("WARN", "VEN left enabled by explicit request") + state.fpga.__exit__(None, None, None) + if cleanup_errors: + evidence.data["cleanup_errors"] = cleanup_errors + if not primary_exception: + raise ValidationError("; ".join(cleanup_errors)) + + +def write_evidence(path, evidence): + evidence["finished_utc"] = utc_now() + output = Path(path) + output.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="ascii") + os.chmod(output, 0o644) + return output + + +def add_output_argument(parser): + parser.add_argument( + "--json", + dest="json_path", + help="optionally save JSON evidence to this path; omitted means console only", + ) + parser.add_argument( + "--debug", + action="store_true", + help="show hardware setup and raw NCI send/receive frames", + ) + + +def build_parser(): + parser = argparse.ArgumentParser( + description="Inventory, read, or validate the Blantons NFC NCI path" + ) + parser.add_argument("--version", action="version", version=TOOL_VERSION) + subparsers = parser.add_subparsers(dest="mode", required=True) + + inventory = subparsers.add_parser( + "inventory", help="read-only discovery; no I2C traffic or hardware writes" + ) + add_output_argument(inventory) + + active = subparsers.add_parser( + "test", help="cycle VEN and execute the vendor NCI polling sequence" + ) + add_output_argument(active) + # Compatibility no-ops for commands written for tool versions 1.0-1.6. + active.add_argument( + "--confirm-active-test", action="store_true", help=argparse.SUPPRESS + ) + active.add_argument( + "--remove-unbound-client", action="store_true", help=argparse.SUPPRESS + ) + active.add_argument( + "--continue-without-presence", action="store_true", help=argparse.SUPPRESS + ) + identity = active.add_mutually_exclusive_group() + identity.add_argument( + "--expected-id", + type=normalize_identifier, + help="validate against a known tag identifier; omit to read and display", + ) + identity.add_argument( + "--accept-any-tag", + action="store_true", + help=argparse.SUPPRESS, + ) + active.add_argument( + "--expected-technology", + choices=("A", "B", "F"), + help="require the activated tag technology", + ) + active.add_argument( + "--ven-low-ms", + type=positive_int, + default=200, + help="VEN-low interval; default 200 ms from Accton/vendor guidance", + ) + active.add_argument( + "--post-enable-ms", + type=positive_int, + default=50, + help="delay after VEN high; default 50 ms from Accton's script", + ) + active.add_argument( + "--poll-interval-ms", + type=positive_int, + default=10, + help="delay between retryable I2C reads; default 10 ms", + ) + active.add_argument( + "--command-settle-ms", + type=positive_int, + default=10, + help="delay after each NCI write before reading; default 10 ms from Accton", + ) + active.add_argument( + "--response-timeout-s", + type=positive_float, + default=2.0, + help="per-command response guardrail; default 2 seconds", + ) + active.add_argument( + "--activation-timeout-s", + type=positive_float, + default=5.0, + help="tag activation guardrail; default 5 seconds", + ) + active.add_argument( + "--activation-poll-interval-ms", + type=positive_int, + default=500, + help="delay after a not-ready tag read; default 500 ms from Accton", + ) + active.add_argument( + "--iterations", + type=positive_int, + default=1, + help="bounded RF discover/activate/deactivate repetitions; default 1", + ) + active.add_argument( + "--max-response-gap-ms", + type=positive_int, + default=1000, + help="vendor response-to-next-command limit; default 1000 ms, enforced as strictly less", + ) + active.add_argument( + "--leave-enabled", + action="store_true", + help="leave VEN enabled after cleanup; default is disabled", + ) + return parser + + +def positive_int(value): + parsed = int(value) + if parsed <= 0: + raise argparse.ArgumentTypeError("value must be greater than zero") + return parsed + + +def positive_float(value): + parsed = float(value) + if parsed <= 0: + raise argparse.ArgumentTypeError("value must be greater than zero") + return parsed + + +def print_inventory(inventory, evidence): + fpga = inventory["fpga"] + adapter = inventory["i2c_adapter"] + owner = inventory["address_owner"] + detail_level = "INFO" if evidence.data["mode"] == "inventory" else "DEBUG" + evidence.event( + detail_level, + "CB FPGA function 0: {} driver={}".format(fpga["bdf"], fpga["driver"]), + ) + evidence.event( + detail_level, + "NFC adapter: {} FPGA={} channel={}".format( + adapter["device_path"], adapter["fpga_id"], adapter["channel"] + ), + ) + if owner["registered"]: + evidence.event( + "WARN", + "address 0x28 has a registered client: name={} driver={}".format( + owner["name"], owner["driver"] or "unbound" + ), + ) + else: + evidence.event(detail_level, "address 0x28 has no registered kernel client") + if inventory["open_i2c_users"]: + evidence.event( + "WARN", + "{} is open by another process".format(adapter["device_path"]), + users=inventory["open_i2c_users"], + ) + if inventory["register_error"]: + evidence.event( + "WARN", "FPGA registers could not be read: {}".format(inventory["register_error"]) + ) + else: + registers = inventory["registers"] + evidence.event( + detail_level, + "NFC registers: ctrl={} stat={} present={} interrupt={}".format( + registers["nfc_ctrl"], + registers["nfc_stat"], + int(registers["present"]), + int(registers["interrupt"]), + ), + ) +def main(argv=None): + args = build_parser().parse_args(argv) + evidence = Evidence(args.mode, debug=args.debug) + json_path = args.json_path + exit_code = 1 + try: + evidence.section("DISCOVERY", debug_only=args.mode == "test") + inventory = collect_inventory() + evidence.data["inventory"] = inventory + print_inventory(inventory, evidence) + if args.mode == "inventory": + evidence.data["result"] = "INVENTORY_ONLY" + exit_code = 0 + else: + activations = run_active_test(args, evidence, inventory) + validation_requested = ( + args.expected_id is not None or args.expected_technology is not None + ) + presence_warning = evidence.data["active_test"].get( + "presence_warning", False + ) + evidence.data["result"] = ( + "PASS_WITH_WARNING" if presence_warning else "PASS" + ) + evidence.data["tag_mode"] = ( + "validate" if validation_requested else "read_display" + ) + evidence.data["detected_identifiers"] = [ + activation["identifier"] for activation in activations + ] + if args.expected_id is not None and args.expected_technology is not None: + message = "NFC tag read; technology and identifier validation passed" + elif args.expected_id is not None: + message = "NFC tag read; identifier validation passed" + elif args.expected_technology is not None: + message = "NFC tag read; technology validation passed" + else: + message = "NFC tag read succeeded" + if presence_warning: + message += "; FPGA module-presence reporting remains failed" + evidence.section("SUMMARY") + evidence.event( + "PASS", + "{} for {} iteration(s)".format(message, len(activations)), + ) + exit_code = 0 + except KeyboardInterrupt: + evidence.section("SUMMARY") + evidence.data["result"] = "FAIL" + evidence.data["failure"] = "interrupted" + evidence.event("FAIL", "test interrupted") + exit_code = 130 + except (OSError, ValueError, ValidationError) as exc: + evidence.section("SUMMARY") + evidence.data["result"] = "FAIL" + evidence.data["failure"] = str(exc) + evidence.event("FAIL", str(exc)) + exit_code = 1 + if json_path is not None: + try: + output = write_evidence(json_path, evidence.data) + print("[INFO] JSON evidence: {}".format(output), flush=True) + except OSError as exc: + print( + "[FAIL] could not write JSON evidence {}: {}".format( + json_path, exc + ), + file=sys.stderr, + ) + exit_code = 1 + return exit_code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/Script_ABC_Blanton/Blanton_Script/nfc_polling.sh b/src/Script_ABC_Blanton/Blanton_Script/nfc_polling.sh new file mode 100644 index 0000000..ec6896c --- /dev/null +++ b/src/Script_ABC_Blanton/Blanton_Script/nfc_polling.sh @@ -0,0 +1,314 @@ +#!/usr/bin/env bash +# ============================================================================= +# nfc_polling.sh -- keep reading the NFC tag in the background, and tally it +# ============================================================================= +# Version History: +# V1.0.0 2026-08-28 First version. start/stop/status/tail/report/clear/fg, +# wraps blantons_nfc_validate.py. 'report' works while the +# poll is still running. +# ============================================================================= +# +# Each round runs blantons_nfc_validate.py once and appends a machine-readable +# verdict line, so 'report' never has to parse the tool's prose: +# +# [2026-08-28 15:04:11] NFC poll #000001 START +# ... the tool's own output ... +# [2026-08-28 15:04:14] NFC poll #000001 RESULT=PASS tag=046F7BD2142290 +# +# The verdict comes from the tool's own [PASS] / [FAIL] summary line. If neither +# appears the round is ERROR, not FAIL -- "the tool did not run" and "the tag was +# not read" are different problems and get counted separately. +# +# Usage: +# ./nfc_polling.sh start # background, CLEARS the log first +# ./nfc_polling.sh status # running? pid, rounds so far, log size +# ./nfc_polling.sh tail [n] # last n lines (default 50) -- does NOT follow +# ./nfc_polling.sh report # PASS/FAIL tally; safe to run WHILE polling +# ./nfc_polling.sh stop +# ./nfc_polling.sh clear # delete the log (must be stopped) +# ./nfc_polling.sh fg # run in the foreground, Ctrl-C to stop +# +# Environment overrides: +# NFC_INTERVAL_SEC=10 NFC_TIMEOUT_SEC=60 NFC_MAX_ROUNDS=0 (0 = forever) +# NFC_ARGS="test --debug" NFC_LOG_DIR= NFC_BIN= +# ============================================================================= + +# --- User Configurable Section ---------------------------------------------- +NFC_INTERVAL_SEC="${NFC_INTERVAL_SEC:-10}" # gap between rounds +NFC_TIMEOUT_SEC="${NFC_TIMEOUT_SEC:-60}" # kill a round that hangs (0 = never) +NFC_MAX_ROUNDS="${NFC_MAX_ROUNDS:-0}" # 0 = until 'stop' + +# Called through python3 on purpose: blantons_nfc_validate.py arrives from git +# (mode 100644) and a Windows/USB copy does not carry the execute bit either, so +# "./blantons_nfc_validate.py" would fail with Permission denied on a fresh DUT. +NFC_PY="${NFC_PY:-python3}" +NFC_SCRIPT_NAME="${NFC_SCRIPT_NAME:-blantons_nfc_validate.py}" + +# --debug is kept because that is the output that has been seen on the bench. +# It costs ~1.5 KB per round (~13 MB/day at a 10s interval), which is fine for a +# soak; drop it to "test" if you want a smaller log. +NFC_ARGS="${NFC_ARGS:-test --debug}" + +NFC_LOG_DIR="${NFC_LOG_DIR:-}" # empty =