diff --git a/.gitignore b/.gitignore index eb382bb..d9a6573 100644 --- a/.gitignore +++ b/.gitignore @@ -10,6 +10,7 @@ dist/ build/ .pytest_cache/ logs/ +runtime_logs/ pr_body.md inventory_results.txt inventory_results.csv diff --git a/README.md b/README.md index 7d3809a..83b78d1 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,31 @@ 通常使う入口は、原則として次の2つです。 +54コマンドの仕様照合・USM02実機事前確認は、通常Inventoryとは分けて次の専用CLIを使います。 + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_v117_validation_cli --rom-number 2052 +``` + +既定はdry-runで、COMポートを開きません。詳細は [USM02 Ver.1.17 54コマンド検証](docs/usm02_v117_54_command_validation.md) を参照してください。 + +bootstrap後のブザー・ANT1一時切替・周波数開始CH一時変更は、次のPhase 2 CLIをdry-runしてから実行します。 + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_safe_controls_cli --target-antenna 1 +``` + +ANT0・ANT1・ANT2を順番に1回ずつInventoryし、タグ応答があったANTだけブザーを鳴らす確認は、次の専用CLIを使います。 + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_inventory_buzzer_cli --antennas 0,1,2 +``` + +実機実行時も最初にROM、設定、ANT0〜ANT3の物理接続を再取得します。指定ANTが1つでも接続OKでない場合、Inventory前に停止します。タグ固有値は表示・JSON保存しません。 + ### 1. 標準Inventoryフロー ```powershell diff --git a/docs/usm02_v117_54_command_validation.md b/docs/usm02_v117_54_command_validation.md new file mode 100644 index 0000000..95e8485 --- /dev/null +++ b/docs/usm02_v117_54_command_validation.md @@ -0,0 +1,177 @@ +# USM02 Ver.1.17 54コマンド検証 + +## 目的 + +UTR-SUN02-4CH / ROMシリーズUSM02について、通信プロトコル説明書Ver.1.17の通常コマンド54件を、漏れと重複なく検証するための入口です。 + +54件を一律に「ACK確認」とは扱いません。正常時の応答形態、機種・ROM対応、安全条件により、次を区別します。 + +- 単一ACK +- 設定によりACKまたは無応答 +- タグ応答0件以上＋完了ACK +- 仕様上の無応答 +- NACK（解析済みでも正常完了にはしない） +- USM02では非対応 +- 実機ROMでは非対応 +- 条件未準備 +- ユーザー判断で非実行 + +## 対象実機 + +- 機種: UTR-SUN02-4CH +- ROMシリーズ: USM02 +- 接続: USBシリアル +- 既定ポート: COM6 +- 既定速度: 115200 bps + +## 1. 実機に接続しないdry-run + +PowerShellでリポジトリ直下から実行します。 + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_v117_validation_cli --rom-number 2052 +``` + +`2052`はVer.2.052を仮定する指定です。実機値ではありません。実ROMが不明な場合は `--rom-number` を省略してください。 + +dry-runではCOMポートを開かず、54件の計画と集計だけを表示します。 + +## 2. 実機の事前情報だけを取得 + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_v117_validation_cli ` + --execute-bootstrap ` + --port COM6 ` + --baudrate 115200 ` + --json-out runtime_logs\usm02_v117_bootstrap.json +``` + +この段階で送信するのは、次の事前確認だけです。 + +1. ROMバージョンを読み、USM02 / UTR-SUN02-4CHを確定する。 +2. RAM上の動作モードをコマンドモードへ切り替える。 +3. コマンドモードのアンテナ切替設定を読む。 +4. ANT0〜ANT3へUHF_CheckAntennaを送り、物理接続状態を確認する。 +5. Inventoryパラメータを読み、TID付加設定を確認する。 +6. コマンドモードと自動読み取りモードのEPC(UII)関連設定を読み、応答件数・順序へ影響する設定を確認する。 + +次の3種類のアンテナ数は別々に記録します。 + +| 項目 | 取得元 | 意味 | +|---|---|---| +| 物理容量 | ROMシリーズから確定した機種仕様 | USM02ではANT0〜ANT3の4ポート | +| 設定アンテナ | 7.4.5 アンテナ切替設定 | 現在「使用する」と設定されたアンテナ | +| 接続OKアンテナ | 7.3.5 UHF_CheckAntenna | 実際に接続OKと応答したアンテナ | + +## 3. 2バイト目の解釈 + +応答の2バイト目は、常にアンテナ番号ではありません。 + +- RFタグデータ応答（コマンド `6Ch`）かつ「アンテナID出力=有効」の場合だけ、アンテナ番号として解釈します。 +- 通常ACK、NACK、完了ACK、または「アンテナID出力=無効」のRFタグ応答では、リーダライタIDとして解釈します。 + +## 4. 強制安全ルール + +- Kill、Lock、UHF_Encode、FLASH設定初期化は送信しません。 +- 予期しないNACK、timeout、復元失敗が発生したら後続送信を止めます。 +- Access/Kill Passwordをログへ出しません。 +- PC/UII/EPC/TIDの実値を共有用ログへ出しません。 +- 設定変更は事前読取、同値書込、再読取、復元確認の順で行います。 +- タグ書込は廃棄可能タグのUser領域だけを対象とし、事前読取と復元を必須とします。 +- UHF_ThroughCmdは、TIDでタグICを特定し、そのIC向け安全コマンドを確定できるまで実行しません。 + +## 5. 現在の実装範囲 + +現時点のCLIは、54件の仕様カタログ、応答検証ロジック、dry-run、実機事前確認までを実装しています。54件すべてを自動送信する機能ではありません。 + +まず実機事前確認結果からROMと設定依存条件を確定し、その結果に基づいて実行可能なコマンドだけを次の実装単位へ進めます。これにより、非対応コマンドや破壊的コマンドを「54件達成」のために誤送信することを防ぎます。 + +## 6. Phase 2: 安全制御確認 + +bootstrapが正常で、USM02、接続ANT、設定値を確認できた場合だけ実行します。 + +dry-run: + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_safe_controls_cli --target-antenna 1 +``` + +実機実行: + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_safe_controls_cli ` + --execute ` + --port COM6 ` + --baudrate 115200 ` + --target-antenna 1 ` + --verify-rf-channel ` + --json-out runtime_logs\usm02_phase2_safe_controls.json +``` + +実行内容: + +1. ROM、応答依存設定、ANT0〜ANT3の接続状態を再取得する。 +2. ブザー制御を応答要求ありで1回実行し、ACKを確認する。 +3. コマンドモードRAMのアンテナを接続OKのANT1へ一時変更する。 +4. アンテナ設定を読み戻す。 +5. コマンドモードRAMの周波数開始CHを、現在使用許可されている26〜32ch内の別CHへ一時変更する。 +6. 周波数設定を読み戻し、使用許可CHマスクが不変であることを確認する。 +7. Inventoryを1回だけ実行し、タグ固有値を保存せず、タグ応答ANTと完了ACKの実使用CHを確認する。 +8. 周波数開始CHを元へ戻して読み戻す。 +9. アンテナ設定を元へ戻して読み戻す。 + +周波数確認で変更するのはRAM上の「開始チャンネル番号」だけです。使用許可CHは変更せず、FLASHへ保存しません。`--verify-rf-channel` を指定した場合だけInventoryを1回実行します。タグのPC/UII/EPC/TIDは表示・保存せず、応答件数、アンテナ番号、完了ACKの実使用CHだけを残します。「現在設定されているチャンネル番号」は最後にキャリア出力したCHなので、Inventory前の読戻し時点では変化していなくても正常です。 + +例外、NACK、読戻し不一致の場合も `finally` で周波数、アンテナの順に復元します。復元確認に失敗した場合は後続操作を行わず、UTRRWManagerまたは読取コマンドで現在値を確認してください。 + +## 7. ANT0〜2 Inventory・タグ検出時ブザー確認 + +ANT0、ANT1、ANT2へタグを1枚ずつ置いた状態で、アンテナ別の読取と条件付きブザーを確認します。これは54コマンド全件検証とは分けた、限定的な実機確認です。 + +dry-run: + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_inventory_buzzer_cli --antennas 0,1,2 +``` + +実機実行: + +```powershell +$env:PYTHONPATH = "." +py -m src.utr_usm02_inventory_buzzer_cli ` + --execute ` + --port COM6 ` + --baudrate 115200 ` + --antennas 0,1,2 ` + --json-out runtime_logs\usm02_ant012_inventory_buzzer.json +``` + +実行順序: + +1. ROM、応答依存設定、ANT0〜ANT3の物理接続を再取得する。 +2. アンテナID出力がON、EPCバッファリングがOFF、ANT0〜2がすべて接続OKであることを確認する。 +3. 開始前のコマンドモードRAMアンテナ設定を読み取る。 +4. ANT0、ANT1、ANT2の順にRAM設定を切り替え、毎回読み戻す。 +5. 各ANTでInventoryを1回だけ実行する。 +6. タグ応答が1件以上のANTだけ、応答要求ありの「ピッピッピ」ブザーを1回送信しACKを確認する。 +7. 開始前のアンテナ設定へ戻し、読み戻す。 + +画面とJSONへ残すのは、ANT別のタグ応答数、ユニークタグ数、完了ACK枚数、実使用CH、ブザーACK、復元結果だけです。PC/UII/EPC/TID、生フレームは残しません。タグが0件のANTではブザーを送信しません。NACK、timeout、応答不正、ANT番号不一致が出た場合は後続ANTへ進まず、`finally` で開始前のアンテナ設定へ復元します。 + +ブザーACKはリーダライタがコマンドを正常受理した確認です。実際に音が聞こえたかは自動判定できないため、ANT0、ANT1、ANT2ごとに人が確認し、ACK結果とは分けて記録します。 + +## 8. 開発者確認 + +```powershell +$env:PYTHONPATH = "." +py -m pytest -q +py -m compileall -q src tests +git diff --check +``` + +実機結果のJSONは `runtime_logs/` に保存し、Gitへコミットしません。 diff --git a/prompts/codex_usm02_ant012_inventory_buzzer.md b/prompts/codex_usm02_ant012_inventory_buzzer.md new file mode 100644 index 0000000..2680ba3 --- /dev/null +++ b/prompts/codex_usm02_ant012_inventory_buzzer.md @@ -0,0 +1,56 @@ +# Codex依頼文: USM02 ANT0〜2 Inventory・タグ検出時ブザー確認 + +`UTR_USB_Python_CodeX` の `feature/usm02-v117-validation-harness` ブランチで、コードを変更せず、ANT0・ANT1・ANT2のInventoryとタグ検出時ブザーだけを実行してください。 + +前提実機: + +- UTR-SUN02-4CH / USM02 Ver.2.052 +- COM6 / 115200 bps +- ANT0、ANT1、ANT2にタグを1枚ずつ配置済み +- ブザー音は「ピッピッピ」を使用 + +禁止: + +- Kill、Lock、UHF_Encode、FLASH設定初期化を送信しない +- FLASH、タグメモリ、Access Passwordを書き換えない +- 指定CLI以外から実機コマンドを追加送信しない +- 生Hex、PC/UII/EPC/TID、タグ固有値、製造番号、パスワードを回答へ載せない +- コード変更、commit、push、PR変更、mergeをしない + +手順: + +```powershell +cd C:\Users\tamaru\Documents\Codex\repos\UTR_USB_Python_CodeX +git fetch origin +git switch feature/usm02-v117-validation-harness +git pull --ff-only +git status --short +$env:PYTHONPATH = "." +py -m pytest -q +py -m src.utr_usm02_inventory_buzzer_cli --antennas 0,1,2 +py -m src.utr_usm02_inventory_buzzer_cli --execute --port COM6 --baudrate 115200 --antennas 0,1,2 --json-out runtime_logs\usm02_ant012_inventory_buzzer.json +git status --short +``` + +実行時の判断: + +- bootstrapでANT0、ANT1、ANT2のいずれかが接続OKでなければ、Inventoryを実行せず停止する。 +- アンテナID出力がONでなければ停止する。 +- EPCバッファリングがOFFでなければ停止する。 +- NACK、timeout、タグ応答ANT不一致、読戻し不一致、復元失敗が出たら、追加コマンドを送らず停止する。 +- タグ0件のANTは異常と決めつけず、ブザー未送信として記録する。 +- プログラムのブザーACKと、人が実際に音を聞いた確認を混同しない。 + +回答項目: + +1. pytest結果 +2. dry-run結果 +3. ROM、機種、接続OKアンテナ、アンテナID出力 +4. ANT0、ANT1、ANT2ごとのタグ応答数とユニークタグ数 +5. ANT0、ANT1、ANT2ごとの完了ACK枚数と実使用CH +6. ANT0、ANT1、ANT2ごとのブザー送信有無とブザーACK +7. ユーザーに、ANT0、ANT1、ANT2の「ピッピッピ」音が実際に聞こえたか確認する +8. 開始前アンテナ設定と復元結果 +9. RF送信はInventory最大3回、FLASH変更なし、タグ書込なし +10. NACK、timeout、例外、復元失敗の有無 +11. `git status --short` diff --git a/prompts/codex_usm02_phase2_safe_controls.md b/prompts/codex_usm02_phase2_safe_controls.md new file mode 100644 index 0000000..85eb1e6 --- /dev/null +++ b/prompts/codex_usm02_phase2_safe_controls.md @@ -0,0 +1,49 @@ +# Codex依頼文: USM02 Phase 2安全制御確認 + +`UTR_USB_Python_CodeX` の `feature/usm02-v117-validation-harness` ブランチで、コードを変更せずPhase 2だけを実行してください。 + +前提実機: + +- UTR-SUN02-4CH / USM02 Ver.2.052 +- COM6 / 115200 bps +- 設定アンテナ: ANT0 +- 接続OK: ANT0、ANT1 +- bootstrapでNACK、timeout、例外なし + +禁止: + +- Kill、Lock、UHF_Encode、FLASH設定初期化を送信しない +- FLASH、タグメモリ、Access Passwordを書き換えない +- RF送信はPhase 2 CLI内のInventory 1回だけに限定する +- 生Hex、タグ固有値、製造番号、パスワードを回答へ載せない +- コード変更、commit、push、PR変更、mergeをしない + +手順: + +```powershell +cd C:\Users\tamaru\Documents\Codex\repos\UTR_USB_Python_CodeX +git fetch origin +git switch feature/usm02-v117-validation-harness +git pull --ff-only +git status --short +$env:PYTHONPATH = "." +py -m pytest -q +py -m src.utr_usm02_safe_controls_cli --target-antenna 1 --verify-rf-channel +py -m src.utr_usm02_safe_controls_cli --execute --port COM6 --baudrate 115200 --target-antenna 1 --verify-rf-channel --json-out runtime_logs\usm02_phase2_safe_controls.json +git status --short +``` + +NACK、timeout、読戻し不一致、復元失敗、対象条件不一致が出たら、追加コマンドを送らず停止してください。 + +回答項目: + +1. pytest結果 +2. dry-run結果 +3. ブザーACKの成否（音が聞こえたかは推測せず、ユーザー確認事項として分ける） +4. アンテナの変更前→一時変更→復元と、各読戻し結果 +5. 開始CHの変更前→一時変更→復元 +6. 使用許可CHが変更されていないこと +7. Inventoryのタグ応答数、タグ応答ANT、完了ACKの実使用CH（タグ固有値は記載しない） +8. RF送信はInventory 1回だけ、FLASH変更なし、タグ書込なし +9. NACK、timeout、例外、復元失敗の有無 +10. `git status --short` diff --git a/prompts/codex_usm02_v117_bootstrap.md b/prompts/codex_usm02_v117_bootstrap.md new file mode 100644 index 0000000..18596a1 --- /dev/null +++ b/prompts/codex_usm02_v117_bootstrap.md @@ -0,0 +1,49 @@ +# Codex依頼文: USM02事前確認だけを実行 + +対象リポジトリで、既存コードを変更せず、以下だけを実行してください。 + +対象: + +- UTR-SUN02-4CH / ROMシリーズUSM02 +- USB COM6 +- 115200 bps + +禁止: + +- Kill、Lock、UHF_Encode、FLASH設定初期化を送信しない +- タグメモリを書き換えない +- FLASHを書き換えない +- PC/UII/EPC/TID、シリアル番号、パスワードの実値を回答やGitへ載せない +- merge、commit、push、PR作成をしない + +手順: + +```powershell +cd C:\Users\tamaru\Documents\Codex\repos\UTR_USB_Python_CodeX +git fetch origin +git switch feature/usm02-v117-validation-harness +git pull --ff-only +git status --short +git branch --show-current +$env:PYTHONPATH = "." +py -m pytest -q +py -m src.utr_usm02_v117_validation_cli --rom-number 2052 +py -m src.utr_usm02_v117_validation_cli --execute-bootstrap --port COM6 --baudrate 115200 --json-out runtime_logs\usm02_v117_bootstrap.json +git status --short +``` + +予期しないNACK、timeout、対象機種不一致、例外が1件でも出たら、その時点で停止してください。 + +回答は次の項目だけにしてください。 + +1. pytest結果 +2. dry-runの54件集計 +3. 実機ROM（シリーズ名USM02とバージョンだけ。製造番号は記載しない） +4. 物理アンテナ容量 +5. 設定アンテナ番号 +6. 接続OKアンテナ番号 +7. アンテナID出力、TID付加、EPCバッファリング、読取サイクル完了応答、アンテナ切替完了応答、キャリア検知応答のON/OFF +8. NACK、timeout、例外の有無 +9. `git status --short`（`runtime_logs/`以外の変更がないこと） + +生の送受信Hexとタグ固有値は回答へ貼らないでください。 diff --git a/src/utr_commands.py b/src/utr_commands.py index 03825c5..ec0e292 100644 --- a/src/utr_commands.py +++ b/src/utr_commands.py @@ -56,6 +56,10 @@ # 出力設定のサブコマンドです。 OUTPUT_SETTING_PARAMETER = 0x01 +FREQUENCY_SETTING_PARAMETER = 0x02 + +# Ver.1.17 7.4.7/7.4.22で指定可能なチャンネルです。 +VALID_FREQUENCY_CHANNELS = (5, 11, 17, *range(23, 38)) # 初期実装で許可する出力設定書き込み先です。 # FLASHデータは、将来の明示設計までは生成対象外にします。 @@ -184,6 +188,62 @@ def build_read_antenna_switching_setting_command(parameter_kind: int = PARAMETER return build_frame(0x55, bytes([DETAIL_READER_SETTING_READ, 0x00, parameter_kind])) +def build_read_frequency_setting_command(parameter_kind: int = PARAMETER_KIND_COMMAND_MODE) -> bytes: + """周波数設定（7.4.7）の読み取りコマンドを生成します。""" + _validate_byte_value(parameter_kind, "parameter_kind") + return build_frame( + 0x55, + bytes([DETAIL_READER_SETTING_READ, FREQUENCY_SETTING_PARAMETER, parameter_kind]), + ) + + +def _encode_frequency_channel_mask(enabled_channels: tuple[int, ...]) -> bytes: + if not enabled_channels: + raise ValueError("enabled_channels must not be empty") + unknown = set(enabled_channels) - set(VALID_FREQUENCY_CHANNELS) + if unknown: + raise ValueError(f"unsupported frequency channels: {sorted(unknown)}") + mask = bytearray(3) + for channel in set(enabled_channels): + index = VALID_FREQUENCY_CHANNELS.index(channel) + mask[index // 8] |= 1 << (index % 8) + return bytes(mask) + + +def build_write_frequency_setting_command( + parameter_kind: int, + starting_channel: int, + enabled_channels: tuple[int, ...], + reserved: bytes = b"\x00\x00\x00\x00", +) -> bytes: + """コマンドモードRAM用の周波数設定（7.4.22）を生成します。 + + FLASHや自動読み取りモードへの誤書き込みを防ぐため、書き込み先は + コマンドモード用パラメータ（00h）だけを許可します。 + """ + if parameter_kind != PARAMETER_KIND_COMMAND_MODE: + raise ValueError("frequency test only allows command-mode RAM parameters") + if starting_channel not in VALID_FREQUENCY_CHANNELS: + raise ValueError(f"unsupported starting channel: {starting_channel}") + if starting_channel not in enabled_channels: + raise ValueError("starting_channel must be included in enabled_channels") + if len(reserved) != 4: + raise ValueError("reserved must be exactly 4 bytes") + channel_mask = _encode_frequency_channel_mask(enabled_channels) + data = ( + bytes([ + DETAIL_READER_SETTING_WRITE, + FREQUENCY_SETTING_PARAMETER, + parameter_kind, + starting_channel, + 0x00, + ]) + + channel_mask + + reserved + ) + return build_frame(0x55, data) + + def build_check_antenna_command(antenna_number: int) -> bytes: """UHF_CheckAntennaコマンドを生成します。 diff --git a/src/utr_device_profile.py b/src/utr_device_profile.py new file mode 100644 index 0000000..ddb36d3 --- /dev/null +++ b/src/utr_device_profile.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""実機応答の解釈に必要なリーダライタ状態のスナップショット。""" + +from __future__ import annotations + +from dataclasses import dataclass, field + +from src.utr_antenna import AntennaSwitchingSetting, RomVersionInfo + + +@dataclass(frozen=True) +class DeviceProfile: + """ROM、機種、設定値、物理アンテナ状態を混同せず保持する。""" + + model_key: str + rom: RomVersionInfo + reader_id: int = 0 + antenna_capacity: int = 4 + configured_antennas: tuple[int, ...] = () + connected_antennas: tuple[int, ...] = () + antenna_id_output_enabled: bool = False + inventory_tid_enabled: bool | None = None + epc_buffering_enabled: bool | None = None + read_cycle_completion_enabled: bool | None = None + antenna_switch_completion_enabled: bool | None = None + carrier_detect_response_enabled: bool | None = None + source_notes: tuple[str, ...] = field(default_factory=tuple) + + @property + def rom_number(self) -> int: + """例: 2.052 -> 2052、2.100 -> 2100。""" + try: + return int(self.rom.major_version) * 1000 + int(self.rom.minor_version) + except ValueError as exc: + raise ValueError(f"ROMバージョンを数値化できません: {self.rom.firmware_version}") from exc + + def validate(self) -> None: + if self.model_key != "UTR-SUN02-4CH": + raise ValueError(f"USM02検証対象外の機種です: {self.model_key}") + if self.rom.series_name != "USM02": + raise ValueError(f"USM02検証対象外のROMシリーズです: {self.rom.series_name}") + if self.antenna_capacity != 4: + raise ValueError("UTR-SUN02-4CHの物理アンテナ容量は4である必要があります") + valid = set(range(self.antenna_capacity)) + if not set(self.configured_antennas).issubset(valid): + raise ValueError("設定アンテナ番号が物理容量外です") + if not set(self.connected_antennas).issubset(valid): + raise ValueError("接続アンテナ番号が物理容量外です") + + +def build_usm02_device_profile( + rom: RomVersionInfo, + command_mode_antenna_setting: AntennaSwitchingSetting, + connected_antennas: list[int], + *, + reader_id: int = 0, + inventory_tid_enabled: bool | None = None, + epc_buffering_enabled: bool | None = None, + read_cycle_completion_enabled: bool | None = None, + antenna_switch_completion_enabled: bool | None = None, + carrier_detect_response_enabled: bool | None = None, +) -> DeviceProfile: + """既存のROM/アンテナ解析結果からUSM02用プロファイルを作る。""" + profile = DeviceProfile( + model_key="UTR-SUN02-4CH", + rom=rom, + reader_id=reader_id, + antenna_capacity=4, + configured_antennas=tuple(command_mode_antenna_setting.enabled_antennas), + connected_antennas=tuple(sorted(set(connected_antennas))), + antenna_id_output_enabled=command_mode_antenna_setting.antenna_id_output_enabled, + inventory_tid_enabled=inventory_tid_enabled, + epc_buffering_enabled=epc_buffering_enabled, + read_cycle_completion_enabled=read_cycle_completion_enabled, + antenna_switch_completion_enabled=antenna_switch_completion_enabled, + carrier_detect_response_enabled=carrier_detect_response_enabled, + source_notes=( + "antenna_capacity=機種仕様", + "configured_antennas=7.4.5 アンテナ切替設定", + "connected_antennas=7.3.5 UHF_CheckAntenna", + ), + ) + profile.validate() + return profile diff --git a/src/utr_reader_settings.py b/src/utr_reader_settings.py index 7b0e4e5..b53e188 100644 --- a/src/utr_reader_settings.py +++ b/src/utr_reader_settings.py @@ -5,6 +5,7 @@ from __future__ import annotations +from dataclasses import dataclass from typing import Any try: @@ -33,6 +34,37 @@ 922.4, 922.6, 922.8, 923.0, 923.2, 923.4, ] +VALID_FREQUENCY_CHANNELS = (5, 11, 17, *range(23, 38)) + + +@dataclass(frozen=True) +class FrequencySetting: + parameter_kind: int + starting_channel: int + current_channel: int + enabled_channels: tuple[int, ...] + channel_mask: bytes + reserved: bytes + raw: bytes + + +def channel_to_frequency_mhz(channel_number: int) -> float | None: + if 1 <= channel_number <= len(OUTPUT_CH_FREQ_LIST): + return OUTPUT_CH_FREQ_LIST[channel_number - 1] + return None + + +def _decode_frequency_channel_mask(mask: bytes) -> tuple[int, ...]: + if len(mask) != 3: + raise ValueError("frequency channel mask must be exactly 3 bytes") + if mask[2] & 0xFC: + raise ValueError("frequency channel mask reserved bits must be zero") + return tuple( + channel + for index, channel in enumerate(VALID_FREQUENCY_CHANNELS) + if mask[index // 8] & (1 << (index % 8)) + ) + def _validate_reader_setting_ack(frame: bytes, parameter_kind: int, min_data_length: int) -> bytes: if len(frame) < 7: @@ -74,7 +106,7 @@ def parse_output_power_setting_response(frame: bytes) -> dict[str, Any]: timing_settings = parse_output_power_timing_settings_from_data(data) return { "detail_command": data[0], - "parameter_kind": data[1], + "parameter_kind": data[2], "raw_data_hex": data.hex(" ").upper(), "raw_value_hex": output_power_bytes.hex(" ").upper(), "output_power_dbm": parse_output_power_dbm(output_power_bytes), @@ -83,23 +115,45 @@ def parse_output_power_setting_response(frame: bytes) -> dict[str, Any]: def parse_frequency_setting_response(frame: bytes) -> dict[str, Any]: - """Parse UHF_READ_FREQ_CH response into display-friendly values.""" + """7.4.7の12バイトACKを開始CH・現在CH・使用CHへ分解します。""" data = _validate_reader_setting_ack( frame, parameter_kind=PARAMETER_KIND_FREQUENCY_CHANNEL, - min_data_length=4, + min_data_length=12, ) - channel_number = data[3] - frequency_mhz = None - if 1 <= channel_number <= len(OUTPUT_CH_FREQ_LIST): - frequency_mhz = OUTPUT_CH_FREQ_LIST[channel_number - 1] + if len(data) != 12: + raise ValueError("周波数設定レスポンスのDATA長は12バイトである必要があります。") + setting = FrequencySetting( + parameter_kind=data[2], + starting_channel=data[3], + current_channel=data[4], + enabled_channels=_decode_frequency_channel_mask(data[5:8]), + channel_mask=data[5:8], + reserved=data[8:12], + raw=data, + ) + if setting.starting_channel not in VALID_FREQUENCY_CHANNELS: + raise ValueError("周波数の開始チャンネル番号が仕様範囲外です。") + if setting.starting_channel not in setting.enabled_channels: + raise ValueError("周波数の開始チャンネルが使用許可されていません。") + if setting.current_channel not in VALID_FREQUENCY_CHANNELS: + raise ValueError("現在設定されているチャンネル番号が仕様範囲外です。") return { "detail_command": data[0], - "parameter_kind": data[1], + "parameter_kind": data[2], "raw_data_hex": data.hex(" ").upper(), - "channel_number": channel_number, - "frequency_mhz": frequency_mhz, + # 後方互換キー。現在設定CHを表します。 + "channel_number": setting.current_channel, + "frequency_mhz": channel_to_frequency_mhz(setting.current_channel), + "starting_channel_number": setting.starting_channel, + "starting_frequency_mhz": channel_to_frequency_mhz(setting.starting_channel), + "current_channel_number": setting.current_channel, + "current_frequency_mhz": channel_to_frequency_mhz(setting.current_channel), + "enabled_channels": setting.enabled_channels, + "channel_mask": setting.channel_mask, + "reserved": setting.reserved, + "setting": setting, } @@ -115,11 +169,13 @@ def format_output_power_setting(parsed: dict[str, Any]) -> list[str]: def format_frequency_setting(parsed: dict[str, Any]) -> list[str]: """Format parsed frequency setting for display.""" lines = [ - f"チャンネル番号: {parsed['channel_number']} ch", + f"開始チャンネル番号: {parsed['starting_channel_number']} ch", + f"現在チャンネル番号: {parsed['current_channel_number']} ch", + f"使用チャンネル: {list(parsed['enabled_channels'])}", ] - if parsed["frequency_mhz"] is not None: - lines.append(f"送信周波数: {parsed['frequency_mhz']} MHz") + if parsed["current_frequency_mhz"] is not None: + lines.append(f"現在チャンネル周波数: {parsed['current_frequency_mhz']} MHz") else: - lines.append("送信周波数: チャンネル番号が既知範囲外のため換算できません") + lines.append("現在チャンネル周波数: チャンネル番号が既知範囲外のため換算できません") lines.append(f"周波数設定Raw: {parsed['raw_data_hex']}") return lines diff --git a/src/utr_response_settings.py b/src/utr_response_settings.py new file mode 100644 index 0000000..0c9d7c6 --- /dev/null +++ b/src/utr_response_settings.py @@ -0,0 +1,38 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""応答の件数・順序・2バイト目へ影響する設定値の解析。""" + +from __future__ import annotations + +from dataclasses import dataclass + +from src.utr_response_v117 import FrameKind, parse_frame + + +@dataclass(frozen=True) +class EpcUiiResponseSettings: + parameter_kind: int + epc_buffering_enabled: bool + read_cycle_completion_enabled: bool + antenna_switch_completion_enabled: bool + carrier_detect_response_enabled: bool + + +def parse_epc_uii_response_settings(frame: bytes) -> EpcUiiResponseSettings: + """7.4.9の4バイトDATAを解析する。""" + parsed = parse_frame(frame) + if parsed.kind != FrameKind.ACK: + raise ValueError("EPC(UII)関連パラメータ応答がACKではありません") + if len(parsed.data) != 4 or parsed.data[:2] != b"\x43\x05": + raise ValueError("EPC(UII)関連パラメータ応答の識別子またはDATA長が不正です") + flags = parsed.data[3] + if flags & 0xF0: + raise ValueError("EPC(UII)関連パラメータの予約ビットが0ではありません") + return EpcUiiResponseSettings( + parameter_kind=parsed.data[2], + epc_buffering_enabled=bool(flags & 0x01), + read_cycle_completion_enabled=bool(flags & 0x02), + antenna_switch_completion_enabled=bool(flags & 0x04), + carrier_detect_response_enabled=bool(flags & 0x08), + ) diff --git a/src/utr_response_v117.py b/src/utr_response_v117.py new file mode 100644 index 0000000..12f8c71 --- /dev/null +++ b/src/utr_response_v117.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""Ver.1.17のACK/NACK/タグ応答を設定依存で検証する純粋ロジック。""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Iterable + +from src.utr_device_profile import DeviceProfile +from src.utr_protocol import calculate_sum_value, parse_nack_frame +from src.utr_v117_catalog import CommandSpec, ResponsePattern + + +class FrameKind(str, Enum): + ACK = "ACK" + NACK = "NACK" + TAG_DATA = "TAG_DATA" + OTHER = "OTHER" + + +class AddressRole(str, Enum): + READER_ID = "reader_id" + ANTENNA_NUMBER = "antenna_number" + + +class VerificationResult(str, Enum): + ACK_VERIFIED = "ACK_VERIFIED" + NACK_OBSERVED = "NACK_OBSERVED" + MULTI_RESPONSE_VERIFIED = "MULTI_RESPONSE_VERIFIED" + NO_RESPONSE_VERIFIED = "NO_RESPONSE_VERIFIED" + CONDITIONAL_NO_RESPONSE_VERIFIED = "CONDITIONAL_NO_RESPONSE_VERIFIED" + FAILED_NEEDS_ANALYSIS = "FAILED_NEEDS_ANALYSIS" + + +@dataclass(frozen=True) +class ParsedFrame: + raw: bytes + address: int + command: int + data: bytes + + @property + def kind(self) -> FrameKind: + return { + 0x30: FrameKind.ACK, + 0x31: FrameKind.NACK, + 0x6C: FrameKind.TAG_DATA, + }.get(self.command, FrameKind.OTHER) + + @property + def detail(self) -> int | None: + return self.data[0] if self.data else None + + +@dataclass(frozen=True) +class AddressInterpretation: + value: int + role: AddressRole + label: str + + +@dataclass(frozen=True) +class CommandVerification: + result: VerificationResult + frames: tuple[ParsedFrame, ...] + notes: tuple[str, ...] + nack: dict[str, object] | None = None + + +def parse_frame(raw: bytes) -> ParsedFrame: + """1フレームを構造・SUMまで検証して解析する。""" + if len(raw) < 7: + raise ValueError("フレームが短すぎます") + if raw[0] != 0x02 or raw[-1] != 0x0D: + raise ValueError("STXまたはCRが一致しません") + data_length = raw[3] + if len(raw) != data_length + 7: + raise ValueError("DATA長とフレーム長が一致しません") + if raw[4 + data_length] != 0x03: + raise ValueError("ETX位置が一致しません") + if calculate_sum_value(raw[:-2]) != raw[-2]: + raise ValueError("SUMが一致しません") + return ParsedFrame(raw=raw, address=raw[1], command=raw[2], data=raw[4:4 + data_length]) + + +def split_frames(received: bytes) -> tuple[ParsedFrame, ...]: + """連結された応答を完全なフレーム列へ分割する。""" + frames: list[ParsedFrame] = [] + index = 0 + while index < len(received): + if received[index] != 0x02: + raise ValueError(f"STX以外の受信データがあります: offset={index}") + if len(received) - index < 4: + raise ValueError("末尾に不完全なフレームがあります") + frame_length = received[index + 3] + 7 + end = index + frame_length + if end > len(received): + raise ValueError("末尾に不完全なフレームがあります") + frames.append(parse_frame(received[index:end])) + index = end + return tuple(frames) + + +def interpret_address(frame: ParsedFrame, profile: DeviceProfile) -> AddressInterpretation: + """2バイト目を、応答種類とアンテナID出力設定に基づき解釈する。""" + if frame.kind == FrameKind.TAG_DATA and profile.antenna_id_output_enabled: + label = f"ANT{frame.address}" + return AddressInterpretation(frame.address, AddressRole.ANTENNA_NUMBER, label) + return AddressInterpretation(frame.address, AddressRole.READER_ID, f"Reader ID {frame.address}") + + +def response_context_notes(profile: DeviceProfile) -> tuple[str, ...]: + """応答数・順序・長さへ影響する設定を明示する。""" + def on_off_unknown(value: bool | None) -> str: + if value is None: + return "未読取" + return "ON" if value else "OFF" + + return ( + f"物理アンテナ容量={profile.antenna_capacity}", + f"設定アンテナ={list(profile.configured_antennas)}", + f"接続OKアンテナ={list(profile.connected_antennas)}", + f"アンテナID出力={on_off_unknown(profile.antenna_id_output_enabled)}", + f"Inventory TID付加={on_off_unknown(profile.inventory_tid_enabled)}", + f"EPCバッファリング={on_off_unknown(profile.epc_buffering_enabled)}", + f"読取サイクル完了応答={on_off_unknown(profile.read_cycle_completion_enabled)}", + f"アンテナ切替完了応答={on_off_unknown(profile.antenna_switch_completion_enabled)}", + f"キャリア検知応答={on_off_unknown(profile.carrier_detect_response_enabled)}", + ) + + +def _length_matches(spec: CommandSpec, data: bytes) -> bool: + """仕様書の固定長・可変長規則を検査する。""" + length = len(data) + fixed: dict[str, int] = { + "7.3.1": 0x04, "7.3.3": 0x01, "7.3.4": 0x02, "7.3.5": 0x03, + "7.3.6": 0x04, "7.3.7": 0x04, "7.3.8": 0x0A, "7.3.11": 0x01, + "7.3.12": 0x03, "7.4.1": 0x09, "7.4.3": 0x0B, "7.4.5": 0x08, + "7.4.6": 0x0B, "7.4.7": 0x0C, "7.4.8": 0x06, "7.4.9": 0x04, + "7.4.10": 0x0C, "7.4.11": 0x05, "7.4.12": 0x05, "7.4.13": 0x02, + "7.4.16": 0x00, "7.4.17": 0x01, "7.4.18": 0x01, "7.4.19": 0x01, + "7.4.20": 0x08, "7.4.21": 0x0B, "7.4.22": 0x0C, "7.4.23": 0x03, + "7.4.24": 0x03, "7.4.25": 0x04, "7.4.26": 0x0C, "7.4.27": 0x01, + "7.4.28": 0x01, "7.4.29": 0x01, "7.4.30": 0x05, "7.4.31": 0x05, + "7.5.4": 0x01, "7.5.5": 0x01, "7.5.6": 0x01, "7.5.7": 0x01, + "7.5.8": 0x01, "7.5.9": 0x01, "7.5.10": 0x01, + } + if spec.section in fixed: + return length == fixed[spec.section] + if spec.section == "7.3.2": + return length == 0 + if spec.section == "7.3.9": + return length in (0x0B, 0x0C) + if spec.section == "7.4.2": + return length >= 9 + if spec.section == "7.4.4": + return length >= 2 and (length - 2) % 23 == 0 + if spec.section == "7.4.14": + return length >= 5 and (length - 5) % 12 == 0 + if spec.section == "7.4.15": + return length >= 5 + if spec.section == "7.5.3": + return length >= 2 and length == data[1] + 2 + if spec.section == "7.5.11": + return 1 <= length <= 255 + return True + + +def _detail_matches(spec: CommandSpec, frame: ParsedFrame) -> bool: + if spec.section in ("7.3.2", "7.4.16"): + return True + if spec.response_detail is not None and frame.detail != spec.response_detail: + return False + if spec.subcommand is not None: + return len(frame.data) >= 2 and frame.data[1] == spec.subcommand + return True + + +def _first_nack(frames: Iterable[ParsedFrame]) -> ParsedFrame | None: + return next((frame for frame in frames if frame.kind == FrameKind.NACK), None) + + +def verify_command_response( + spec: CommandSpec, + received: bytes, + profile: DeviceProfile, + *, + response_requested: bool = True, +) -> CommandVerification: + """1コマンドの観測応答を仕様・設定と照合する。""" + notes = list(response_context_notes(profile)) + if not received: + if spec.response_pattern == ResponsePattern.NO_RESPONSE: + return CommandVerification(VerificationResult.NO_RESPONSE_VERIFIED, (), tuple(notes)) + if spec.response_pattern == ResponsePattern.CONDITIONAL_ACK and not response_requested: + return CommandVerification(VerificationResult.CONDITIONAL_NO_RESPONSE_VERIFIED, (), tuple(notes)) + notes.append("期待した応答を受信できませんでした") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, (), tuple(notes)) + + try: + frames = split_frames(received) + except ValueError as exc: + notes.append(str(exc)) + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, (), tuple(notes)) + + nack_frame = _first_nack(frames) + if nack_frame is not None: + parsed_nack = parse_nack_frame(nack_frame.raw) + notes.append("NACKは応答として解析済みだが、正常完了とは判定しません") + return CommandVerification(VerificationResult.NACK_OBSERVED, frames, tuple(notes), parsed_nack) + + if spec.response_pattern == ResponsePattern.NO_RESPONSE: + notes.append("応答なし仕様のコマンドで応答を受信しました") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + + if spec.response_pattern == ResponsePattern.MULTI_TAG_THEN_ACK: + completion = frames[-1] + expected_detail = spec.detail + tag_frames = frames[:-1] + if completion.kind != FrameKind.ACK or completion.detail != expected_detail or len(completion.data) != 5: + notes.append("最終フレームが仕様どおりの完了ACKではありません") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + for frame in tag_frames: + if frame.kind != FrameKind.TAG_DATA or frame.detail != expected_detail: + notes.append("完了ACKより前に想定外のフレームがあります") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + address = interpret_address(frame, profile) + if address.role == AddressRole.ANTENNA_NUMBER and address.value >= profile.antenna_capacity: + notes.append(f"物理容量外のアンテナ番号です: {address.value}") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + notes.append(f"タグ応答={len(tag_frames)}件、完了ACK=1件") + return CommandVerification(VerificationResult.MULTI_RESPONSE_VERIFIED, frames, tuple(notes)) + + if len(frames) != 1 or frames[0].kind != FrameKind.ACK: + notes.append("単一ACKを期待しましたが応答形態が一致しません") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + ack = frames[0] + if not _detail_matches(spec, ack): + notes.append("ACKの詳細コマンドが送信コマンドと一致しません") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + if not _length_matches(spec, ack.data): + notes.append(f"ACKのDATA長が仕様と一致しません: {len(ack.data)}") + return CommandVerification(VerificationResult.FAILED_NEEDS_ANALYSIS, frames, tuple(notes)) + return CommandVerification(VerificationResult.ACK_VERIFIED, frames, tuple(notes)) diff --git a/src/utr_usm02_inventory_buzzer.py b/src/utr_usm02_inventory_buzzer.py new file mode 100644 index 0000000..5c76f06 --- /dev/null +++ b/src/utr_usm02_inventory_buzzer.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""USM02のアンテナ別Inventoryとタグ検出時ブザー確認。""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass + +from src.utr_commands import BUZZER_SOUND_PIPIPI, build_buzzer_command, build_frame +from src.utr_device_profile import DeviceProfile +from src.utr_response_v117 import AddressRole, VerificationResult, interpret_address, verify_command_response +from src.utr_usm02_safe_controls import ( + Exchange, + clear_before_restore, + read_command_mode_antenna_setting, + verify_ack, + write_command_mode_antenna_setting, +) +from src.utr_v117_catalog import get_command_spec + + +@dataclass(frozen=True) +class AntennaInventoryBuzzerResult: + antenna: int + tag_detected: bool + tag_response_count: int + unique_tag_count: int + completion_reported_count: int + response_count_matches_completion: bool + observed_channel: int + buzzer_command_sent: bool + buzzer_ack_verified: bool + + +@dataclass(frozen=True) +class SequentialInventoryBuzzerResult: + requested_antennas: tuple[int, ...] + original_antennas: tuple[int, ...] + antenna_results: tuple[AntennaInventoryBuzzerResult, ...] + antenna_restored: bool + inventory_execution_count: int + buzzer_ack_count: int + flash_write_executed: bool = False + tag_memory_write_executed: bool = False + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +def normalize_target_antennas( + antennas: tuple[int, ...] | list[int], + antenna_capacity: int, +) -> tuple[int, ...]: + """指定順を保持したまま、重複・物理容量外を拒否する。""" + normalized = tuple(antennas) + if not normalized: + raise ValueError("対象アンテナを1件以上指定してください") + if len(set(normalized)) != len(normalized): + raise ValueError("対象アンテナに重複があります") + invalid = tuple(value for value in normalized if value < 0 or value >= antenna_capacity) + if invalid: + raise ValueError(f"物理容量外のアンテナ番号です: {list(invalid)}") + return normalized + + +def _tag_identity(frame_data: bytes) -> bytes: + """タグ固有値を外部へ返さず、メモリ上の重複判定キーだけを切り出す。""" + if len(frame_data) < 5: + raise RuntimeError("Inventoryタグ応答のDATA長が不足しています") + pc_uii_length = frame_data[4] + if pc_uii_length == 0: + raise RuntimeError("Inventoryタグ応答のPC+UII長が0です") + end = 5 + pc_uii_length + if len(frame_data) < end: + raise RuntimeError("Inventoryタグ応答のPC+UII長とDATA長が一致しません") + return frame_data[5:end] + + +def _inventory_once( + ser, + exchange: Exchange, + profile: DeviceProfile, + target_antenna: int, +) -> tuple[int, int, int, bool, int]: + response = exchange(ser, build_frame(0x55, b"\x10")) + verification = verify_command_response(get_command_spec("7.5.1"), response, profile) + if verification.result != VerificationResult.MULTI_RESPONSE_VERIFIED: + raise RuntimeError( + "7.5.1 応答異常: " + f"{verification.result.value}; {' / '.join(verification.notes)}" + ) + + tag_frames = verification.frames[:-1] + completion = verification.frames[-1] + unique_tags: set[bytes] = set() + for frame in tag_frames: + address = interpret_address(frame, profile) + if address.role != AddressRole.ANTENNA_NUMBER: + raise RuntimeError("タグ応答2バイト目をアンテナ番号として確定できません") + if address.value != target_antenna: + raise RuntimeError( + f"タグ応答アンテナが選択中のANT{target_antenna}と一致しません: " + f"ANT{address.value}" + ) + unique_tags.add(_tag_identity(frame.data)) + + completion_count = int.from_bytes(completion.data[2:4], byteorder="little") + observed_channel = completion.data[4] + tag_response_count = len(tag_frames) + return ( + tag_response_count, + len(unique_tags), + completion_count, + tag_response_count == completion_count, + observed_channel, + ) + + +def execute_sequential_inventory_buzzer( + ser, + profile: DeviceProfile, + exchange: Exchange, + *, + target_antennas: tuple[int, ...] = (0, 1, 2), +) -> SequentialInventoryBuzzerResult: + """ANTを順次選び、タグ応答があったANTだけブザーを鳴らして復元する。""" + profile.validate() + targets = normalize_target_antennas(target_antennas, profile.antenna_capacity) + missing = tuple(antenna for antenna in targets if antenna not in profile.connected_antennas) + if missing: + raise ValueError( + "UHF_CheckAntennaで接続OKではない対象があります: " + + ", ".join(f"ANT{value}" for value in missing) + ) + if profile.antenna_id_output_enabled is not True: + raise ValueError("アンテナID出力がONではないため、タグ応答ANTを判定できません") + if profile.epc_buffering_enabled is not False: + raise ValueError( + "EPCバッファリングがOFFと確認できないため、直接タグ応答を判定できません" + ) + + original = read_command_mode_antenna_setting(ser, exchange, profile) + if original.antenna_id_output_enabled is not True: + raise RuntimeError("実行直前のアンテナID出力がONではないため停止します") + if original.antenna_mask == 0: + raise RuntimeError("実行直前の使用アンテナ設定が0件のため停止します") + + current_mask = original.antenna_mask + write_attempted = False + results: list[AntennaInventoryBuzzerResult] = [] + primary_error: Exception | None = None + restore_error: Exception | None = None + restored = False + + try: + for antenna in targets: + target_mask = 1 << antenna + if current_mask != target_mask: + # ACK後の読戻しで失敗しても実機設定は変化した可能性があるため、 + # 送信前に復元対象として記録する。 + write_attempted = True + write_command_mode_antenna_setting( + ser, + exchange, + profile, + original, + target_mask, + ) + current_mask = target_mask + + ( + tag_response_count, + unique_tag_count, + completion_count, + count_matches, + observed_channel, + ) = _inventory_once(ser, exchange, profile, antenna) + tag_detected = tag_response_count > 0 + buzzer_verified = False + if tag_detected: + buzzer_response = exchange( + ser, + build_buzzer_command( + response_required=True, + sound_type=BUZZER_SOUND_PIPIPI, + ), + ) + verify_ack("7.3.2", buzzer_response, profile) + buzzer_verified = True + + results.append( + AntennaInventoryBuzzerResult( + antenna=antenna, + tag_detected=tag_detected, + tag_response_count=tag_response_count, + unique_tag_count=unique_tag_count, + completion_reported_count=completion_count, + response_count_matches_completion=count_matches, + observed_channel=observed_channel, + buzzer_command_sent=tag_detected, + buzzer_ack_verified=buzzer_verified, + ) + ) + except Exception as exc: + primary_error = exc + finally: + if write_attempted: + try: + clear_before_restore(ser) + readback = write_command_mode_antenna_setting( + ser, + exchange, + profile, + original, + original.antenna_mask, + ) + restored = readback.antenna_mask == original.antenna_mask + if not restored: + raise RuntimeError("アンテナ復元後の読戻しが開始値と一致しません") + except Exception as exc: + restore_error = exc + else: + restored = True + + if restore_error is not None: + suffix = "" if primary_error is None else f" / 実行時エラー: {primary_error}" + raise RuntimeError(f"アンテナ復元失敗: {restore_error}{suffix}") + if primary_error is not None: + raise primary_error + + return SequentialInventoryBuzzerResult( + requested_antennas=targets, + original_antennas=tuple(original.enabled_antennas), + antenna_results=tuple(results), + antenna_restored=restored, + inventory_execution_count=len(results), + buzzer_ack_count=sum(result.buzzer_ack_verified for result in results), + ) diff --git a/src/utr_usm02_inventory_buzzer_cli.py b/src/utr_usm02_inventory_buzzer_cli.py new file mode 100644 index 0000000..41d0f60 --- /dev/null +++ b/src/utr_usm02_inventory_buzzer_cli.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""USM02 ANT0〜2 Inventory・タグ検出時ブザー確認CLI。""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from src.utr_usm02_inventory_buzzer import execute_sequential_inventory_buzzer +from src.utr_usm02_v117_validation_cli import collect_bootstrap_profile + + +def parse_antenna_list(value: str) -> tuple[int, ...]: + try: + antennas = tuple(int(part.strip()) for part in value.split(",") if part.strip()) + except ValueError as exc: + raise argparse.ArgumentTypeError("アンテナは 0,1,2 の形式で指定してください") from exc + if not antennas: + raise argparse.ArgumentTypeError("アンテナを1件以上指定してください") + return antennas + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="USM02 アンテナ別Inventory・条件付きブザー確認") + parser.add_argument("--execute", action="store_true", help="実機送信を明示的に許可") + parser.add_argument("--port", default="COM6") + parser.add_argument("--baudrate", type=int, default=115200) + parser.add_argument("--antennas", type=parse_antenna_list, default=(0, 1, 2)) + parser.add_argument("--json-out", type=Path) + return parser.parse_args() + + +def main() -> int: + args = _parse_args() + labels = ", ".join(f"ANT{value}" for value in args.antennas) + print("=== USM02 アンテナ別Inventory・条件付きブザー確認 ===") + print("対象:", labels) + print("各ANTでInventoryを1回実行し、タグ応答が1件以上ならピッピッピ音を送信します。") + print("開始前にROM・設定・ANT0〜3の物理接続を再取得します。") + print("FLASH変更なし / タグ書込なし / タグ固有値の表示・保存なし") + + if not args.execute: + print("dry-run完了。実機送信には --execute が必要です。") + return 0 + + try: + import serial + + from src.utr_usb_sample_legacy import communicate + + with serial.Serial(port=args.port, baudrate=args.baudrate, timeout=0.05) as ser: + ser.reset_input_buffer() + ser.reset_output_buffer() + profile = collect_bootstrap_profile(ser, communicate) + result = execute_sequential_inventory_buzzer( + ser, + profile, + communicate, + target_antennas=args.antennas, + ) + except (RuntimeError, ValueError) as exc: + print("安全停止:", exc) + return 1 + + print("接続OKアンテナ:", list(profile.connected_antennas)) + for item in result.antenna_results: + print( + f"ANT{item.antenna}: タグ応答={item.tag_response_count}, " + f"ユニークタグ={item.unique_tag_count}, 完了ACK枚数={item.completion_reported_count}, " + f"件数照合={'OK' if item.response_count_matches_completion else '差異あり'}, " + f"CH={item.observed_channel}, ブザーACK={'OK' if item.buzzer_ack_verified else '対象外'}" + ) + print("アンテナ復元:", "OK" if result.antenna_restored else "NG") + print("実際に音が聞こえたかは、ANTごとにユーザー確認が必要です。") + + if args.json_out: + payload = { + "target": "UTR-SUN02-4CH / USM02", + "rom": profile.rom.firmware_version, + "port": args.port, + "baudrate": args.baudrate, + "audible_buzzer_confirmation": "USER_CONFIRMATION_REQUIRED", + "result": result.to_dict(), + } + args.json_out.parent.mkdir(parents=True, exist_ok=True) + args.json_out.write_text( + json.dumps(payload, ensure_ascii=False, indent=2) + "\n", + encoding="utf-8", + ) + print("JSON出力:", args.json_out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/utr_usm02_safe_controls.py b/src/utr_usm02_safe_controls.py new file mode 100644 index 0000000..2a97eb0 --- /dev/null +++ b/src/utr_usm02_safe_controls.py @@ -0,0 +1,323 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""USM02 Phase 2: ブザー・アンテナ・周波数の一時変更と復元。""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass +from typing import Callable + +from src.utr_antenna import ( + AntennaSwitchingSetting, + parse_antenna_switching_setting_response, + parse_antenna_switching_setting_write_response, +) +from src.utr_commands import ( + PARAMETER_KIND_COMMAND_MODE, + build_buzzer_command, + build_frame, + build_read_antenna_switching_setting_command, + build_read_frequency_setting_command, + build_write_antenna_switching_setting_command, + build_write_frequency_setting_command, +) +from src.utr_device_profile import DeviceProfile +from src.utr_reader_settings import FrequencySetting, parse_frequency_setting_response +from src.utr_response_v117 import ( + AddressRole, + VerificationResult, + interpret_address, + verify_command_response, +) +from src.utr_v117_catalog import get_command_spec + + +Exchange = Callable[[object, bytes], bytes] +PREFERRED_TEST_CHANNELS = tuple(range(26, 33)) + + +@dataclass(frozen=True) +class SafeControlsResult: + buzzer_ack_verified: bool + original_antenna: tuple[int, ...] + temporary_antenna: int + antenna_restored: bool + original_starting_channel: int + temporary_starting_channel: int + enabled_channels_unchanged: bool + frequency_restored: bool + rf_transmission_executed: bool = False + flash_write_executed: bool = False + inventory_tag_response_count: int = 0 + observed_rf_channel: int | None = None + observed_tag_antennas: tuple[int, ...] = () + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +def choose_temporary_channel(setting: FrequencySetting, requested: int | None = None) -> int: + """使用許可済みCHから、現在の開始CHとは異なる試験CHを選ぶ。""" + enabled = tuple(setting.enabled_channels) + if requested is not None: + if requested not in enabled: + raise ValueError(f"指定CH {requested} は現在の使用許可CHではありません") + if requested == setting.starting_channel: + raise ValueError("試験CHは現在の開始CHと異なる必要があります") + return requested + + preferred = [channel for channel in PREFERRED_TEST_CHANNELS if channel in enabled] + candidates = preferred or list(enabled) + for channel in candidates: + if channel != setting.starting_channel: + return channel + raise ValueError("現在の開始CH以外に、使用許可済みの試験CHがありません") + + +def verify_ack(section: str, response: bytes, profile: DeviceProfile) -> None: + verification = verify_command_response(get_command_spec(section), response, profile) + if verification.result != VerificationResult.ACK_VERIFIED: + raise RuntimeError( + f"{section} 応答異常: {verification.result.value}; {' / '.join(verification.notes)}" + ) + + +def read_command_mode_antenna_setting( + ser, + exchange: Exchange, + profile: DeviceProfile, +) -> AntennaSwitchingSetting: + response = exchange( + ser, + build_read_antenna_switching_setting_command(PARAMETER_KIND_COMMAND_MODE), + ) + verify_ack("7.4.5", response, profile) + setting = parse_antenna_switching_setting_response(response) + if setting.parameter_kind != PARAMETER_KIND_COMMAND_MODE: + raise RuntimeError("コマンドモード以外のアンテナ設定応答です") + return setting + + +def _read_frequency_setting(ser, exchange: Exchange, profile: DeviceProfile) -> FrequencySetting: + response = exchange( + ser, + build_read_frequency_setting_command(PARAMETER_KIND_COMMAND_MODE), + ) + verify_ack("7.4.7", response, profile) + setting = parse_frequency_setting_response(response)["setting"] + if setting.parameter_kind != PARAMETER_KIND_COMMAND_MODE: + raise RuntimeError("コマンドモード以外の周波数設定応答です") + return setting + + +def write_command_mode_antenna_setting( + ser, + exchange: Exchange, + profile: DeviceProfile, + original: AntennaSwitchingSetting, + antenna_mask: int, +) -> AntennaSwitchingSetting: + response = exchange( + ser, + build_write_antenna_switching_setting_command( + parameter_kind=PARAMETER_KIND_COMMAND_MODE, + switching_mode=original.switching_mode, + antenna_id_output_enabled=original.antenna_id_output_enabled, + antenna_mask=antenna_mask, + ), + ) + verify_ack("7.4.20", response, profile) + acknowledged = parse_antenna_switching_setting_write_response(response) + if acknowledged.antenna_mask != antenna_mask: + raise RuntimeError("アンテナ書込ACKの設定値が要求値と一致しません") + readback = read_command_mode_antenna_setting(ser, exchange, profile) + if ( + readback.antenna_mask != antenna_mask + or readback.switching_mode != original.switching_mode + or readback.antenna_id_output_enabled != original.antenna_id_output_enabled + ): + raise RuntimeError("アンテナ設定の読戻しが要求値と一致しません") + return readback + + +def _write_frequency( + ser, + exchange: Exchange, + profile: DeviceProfile, + original: FrequencySetting, + starting_channel: int, +) -> FrequencySetting: + response = exchange( + ser, + build_write_frequency_setting_command( + parameter_kind=PARAMETER_KIND_COMMAND_MODE, + starting_channel=starting_channel, + enabled_channels=original.enabled_channels, + reserved=original.reserved, + ), + ) + verify_ack("7.4.22", response, profile) + readback = _read_frequency_setting(ser, exchange, profile) + if readback.starting_channel != starting_channel: + raise RuntimeError("周波数設定の開始CH読戻しが要求値と一致しません") + if readback.enabled_channels != original.enabled_channels: + raise RuntimeError("周波数設定の使用許可CHが変更されています") + return readback + + +def clear_before_restore(ser) -> None: + reset = getattr(ser, "reset_input_buffer", None) + if callable(reset): + reset() + + +def execute_safe_controls( + ser, + profile: DeviceProfile, + exchange: Exchange, + *, + target_antenna: int = 1, + requested_channel: int | None = None, + verify_rf_channel: bool = False, +) -> SafeControlsResult: + """3操作を行い、例外時を含め周波数→アンテナの順に復元する。""" + if target_antenna not in profile.connected_antennas: + raise ValueError(f"ANT{target_antenna} はUHF_CheckAntennaで接続OKではありません") + + original_antenna = read_command_mode_antenna_setting(ser, exchange, profile) + original_frequency = _read_frequency_setting(ser, exchange, profile) + temporary_channel = choose_temporary_channel(original_frequency, requested_channel) + temporary_mask = 1 << target_antenna + if len(original_antenna.enabled_antennas) != 1: + raise RuntimeError("コマンドモードの使用アンテナが1件ではないため停止します") + if original_frequency.reserved != b"\x00\x00\x00\x00": + raise RuntimeError("周波数設定の予約領域が00hではないため書き込みを停止します") + # 実機へ送信する前に、変更用・復元用フレームの全入力を検証します。 + build_write_frequency_setting_command( + PARAMETER_KIND_COMMAND_MODE, + temporary_channel, + original_frequency.enabled_channels, + original_frequency.reserved, + ) + build_write_frequency_setting_command( + PARAMETER_KIND_COMMAND_MODE, + original_frequency.starting_channel, + original_frequency.enabled_channels, + original_frequency.reserved, + ) + + antenna_restore_needed = False + frequency_restore_needed = False + buzzer_verified = False + antenna_restored = False + frequency_restored = False + primary_error: Exception | None = None + restore_errors: list[str] = [] + inventory_tag_response_count = 0 + observed_rf_channel: int | None = None + observed_tag_antennas: tuple[int, ...] = () + + try: + buzzer_response = exchange(ser, build_buzzer_command(response_required=True, sound_type=0x00)) + verify_ack("7.3.2", buzzer_response, profile) + buzzer_verified = True + + if original_antenna.antenna_mask != temporary_mask: + antenna_restore_needed = True + write_command_mode_antenna_setting( + ser, + exchange, + profile, + original_antenna, + temporary_mask, + ) + + frequency_restore_needed = True + _write_frequency(ser, exchange, profile, original_frequency, temporary_channel) + + if verify_rf_channel: + inventory_response = exchange(ser, build_frame(0x55, b"\x10")) + inventory = verify_command_response( + get_command_spec("7.5.1"), + inventory_response, + profile, + ) + if inventory.result != VerificationResult.MULTI_RESPONSE_VERIFIED: + raise RuntimeError( + "7.5.1 応答異常: " + f"{inventory.result.value}; {' / '.join(inventory.notes)}" + ) + tag_frames = inventory.frames[:-1] + completion = inventory.frames[-1] + inventory_tag_response_count = len(tag_frames) + observed_rf_channel = completion.data[4] + if observed_rf_channel not in original_frequency.enabled_channels: + raise RuntimeError("Inventory完了ACKの実使用CHが使用許可CH外です") + antenna_values: list[int] = [] + for frame in tag_frames: + interpreted = interpret_address(frame, profile) + if interpreted.role == AddressRole.ANTENNA_NUMBER: + antenna_values.append(interpreted.value) + observed_tag_antennas = tuple(sorted(set(antenna_values))) + if observed_tag_antennas and observed_tag_antennas != (target_antenna,): + raise RuntimeError( + f"タグ応答のアンテナ番号がANT{target_antenna}と一致しません: " + f"{observed_tag_antennas}" + ) + except Exception as exc: + primary_error = exc + finally: + if frequency_restore_needed: + try: + clear_before_restore(ser) + restored_frequency = _write_frequency( + ser, + exchange, + profile, + original_frequency, + original_frequency.starting_channel, + ) + frequency_restored = ( + restored_frequency.starting_channel == original_frequency.starting_channel + and restored_frequency.enabled_channels == original_frequency.enabled_channels + ) + except Exception as exc: + restore_errors.append(f"周波数復元失敗: {exc}") + + if antenna_restore_needed: + try: + clear_before_restore(ser) + restored_antenna = write_command_mode_antenna_setting( + ser, + exchange, + profile, + original_antenna, + original_antenna.antenna_mask, + ) + antenna_restored = restored_antenna.antenna_mask == original_antenna.antenna_mask + except Exception as exc: + restore_errors.append(f"アンテナ復元失敗: {exc}") + else: + antenna_restored = True + + if restore_errors: + primary = "" if primary_error is None else f" / 実行時エラー: {primary_error}" + raise RuntimeError(" / ".join(restore_errors) + primary) + if primary_error is not None: + raise primary_error + + return SafeControlsResult( + buzzer_ack_verified=buzzer_verified, + original_antenna=tuple(original_antenna.enabled_antennas), + temporary_antenna=target_antenna, + antenna_restored=antenna_restored, + original_starting_channel=original_frequency.starting_channel, + temporary_starting_channel=temporary_channel, + enabled_channels_unchanged=True, + frequency_restored=frequency_restored, + rf_transmission_executed=verify_rf_channel, + inventory_tag_response_count=inventory_tag_response_count, + observed_rf_channel=observed_rf_channel, + observed_tag_antennas=observed_tag_antennas, + ) diff --git a/src/utr_usm02_safe_controls_cli.py b/src/utr_usm02_safe_controls_cli.py new file mode 100644 index 0000000..a1b56da --- /dev/null +++ b/src/utr_usm02_safe_controls_cli.py @@ -0,0 +1,95 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""USM02のブザー・ANT・開始CHを一時変更して復元するPhase 2 CLI。""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from src.utr_usm02_safe_controls import execute_safe_controls +from src.utr_usm02_v117_validation_cli import collect_bootstrap_profile + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="USM02 Phase 2安全制御確認") + parser.add_argument("--execute", action="store_true", help="実機送信を明示的に許可") + parser.add_argument("--port", default="COM6") + parser.add_argument("--baudrate", type=int, default=115200) + parser.add_argument("--target-antenna", type=int, default=1) + parser.add_argument("--test-channel", type=int, help="省略時は許可済み26〜32chから自動選択") + parser.add_argument( + "--verify-rf-channel", + action="store_true", + help="一時変更後にInventoryを1回実行し、実使用ANT/CHを確認", + ) + parser.add_argument("--json-out", type=Path) + return parser.parse_args() + + +def main() -> int: + args = _parse_args() + print("=== USM02 Phase 2 ===") + print("1. ブザー（応答要求あり）") + print(f"2. コマンドモードRAMのANT{args.target_antenna}一時切替・読戻し・復元") + print("3. コマンドモードRAMの開始CH一時変更・読戻し・復元") + if args.verify_rf_channel: + print("4. Inventoryを1回実行し、タグ固有値を保存せず実使用ANT/CHを確認") + print( + "FLASH変更なし / タグ書込なし / " + f"RF送信={'Inventory 1回のみ' if args.verify_rf_channel else 'なし'}" + ) + + if not args.execute: + print("dry-run完了。実機送信には --execute が必要です。") + return 0 + + import serial + + from src.utr_usb_sample_legacy import communicate + + with serial.Serial(port=args.port, baudrate=args.baudrate, timeout=0.05) as ser: + ser.reset_input_buffer() + ser.reset_output_buffer() + profile = collect_bootstrap_profile(ser, communicate) + result = execute_safe_controls( + ser, + profile, + communicate, + target_antenna=args.target_antenna, + requested_channel=args.test_channel, + verify_rf_channel=args.verify_rf_channel, + ) + + print("ブザーACK:", "OK" if result.buzzer_ack_verified else "NG") + print("アンテナ:", list(result.original_antenna), "->", result.temporary_antenna, "-> 復元") + print("アンテナ復元:", "OK" if result.antenna_restored else "NG") + print("開始CH:", result.original_starting_channel, "->", result.temporary_starting_channel, "-> 復元") + print("周波数復元:", "OK" if result.frequency_restored else "NG") + print("使用許可CH:", "変更なし" if result.enabled_channels_unchanged else "要確認") + if result.rf_transmission_executed: + print("RF送信: Inventory 1回") + print("タグ応答数:", result.inventory_tag_response_count) + print("タグ応答ANT:", list(result.observed_tag_antennas)) + print("完了ACK実使用CH:", result.observed_rf_channel) + else: + print("RF送信: なし") + + if args.json_out: + payload = { + "target": "UTR-SUN02-4CH / USM02", + "rom": profile.rom.firmware_version, + "port": args.port, + "baudrate": args.baudrate, + "result": result.to_dict(), + } + args.json_out.parent.mkdir(parents=True, exist_ok=True) + args.json_out.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print("JSON出力:", args.json_out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/utr_usm02_v117_validation.py b/src/utr_usm02_v117_validation.py new file mode 100644 index 0000000..8f30b4c --- /dev/null +++ b/src/utr_usm02_v117_validation.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""USM02向けVer.1.17検証計画を54件漏れなく生成する。""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import asdict, dataclass +from typing import Iterable + +from src.utr_v117_catalog import COMMAND_SPECS, CommandSpec, PlanStatus, classify_for_usm02 + + +@dataclass(frozen=True) +class ValidationPlanEntry: + section: str + name: str + command_hex: str + detail_hex: str | None + subcommand_hex: str | None + response_pattern: str + ack_data_length: str + operation: str + plan_status: str + reason: str + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +def _hex(value: int | None) -> str | None: + return None if value is None else f"{value:02X}h" + + +def _reason(spec: CommandSpec, status: PlanStatus) -> str: + if status == PlanStatus.READY: + return "対象機種・ROM・ユーザー条件を満たせば検証可能" + if status == PlanStatus.ROM_CHECK_REQUIRED: + return f"実機ROM確認が必要（Ver.{spec.minimum_rom // 1000}.{spec.minimum_rom % 1000:03d}以降）" + if status == PlanStatus.NOT_SUPPORTED_BY_DEVICE: + return "USM02（4CH）では対象外のコマンド" + if status == PlanStatus.NOT_SUPPORTED_BY_ROM: + return f"実機ROMが必要版未満（必要={spec.minimum_rom}）" + if status == PlanStatus.NOT_EXECUTED_BY_USER_DECISION: + return spec.hold_reason or "ユーザー判断により実行しない" + if status == PlanStatus.CONDITION_NOT_AVAILABLE: + return spec.condition or "必要条件を準備できていない" + return status.value + + +def build_validation_plan( + rom_number: int | None, + *, + conditions_available: Iterable[str] = (), +) -> tuple[ValidationPlanEntry, ...]: + conditions = frozenset(conditions_available) + entries: list[ValidationPlanEntry] = [] + for spec in COMMAND_SPECS: + status = classify_for_usm02(spec, rom_number, conditions_available=conditions) + entries.append( + ValidationPlanEntry( + section=spec.section, + name=spec.name, + command_hex=_hex(spec.command) or "", + detail_hex=_hex(spec.detail), + subcommand_hex=_hex(spec.subcommand), + response_pattern=spec.response_pattern.value, + ack_data_length=spec.ack_data_length, + operation=spec.operation.value, + plan_status=status.value, + reason=_reason(spec, status), + ) + ) + if len(entries) != 54: + raise RuntimeError("検証計画が54件ではありません") + return tuple(entries) + + +def summarize_plan(entries: Iterable[ValidationPlanEntry]) -> dict[str, int]: + entries_tuple = tuple(entries) + if len(entries_tuple) != 54: + raise ValueError("集計対象は54件である必要があります") + return dict(sorted(Counter(entry.plan_status for entry in entries_tuple).items())) + + +HARD_SAFETY_RULES: tuple[str, ...] = ( + "Kill(7.5.5)は送信しない", + "Lock(7.5.6)は送信しない", + "UHF_Encode(7.5.10)は送信しない", + "FLASH設定初期化(7.3.11)は送信しない", + "予期しないNACK・timeout・復元失敗で後続送信を停止する", + "Access/Kill Passwordをログへ出さない", + "PC/UII/EPC/TIDの実値を共有用ログへ出さない", + "設定変更は事前読取・同値書込・再読取・復元確認を必須とする", + "タグ書込は廃棄可能タグのUser領域だけを対象にし、事前読取と復元を必須とする", +) diff --git a/src/utr_usm02_v117_validation_cli.py b/src/utr_usm02_v117_validation_cli.py new file mode 100644 index 0000000..9384835 --- /dev/null +++ b/src/utr_usm02_v117_validation_cli.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""USM02 Ver.1.17検証計画CLI。 + +既定はdry-runであり、COMポートを開かない。実機の事前確認だけを行う場合も +``--execute-bootstrap`` を明示する必要がある。 +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from src.utr_antenna import ( + get_model_profile, + identify_model_key_from_rom, + parse_antenna_switching_setting_response, + parse_check_antenna_response, + parse_rom_version_response, +) +from src.utr_commands import ( + COMMANDS, + PARAMETER_KIND_COMMAND_MODE, + build_check_antenna_command, + build_frame, + build_read_antenna_switching_setting_command, +) +from src.utr_device_profile import build_usm02_device_profile +from src.utr_inventory import parse_inventory_param_response +from src.utr_response_settings import parse_epc_uii_response_settings +from src.utr_response_v117 import VerificationResult, verify_command_response +from src.utr_usm02_v117_validation import HARD_SAFETY_RULES, build_validation_plan, summarize_plan +from src.utr_v117_catalog import get_command_spec + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="UTR-SUN02-4CH/USM02 Ver.1.17 54コマンド検証計画") + parser.add_argument("--port", default="COM6") + parser.add_argument("--baudrate", type=int, default=115200) + parser.add_argument("--rom-number", type=int, help="dry-run用。例: Ver.2.052は2052") + parser.add_argument("--execute-bootstrap", action="store_true", help="ROM・設定・ANT0〜3だけを実機から読み取る") + parser.add_argument("--json-out", type=Path, help="マスク済み計画JSONの保存先（任意）") + return parser.parse_args() + + +def _print_plan(rom_number: int | None) -> tuple[dict[str, object], ...]: + plan = build_validation_plan(rom_number) + print("\n=== 54コマンド検証計画 ===") + for entry in plan: + print(f"{entry.section:6} {entry.plan_status:31} {entry.name} / 応答={entry.response_pattern} DATA長={entry.ack_data_length}") + print("\n集計:", summarize_plan(plan)) + return tuple(entry.to_dict() for entry in plan) + + +def _assert_verified(section: str, response: bytes, profile, *, response_requested: bool = True) -> None: + result = verify_command_response( + get_command_spec(section), response, profile, response_requested=response_requested + ) + if result.result not in { + VerificationResult.ACK_VERIFIED, + VerificationResult.NO_RESPONSE_VERIFIED, + VerificationResult.CONDITIONAL_NO_RESPONSE_VERIFIED, + }: + raise RuntimeError(f"{section} の応答検証に失敗しました: {result.result.value}; {' / '.join(result.notes)}") + + +def collect_bootstrap_profile(ser, communicate): + """接続済みserialからUSM02の応答解釈用プロファイルを収集する。""" + rom_response = communicate(ser, COMMANDS["ROM_VERSION_CHECK"]) + rom = parse_rom_version_response(rom_response) + model_key = identify_model_key_from_rom(rom) + if model_key != "UTR-SUN02-4CH": + raise RuntimeError(f"対象機種不一致のため停止します: ROM={rom.raw_text}, model={model_key}") + + # コマンドモードへ切替。FLASHは変更しない。 + mode_response = communicate(ser, COMMANDS["COMMAND_MODE_SET"]) + + antenna_response = communicate( + ser, + build_read_antenna_switching_setting_command(PARAMETER_KIND_COMMAND_MODE), + ) + antenna_setting = parse_antenna_switching_setting_response(antenna_response) + + model = get_model_profile(model_key) + connected: list[int] = [] + for target in model.check_targets: + response = communicate(ser, build_check_antenna_command(target.number)) + check = parse_check_antenna_response(response) + if check.is_connected: + connected.append(target.number) + + inventory_response = communicate(ser, build_frame(0x55, b"\x41\x00")) + inventory_settings = parse_inventory_param_response(inventory_response) + + epc_command_response = communicate(ser, build_frame(0x55, b"\x43\x05\x00")) + epc_command = parse_epc_uii_response_settings(epc_command_response) + epc_auto_response = communicate(ser, build_frame(0x55, b"\x43\x05\x01")) + epc_auto = parse_epc_uii_response_settings(epc_auto_response) + + profile = build_usm02_device_profile( + rom, + antenna_setting, + connected, + inventory_tid_enabled=inventory_settings["tid_enabled"], + epc_buffering_enabled=epc_command.epc_buffering_enabled, + read_cycle_completion_enabled=epc_auto.read_cycle_completion_enabled, + antenna_switch_completion_enabled=epc_auto.antenna_switch_completion_enabled, + carrier_detect_response_enabled=epc_auto.carrier_detect_response_enabled, + ) + _assert_verified("7.4.16", mode_response, profile) + _assert_verified("7.4.5", antenna_response, profile) + _assert_verified("7.4.3", inventory_response, profile) + _assert_verified("7.4.9", epc_command_response, profile) + _assert_verified("7.4.9", epc_auto_response, profile) + return profile + + +def _execute_bootstrap(port: str, baudrate: int): + """ROM、コマンドモード設定、応答依存設定、物理ANT状態を安全に取得する。""" + import serial + + from src.utr_usb_sample_legacy import communicate + + print(f"実機事前確認: port={port}, baudrate={baudrate}") + with serial.Serial(port=port, baudrate=baudrate, timeout=0.05) as ser: + ser.reset_input_buffer() + ser.reset_output_buffer() + profile = collect_bootstrap_profile(ser, communicate) + + print("ROM:", profile.rom.raw_text) + print("機種:", profile.model_key) + print("物理容量:", profile.antenna_capacity) + print("設定アンテナ:", list(profile.configured_antennas)) + print("接続OKアンテナ:", list(profile.connected_antennas)) + print("アンテナID出力:", profile.antenna_id_output_enabled) + print("TID付加:", profile.inventory_tid_enabled) + print("EPCバッファリング:", profile.epc_buffering_enabled) + print("自動読取サイクル完了応答:", profile.read_cycle_completion_enabled) + print("自動アンテナ切替完了応答:", profile.antenna_switch_completion_enabled) + print("キャリア検知応答:", profile.carrier_detect_response_enabled) + return profile + + +def main() -> int: + args = _parse_args() + print("=== 強制安全ルール ===") + for rule in HARD_SAFETY_RULES: + print("-", rule) + + profile = None + rom_number = args.rom_number + if args.execute_bootstrap: + profile = _execute_bootstrap(args.port, args.baudrate) + rom_number = profile.rom_number + + plan = _print_plan(rom_number) + if args.json_out: + payload: dict[str, object] = { + "target": "UTR-SUN02-4CH / USM02", + "port": args.port, + "baudrate": args.baudrate, + "real_device_bootstrap_executed": bool(args.execute_bootstrap), + "rom_number": rom_number, + "profile": None if profile is None else { + "model_key": profile.model_key, + "rom": profile.rom.raw_text, + "antenna_capacity": profile.antenna_capacity, + "configured_antennas": list(profile.configured_antennas), + "connected_antennas": list(profile.connected_antennas), + "antenna_id_output_enabled": profile.antenna_id_output_enabled, + "inventory_tid_enabled": profile.inventory_tid_enabled, + "epc_buffering_enabled": profile.epc_buffering_enabled, + "read_cycle_completion_enabled": profile.read_cycle_completion_enabled, + "antenna_switch_completion_enabled": profile.antenna_switch_completion_enabled, + "carrier_detect_response_enabled": profile.carrier_detect_response_enabled, + }, + "hard_safety_rules": list(HARD_SAFETY_RULES), + "plan": plan, + } + args.json_out.parent.mkdir(parents=True, exist_ok=True) + args.json_out.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print("JSON出力:", args.json_out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/utr_v117_catalog.py b/src/utr_v117_catalog.py new file mode 100644 index 0000000..a18a7b4 --- /dev/null +++ b/src/utr_v117_catalog.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +"""UTR-S201通信プロトコル Ver.1.17 の54コマンド仕様カタログ。 + +このモジュールは実機通信を行わない。公式説明書6.1/7.3/7.4/7.5の +コマンド集合、応答形態、USM02での適用可否、安全上の保留条件を、 +検証プログラムから機械判定できる形で保持する。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class ResponsePattern(str, Enum): + """正常系で期待する応答形態。""" + + SINGLE_ACK = "single_ack" + CONDITIONAL_ACK = "conditional_ack" + NO_RESPONSE = "no_response" + MULTI_TAG_THEN_ACK = "multi_tag_then_ack" + VARIABLE_ACK = "variable_ack" + + +class OperationKind(str, Enum): + READ_ONLY = "read_only" + CONTROL = "control" + RAM_WRITE = "ram_write" + FLASH_WRITE = "flash_write" + RF_READ = "rf_read" + TAG_WRITE = "tag_write" + DESTRUCTIVE = "destructive" + + +class PlanStatus(str, Enum): + READY = "READY" + ROM_CHECK_REQUIRED = "ROM_CHECK_REQUIRED" + NOT_SUPPORTED_BY_DEVICE = "NOT_SUPPORTED_BY_DEVICE" + NOT_SUPPORTED_BY_ROM = "NOT_SUPPORTED_BY_ROM" + CONDITION_NOT_AVAILABLE = "CONDITION_NOT_AVAILABLE" + NOT_EXECUTED_BY_USER_DECISION = "NOT_EXECUTED_BY_USER_DECISION" + + +@dataclass(frozen=True) +class CommandSpec: + section: str + name: str + command: int + detail: int | None + subcommand: int | None + response_pattern: ResponsePattern + ack_data_length: str + operation: OperationKind + response_detail: int | None = None + minimum_rom: int | None = None + usm02_supported: bool = True + hold_reason: str | None = None + condition: str | None = None + + @property + def key(self) -> str: + return self.section + + +def _spec( + section: str, + name: str, + command: int, + detail: int | None, + response_pattern: ResponsePattern, + ack_data_length: str, + operation: OperationKind, + *, + subcommand: int | None = None, + minimum_rom: int | None = None, + usm02_supported: bool = True, + hold_reason: str | None = None, + condition: str | None = None, + response_detail: int | None = None, +) -> CommandSpec: + return CommandSpec( + section=section, + name=name, + command=command, + detail=detail, + subcommand=subcommand, + response_pattern=response_pattern, + ack_data_length=ack_data_length, + operation=operation, + response_detail=detail if response_detail is None else response_detail, + minimum_rom=minimum_rom, + usm02_supported=usm02_supported, + hold_reason=hold_reason, + condition=condition, + ) + + +# ack_data_length はACKフレームのDATA長。可変長は式で表す。 +COMMAND_SPECS: tuple[CommandSpec, ...] = ( + # 7.3 リーダライタ制御コマンド（12件） + _spec("7.3.1", "エラー情報の読み取り", 0x4F, 0x80, ResponsePattern.SINGLE_ACK, "04", OperationKind.READ_ONLY), + _spec("7.3.2", "ブザーの制御", 0x42, None, ResponsePattern.CONDITIONAL_ACK, "00 (応答要求=1)", OperationKind.CONTROL), + _spec("7.3.3", "LED&ブザーの制御", 0x4E, 0x57, ResponsePattern.SINGLE_ACK, "01", OperationKind.CONTROL), + _spec("7.3.4", "RF送信信号の制御", 0x4E, 0x9E, ResponsePattern.SINGLE_ACK, "02", OperationKind.CONTROL), + _spec("7.3.5", "UHF_CheckAntenna", 0x55, 0x44, ResponsePattern.SINGLE_ACK, "03", OperationKind.READ_ONLY), + _spec("7.3.6", "使用アンテナ番号の読み取り", 0x55, 0x48, ResponsePattern.SINGLE_ACK, "04", OperationKind.READ_ONLY, usm02_supported=False), + _spec("7.3.7", "使用アンテナ番号の書き込み", 0x55, 0x38, ResponsePattern.SINGLE_ACK, "04", OperationKind.RAM_WRITE, usm02_supported=False, response_detail=0x48), + _spec("7.3.8", "ROMバージョンの読み取り", 0x4F, 0x90, ResponsePattern.SINGLE_ACK, "0A", OperationKind.READ_ONLY), + _spec("7.3.9", "チップバージョンの読み取り", 0x55, 0x90, ResponsePattern.VARIABLE_ACK, "0B(ファームウェア) / 0C(シリアル)", OperationKind.READ_ONLY), + _spec("7.3.10", "リスタート", 0x4E, 0x9D, ResponsePattern.NO_RESPONSE, "応答なし", OperationKind.CONTROL), + _spec( + "7.3.11", "FLASH設定の初期化", 0x4E, 0x6F, ResponsePattern.SINGLE_ACK, "01", OperationKind.FLASH_WRITE, + hold_reason="FLASH全設定を初期化するため実行しない", + ), + _spec("7.3.12", "UHF_GetHandle", 0x55, 0x46, ResponsePattern.SINGLE_ACK, "03", OperationKind.RF_READ, minimum_rom=2050), + + # 7.4 リーダライタ設定コマンド（31件） + _spec("7.4.1", "リーダライタ動作モードの読み取り", 0x4F, 0x00, ResponsePattern.SINGLE_ACK, "09", OperationKind.READ_ONLY), + _spec("7.4.2", "UHF_GetSelectParam", 0x55, 0x40, ResponsePattern.VARIABLE_ACK, "9+n", OperationKind.READ_ONLY), + _spec("7.4.3", "UHF_GetInventoryParam", 0x55, 0x41, ResponsePattern.SINGLE_ACK, "0B", OperationKind.READ_ONLY), + _spec("7.4.4", "UHF_GetExpandSelectParam", 0x55, 0x42, ResponsePattern.VARIABLE_ACK, "23*n+2", OperationKind.READ_ONLY), + _spec("7.4.5", "アンテナ切替設定の読み取り", 0x55, 0x43, ResponsePattern.SINGLE_ACK, "08", OperationKind.READ_ONLY, subcommand=0x00), + _spec("7.4.6", "出力設定の読み取り", 0x55, 0x43, ResponsePattern.SINGLE_ACK, "0B", OperationKind.READ_ONLY, subcommand=0x01), + _spec("7.4.7", "周波数設定の読み取り", 0x55, 0x43, ResponsePattern.SINGLE_ACK, "0C", OperationKind.READ_ONLY, subcommand=0x02), + _spec("7.4.8", "RFタグ通信関連パラメータの読み取り", 0x55, 0x43, ResponsePattern.SINGLE_ACK, "06", OperationKind.READ_ONLY, subcommand=0x04), + _spec("7.4.9", "EPC(UII)関連パラメータの読み取り", 0x55, 0x43, ResponsePattern.SINGLE_ACK, "04", OperationKind.READ_ONLY, subcommand=0x05), + _spec("7.4.10", "外部アンテナ自動切替設定の読み取り", 0x55, 0x47, ResponsePattern.SINGLE_ACK, "0C", OperationKind.READ_ONLY, usm02_supported=False), + _spec("7.4.11", "汎用ポート値の読み取り", 0x4F, 0x9F, ResponsePattern.SINGLE_ACK, "05", OperationKind.READ_ONLY), + _spec("7.4.12", "拡張ポート値の読み取り", 0x4F, 0xA0, ResponsePattern.SINGLE_ACK, "05", OperationKind.READ_ONLY, usm02_supported=False), + _spec("7.4.13", "FLASH設定値の読み取り(1バイトアクセス)", 0x4F, 0xB4, ResponsePattern.SINGLE_ACK, "02", OperationKind.READ_ONLY), + _spec("7.4.14", "RSSIフィルタ設定の読み取り", 0x55, 0x49, ResponsePattern.VARIABLE_ACK, "12*n+5", OperationKind.READ_ONLY, minimum_rom=2100), + _spec("7.4.15", "アンテナ個別送信出力設定の読み取り", 0x55, 0x4A, ResponsePattern.VARIABLE_ACK, "5+n", OperationKind.READ_ONLY, minimum_rom=2100), + _spec("7.4.16", "リーダライタ動作モードの書き込み", 0x4E, 0x00, ResponsePattern.SINGLE_ACK, "00", OperationKind.RAM_WRITE, subcommand=0x10), + _spec("7.4.17", "UHF_SetSelectParam", 0x55, 0x30, ResponsePattern.SINGLE_ACK, "01", OperationKind.RAM_WRITE), + _spec("7.4.18", "UHF_SetInventoryParam", 0x55, 0x31, ResponsePattern.SINGLE_ACK, "01", OperationKind.RAM_WRITE), + _spec("7.4.19", "UHF_SetExpandSelectParam", 0x55, 0x32, ResponsePattern.SINGLE_ACK, "01", OperationKind.RAM_WRITE), + _spec("7.4.20", "アンテナ切替設定の書き込み", 0x55, 0x33, ResponsePattern.SINGLE_ACK, "08", OperationKind.RAM_WRITE, subcommand=0x00), + _spec("7.4.21", "出力設定の書き込み", 0x55, 0x33, ResponsePattern.SINGLE_ACK, "0B", OperationKind.RAM_WRITE, subcommand=0x01), + _spec("7.4.22", "周波数設定の書き込み", 0x55, 0x33, ResponsePattern.SINGLE_ACK, "0C", OperationKind.RAM_WRITE, subcommand=0x02), + _spec("7.4.23", "Accessパスワードの書き込み", 0x55, 0x33, ResponsePattern.SINGLE_ACK, "03", OperationKind.RAM_WRITE, subcommand=0x03), + _spec("7.4.24", "RFタグ通信関連パラメータの書き込み", 0x55, 0x33, ResponsePattern.SINGLE_ACK, "03", OperationKind.RAM_WRITE, subcommand=0x04), + _spec("7.4.25", "EPC(UII)関連パラメータの書き込み", 0x55, 0x33, ResponsePattern.SINGLE_ACK, "04", OperationKind.RAM_WRITE, subcommand=0x05), + _spec("7.4.26", "外部アンテナ自動切替設定の書き込み", 0x55, 0x37, ResponsePattern.SINGLE_ACK, "0C", OperationKind.RAM_WRITE, usm02_supported=False), + _spec("7.4.27", "汎用ポート値の書き込み", 0x4E, 0x9F, ResponsePattern.SINGLE_ACK, "01", OperationKind.RAM_WRITE), + _spec("7.4.28", "拡張ポート値の書き込み", 0x4E, 0xA0, ResponsePattern.SINGLE_ACK, "01", OperationKind.RAM_WRITE, usm02_supported=False), + _spec("7.4.29", "FLASH設定値の書き込み(1バイトアクセス)", 0x4E, 0xB4, ResponsePattern.SINGLE_ACK, "01", OperationKind.FLASH_WRITE), + _spec("7.4.30", "RSSIフィルタ設定の書き込み", 0x55, 0x39, ResponsePattern.SINGLE_ACK, "05", OperationKind.RAM_WRITE, minimum_rom=2100), + _spec("7.4.31", "アンテナ個別送信出力設定の書き込み", 0x55, 0x3A, ResponsePattern.SINGLE_ACK, "05", OperationKind.RAM_WRITE, minimum_rom=2100), + + # 7.5 RFタグ通信コマンド（11件） + _spec("7.5.1", "UHF_Inventory", 0x55, 0x10, ResponsePattern.MULTI_TAG_THEN_ACK, "TAG:5+n / 完了ACK:05", OperationKind.RF_READ), + _spec("7.5.2", "UHF_InventoryRead", 0x55, 0x14, ResponsePattern.MULTI_TAG_THEN_ACK, "TAG:7+n1+n2+n3 / 完了ACK:05", OperationKind.RF_READ), + _spec("7.5.3", "UHF_Read", 0x55, 0x15, ResponsePattern.VARIABLE_ACK, "n+2", OperationKind.RF_READ), + _spec("7.5.4", "UHF_Write", 0x55, 0x16, ResponsePattern.SINGLE_ACK, "01", OperationKind.TAG_WRITE), + _spec( + "7.5.5", "UHF_Kill", 0x55, 0x17, ResponsePattern.SINGLE_ACK, "01", OperationKind.DESTRUCTIVE, + hold_reason="ユーザー指定によりKillは実行しない", + ), + _spec( + "7.5.6", "UHF_Lock", 0x55, 0x18, ResponsePattern.SINGLE_ACK, "01", OperationKind.DESTRUCTIVE, + hold_reason="ユーザー指定によりLockは実行しない", + ), + _spec("7.5.7", "UHF_BlockWrite", 0x55, 0x1A, ResponsePattern.SINGLE_ACK, "01", OperationKind.TAG_WRITE), + _spec("7.5.8", "UHF_BlockErase", 0x55, 0x1B, ResponsePattern.SINGLE_ACK, "01", OperationKind.TAG_WRITE), + _spec("7.5.9", "UHF_BlockWrite2", 0x55, 0x1D, ResponsePattern.SINGLE_ACK, "01", OperationKind.TAG_WRITE), + _spec( + "7.5.10", "UHF_Encode", 0x55, 0x1E, ResponsePattern.SINGLE_ACK, "01", OperationKind.DESTRUCTIVE, + hold_reason="複数領域変更とLockを含み得るため実行しない", + ), + _spec( + "7.5.11", "UHF_ThroughCmd", 0x55, 0xFF, ResponsePattern.VARIABLE_ACK, "1..255", OperationKind.RF_READ, + minimum_rom=2050, + condition="TIDでタグICを特定し、そのIC向けの安全なコマンド仕様を用意できた場合のみ", + ), +) + + +def get_command_spec(section: str) -> CommandSpec: + """PDF節番号で1件を取得する。""" + for spec in COMMAND_SPECS: + if spec.section == section: + return spec + raise KeyError(section) + + +def classify_for_usm02( + spec: CommandSpec, + rom_number: int | None, + *, + conditions_available: frozenset[str] = frozenset(), +) -> PlanStatus: + """USM02での実機検証可否を安全側に分類する。""" + if not spec.usm02_supported: + return PlanStatus.NOT_SUPPORTED_BY_DEVICE + if rom_number is None and spec.minimum_rom is not None: + return PlanStatus.ROM_CHECK_REQUIRED + if spec.minimum_rom is not None and rom_number < spec.minimum_rom: + return PlanStatus.NOT_SUPPORTED_BY_ROM + if spec.hold_reason is not None: + return PlanStatus.NOT_EXECUTED_BY_USER_DECISION + if spec.condition is not None and spec.section not in conditions_available: + return PlanStatus.CONDITION_NOT_AVAILABLE + return PlanStatus.READY + + +def validate_catalog() -> None: + """件数、節番号、コマンド識別子の基本整合性を検査する。""" + if len(COMMAND_SPECS) != 54: + raise ValueError(f"command catalog must contain 54 entries: {len(COMMAND_SPECS)}") + sections = [spec.section for spec in COMMAND_SPECS] + if len(set(sections)) != len(sections): + raise ValueError("duplicate PDF section in command catalog") + for spec in COMMAND_SPECS: + for name, value in (("command", spec.command), ("detail", spec.detail), ("subcommand", spec.subcommand)): + if value is not None and not 0 <= value <= 0xFF: + raise ValueError(f"{spec.section}: {name} is not a byte") + + +validate_catalog() diff --git a/tests/test_antenna_helpers.py b/tests/test_antenna_helpers.py index 179c345..29e17f0 100644 --- a/tests/test_antenna_helpers.py +++ b/tests/test_antenna_helpers.py @@ -285,4 +285,3 @@ def test_parse_antenna_switching_setting_write_response_for_command_mode_ant1(): assert setting.antenna_id_output_enabled is True assert setting.antenna_mask == 0x02 assert setting.enabled_antennas == [1] - diff --git a/tests/test_device_profile.py b/tests/test_device_profile.py new file mode 100644 index 0000000..f0540ba --- /dev/null +++ b/tests/test_device_profile.py @@ -0,0 +1,34 @@ +import pytest + +from src.utr_antenna import AntennaSwitchingSetting, RomVersionInfo +from src.utr_device_profile import DeviceProfile, build_usm02_device_profile + + +def _rom(series="USM02", minor="052"): + return RomVersionInfo(raw_text=f"2{minor}{series}", major_version="2", minor_version=minor, series_name=series) + + +def _antenna_setting(mask=0x05, antenna_id=True): + return AntennaSwitchingSetting( + parameter_kind=0, + switching_mode=0, + antenna_id_output_enabled=antenna_id, + antenna_mask=mask, + reserved=b"\x00\x00\x00", + raw=b"", + ) + + +def test_profile_keeps_capacity_configured_and_connected_counts_separate(): + profile = build_usm02_device_profile(_rom(), _antenna_setting(), [0, 2, 3]) + assert profile.rom_number == 2052 + assert profile.antenna_capacity == 4 + assert profile.configured_antennas == (0, 2) + assert profile.connected_antennas == (0, 2, 3) + + +def test_profile_rejects_non_usm02_or_out_of_range_antenna(): + with pytest.raises(ValueError): + DeviceProfile("UTR-S201", _rom(series="USM01"), antenna_capacity=4).validate() + with pytest.raises(ValueError): + build_usm02_device_profile(_rom(), _antenna_setting(), [4]) diff --git a/tests/test_frequency_setting_commands.py b/tests/test_frequency_setting_commands.py new file mode 100644 index 0000000..f1bd34c --- /dev/null +++ b/tests/test_frequency_setting_commands.py @@ -0,0 +1,37 @@ +import pytest + +from src.utr_commands import ( + PARAMETER_KIND_AUTO_READ_MODE, + PARAMETER_KIND_COMMAND_MODE, + PARAMETER_KIND_FLASH, + build_read_frequency_setting_command, + build_write_frequency_setting_command, +) + + +def test_build_read_frequency_setting_command_for_command_mode(): + assert build_read_frequency_setting_command(PARAMETER_KIND_COMMAND_MODE) == bytes.fromhex( + "02 00 55 03 43 02 00 03 A2 0D" + ) + + +def test_build_write_frequency_preserves_enabled_mask_and_changes_start_only(): + frame = build_write_frequency_setting_command( + PARAMETER_KIND_COMMAND_MODE, + 27, + (26, 27, 28, 29, 30, 31, 32), + ) + assert frame == bytes.fromhex( + "02 00 55 0C 33 02 00 1B 00 C0 1F 00 00 00 00 00 03 95 0D" + ) + + +@pytest.mark.parametrize("parameter_kind", [PARAMETER_KIND_AUTO_READ_MODE, PARAMETER_KIND_FLASH]) +def test_write_frequency_rejects_non_command_mode_targets(parameter_kind): + with pytest.raises(ValueError, match="command-mode"): + build_write_frequency_setting_command(parameter_kind, 27, (26, 27)) + + +def test_write_frequency_requires_start_channel_to_be_enabled(): + with pytest.raises(ValueError, match="included"): + build_write_frequency_setting_command(PARAMETER_KIND_COMMAND_MODE, 27, (26, 28)) diff --git a/tests/test_reader_settings.py b/tests/test_reader_settings.py index 36045d5..838839b 100644 --- a/tests/test_reader_settings.py +++ b/tests/test_reader_settings.py @@ -20,7 +20,7 @@ def test_parse_output_power_setting_response_reads_dbm_and_raw_hex(): parsed = parse_output_power_setting_response(response) assert parsed["detail_command"] == 0x43 - assert parsed["parameter_kind"] == 0x01 + assert parsed["parameter_kind"] == 0x00 assert parsed["output_power_dbm"] == 24.0 assert parsed["raw_value_hex"] == "F0 00" assert parsed["raw_data_hex"] == "43 01 00 F0 00" @@ -49,27 +49,24 @@ def test_parse_output_power_setting_response_reads_timing_values_when_available( def test_parse_frequency_setting_response_reads_channel_and_frequency(): - response = bytes.fromhex("02 00 30 04 43 02 00 05 03 83 0D") + response = bytes.fromhex("02 00 30 0C 43 02 00 1A 1A C0 1F 00 00 00 00 00 03 99 0D") parsed = parse_frequency_setting_response(response) assert parsed["detail_command"] == 0x43 - assert parsed["parameter_kind"] == 0x02 - assert parsed["channel_number"] == 5 - assert parsed["frequency_mhz"] == 916.8 - assert parsed["raw_data_hex"] == "43 02 00 05" - assert "送信周波数: 916.8 MHz" in format_frequency_setting(parsed) - + assert parsed["parameter_kind"] == 0x00 + assert parsed["starting_channel_number"] == 26 + assert parsed["current_channel_number"] == 26 + assert parsed["enabled_channels"] == (26, 27, 28, 29, 30, 31, 32) + assert parsed["frequency_mhz"] == 921.0 + assert "現在チャンネル周波数: 921.0 MHz" in format_frequency_setting(parsed) -def test_parse_frequency_setting_response_keeps_unknown_channel_as_raw_hex(): - response = bytes.fromhex("02 00 30 04 43 02 00 FF 03 7D 0D") - parsed = parse_frequency_setting_response(response) +def test_parse_frequency_setting_response_rejects_short_legacy_shape(): + response = bytes.fromhex("02 00 30 04 43 02 00 05 03 83 0D") - assert parsed["channel_number"] == 0xFF - assert parsed["frequency_mhz"] is None - assert parsed["raw_data_hex"] == "43 02 00 FF" - assert "既知範囲外" in "\n".join(format_frequency_setting(parsed)) + with pytest.raises(ValueError, match="短すぎ"): + parse_frequency_setting_response(response) def test_parse_output_power_setting_response_rejects_broken_sum(): diff --git a/tests/test_response_settings.py b/tests/test_response_settings.py new file mode 100644 index 0000000..9a30be0 --- /dev/null +++ b/tests/test_response_settings.py @@ -0,0 +1,18 @@ +import pytest + +from src.utr_commands import build_frame +from src.utr_response_settings import parse_epc_uii_response_settings + + +def test_parse_epc_uii_response_flags(): + parsed = parse_epc_uii_response_settings(build_frame(0x30, b"\x43\x05\x01\x0D")) + assert parsed.parameter_kind == 1 + assert parsed.epc_buffering_enabled + assert not parsed.read_cycle_completion_enabled + assert parsed.antenna_switch_completion_enabled + assert parsed.carrier_detect_response_enabled + + +def test_parse_epc_uii_response_rejects_reserved_bits(): + with pytest.raises(ValueError): + parse_epc_uii_response_settings(build_frame(0x30, b"\x43\x05\x00\x80")) diff --git a/tests/test_response_v117.py b/tests/test_response_v117.py new file mode 100644 index 0000000..c1493ee --- /dev/null +++ b/tests/test_response_v117.py @@ -0,0 +1,83 @@ +from src.utr_antenna import RomVersionInfo +from src.utr_commands import build_frame +from src.utr_device_profile import DeviceProfile +from src.utr_response_v117 import ( + AddressRole, + FrameKind, + VerificationResult, + interpret_address, + parse_frame, + split_frames, + verify_command_response, +) +from src.utr_v117_catalog import get_command_spec + + +def _profile(antenna_id=True): + return DeviceProfile( + model_key="UTR-SUN02-4CH", + rom=RomVersionInfo("2052USM02", "2", "052", "USM02"), + antenna_capacity=4, + configured_antennas=(0,), + connected_antennas=(0, 1), + antenna_id_output_enabled=antenna_id, + ) + + +def test_parse_and_split_frames_verify_sum_and_structure(): + first = build_frame(0x30, b"\x80\x00\x00\x00") + second = build_frame(0x30, b"\x6F") + frames = split_frames(first + second) + assert [frame.kind for frame in frames] == [FrameKind.ACK, FrameKind.ACK] + broken = bytearray(first) + broken[-2] ^= 0x01 + try: + parse_frame(bytes(broken)) + except ValueError as exc: + assert "SUM" in str(exc) + else: + raise AssertionError("SUM不一致を受理してはいけません") + + +def test_tag_response_address_depends_on_antenna_id_setting(): + frame = parse_frame(build_frame(0x6C, b"\x10\x00\x00\x00\x00", address=2)) + assert interpret_address(frame, _profile(True)).role == AddressRole.ANTENNA_NUMBER + assert interpret_address(frame, _profile(False)).role == AddressRole.READER_ID + + +def test_fixed_ack_and_no_response_are_different_results(): + ack = build_frame(0x30, b"\x80\x00\x00\x00") + result = verify_command_response(get_command_spec("7.3.1"), ack, _profile()) + assert result.result == VerificationResult.ACK_VERIFIED + restart = verify_command_response(get_command_spec("7.3.10"), b"", _profile()) + assert restart.result == VerificationResult.NO_RESPONSE_VERIFIED + + +def test_conditional_buzzer_no_response_requires_response_flag_off(): + spec = get_command_spec("7.3.2") + assert verify_command_response(spec, b"", _profile(), response_requested=False).result == VerificationResult.CONDITIONAL_NO_RESPONSE_VERIFIED + assert verify_command_response(spec, b"", _profile(), response_requested=True).result == VerificationResult.FAILED_NEEDS_ANALYSIS + + +def test_inventory_accepts_zero_or_more_tag_frames_then_completion_ack(): + tag = build_frame(0x6C, b"\x10\x00\x00\x00\x00", address=1) + completion = build_frame(0x30, b"\x10\x00\x00\x00\x00") + result = verify_command_response(get_command_spec("7.5.1"), tag + completion, _profile()) + assert result.result == VerificationResult.MULTI_RESPONSE_VERIFIED + assert "タグ応答=1件" in result.notes[-1] + + +def test_nack_is_parsed_but_not_counted_as_ack_success(): + nack_data = b"\x15\x04\x00\x00\x00\x00\x00\x00\x00\x00" + result = verify_command_response(get_command_spec("7.5.3"), build_frame(0x31, nack_data), _profile()) + assert result.result == VerificationResult.NACK_OBSERVED + assert result.nack["error_code_1"] == 0x04 + + +def test_wrong_subcommand_and_wrong_length_fail(): + wrong_sub = build_frame(0x30, b"\x43\x01\x00\x80\x01\x00\x00\x00") + result = verify_command_response(get_command_spec("7.4.5"), wrong_sub, _profile()) + assert result.result == VerificationResult.FAILED_NEEDS_ANALYSIS + wrong_length = build_frame(0x30, b"\x80\x00") + result = verify_command_response(get_command_spec("7.3.1"), wrong_length, _profile()) + assert result.result == VerificationResult.FAILED_NEEDS_ANALYSIS diff --git a/tests/test_usm02_inventory_buzzer.py b/tests/test_usm02_inventory_buzzer.py new file mode 100644 index 0000000..b6c9e97 --- /dev/null +++ b/tests/test_usm02_inventory_buzzer.py @@ -0,0 +1,181 @@ +from dataclasses import replace + +import pytest + +from src.utr_antenna import RomVersionInfo +from src.utr_commands import build_frame +from src.utr_device_profile import DeviceProfile +from src.utr_usm02_inventory_buzzer import ( + execute_sequential_inventory_buzzer, + normalize_target_antennas, +) +from src.utr_usm02_inventory_buzzer_cli import parse_antenna_list + + +def _profile() -> DeviceProfile: + return DeviceProfile( + model_key="UTR-SUN02-4CH", + rom=RomVersionInfo("2052USM02", "2", "052", "USM02"), + antenna_capacity=4, + configured_antennas=(0,), + connected_antennas=(0, 1, 2), + antenna_id_output_enabled=True, + epc_buffering_enabled=False, + ) + + +def _antenna_read(mask: int) -> bytes: + return build_frame(0x30, bytes([0x43, 0x00, 0x00, 0x80, mask, 0, 0, 0])) + + +def _antenna_write(mask: int) -> bytes: + return build_frame(0x30, bytes([0x33, 0x00, 0x00, 0x80, mask, 0, 0, 0])) + + +def _inventory(antenna: int, tag_suffix: int, *, channel: int = 26) -> bytes: + tag = build_frame( + 0x6C, + bytes([0x10, 0x00, 0x00, 0x00, 0x02, 0xAA, tag_suffix]), + address=antenna, + ) + completion = build_frame(0x30, bytes([0x10, 0x00, 0x01, 0x00, channel])) + return tag + completion + + +def test_parse_and_normalize_antennas(): + assert parse_antenna_list("0, 1,2") == (0, 1, 2) + assert normalize_target_antennas((0, 1, 2), 4) == (0, 1, 2) + with pytest.raises(ValueError, match="重複"): + normalize_target_antennas((0, 1, 1), 4) + with pytest.raises(ValueError, match="物理容量外"): + normalize_target_antennas((0, 4), 4) + + +def test_tag_on_each_antenna_triggers_buzzer_and_restores_without_identifier_output(): + responses = [ + _antenna_read(0x01), + _inventory(0, 0x01), + build_frame(0x30, b""), + _antenna_write(0x02), + _antenna_read(0x02), + _inventory(1, 0x02), + build_frame(0x30, b""), + _antenna_write(0x04), + _antenna_read(0x04), + _inventory(2, 0x03), + build_frame(0x30, b""), + _antenna_write(0x01), + _antenna_read(0x01), + ] + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return responses[len(sent) - 1] + + class Serial: + def reset_input_buffer(self): + pass + + result = execute_sequential_inventory_buzzer(Serial(), _profile(), exchange) + + assert result.requested_antennas == (0, 1, 2) + assert result.inventory_execution_count == 3 + assert result.buzzer_ack_count == 3 + assert result.antenna_restored + assert all(item.tag_detected for item in result.antenna_results) + assert all(item.unique_tag_count == 1 for item in result.antenna_results) + assert all(item.response_count_matches_completion for item in result.antenna_results) + assert not result.flash_write_executed + assert not result.tag_memory_write_executed + + sent_data = [frame[4:4 + frame[3]] for frame in sent] + assert sent_data.count(b"\x10") == 3 + assert sum(data == b"\x01\x01" for data in sent_data) == 3 + assert all(not (data and data[0] in {0x17, 0x18, 0x1E, 0x6F}) for data in sent_data) + serialized = str(result.to_dict()) + assert "AA01" not in serialized + assert "AA02" not in serialized + assert "AA03" not in serialized + + +def test_zero_tag_response_does_not_send_buzzer(): + responses = [ + _antenna_read(0x01), + build_frame(0x30, b"\x10\x00\x00\x00\x1A"), + ] + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return responses[len(sent) - 1] + + result = execute_sequential_inventory_buzzer(object(), _profile(), exchange, target_antennas=(0,)) + + item = result.antenna_results[0] + assert not item.tag_detected + assert item.tag_response_count == 0 + assert item.unique_tag_count == 0 + assert not item.buzzer_command_sent + assert not item.buzzer_ack_verified + assert len(sent) == 2 + + +def test_missing_ant2_stops_before_any_exchange(): + profile = replace(_profile(), connected_antennas=(0, 1)) + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return b"" + + with pytest.raises(ValueError, match="ANT2"): + execute_sequential_inventory_buzzer(object(), profile, exchange) + + assert sent == [] + + +def test_epc_buffering_must_be_confirmed_off_before_any_exchange(): + profile = replace(_profile(), epc_buffering_enabled=True) + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return b"" + + with pytest.raises(ValueError, match="EPCバッファリング"): + execute_sequential_inventory_buzzer(object(), profile, exchange) + + assert sent == [] + + +def test_timeout_after_switch_still_restores_original_antenna(): + responses = [ + _antenna_read(0x01), + build_frame(0x30, b"\x10\x00\x00\x00\x1A"), + _antenna_write(0x02), + _antenna_read(0x02), + b"", + _antenna_write(0x01), + _antenna_read(0x01), + ] + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return responses[len(sent) - 1] + + class Serial: + def reset_input_buffer(self): + pass + + with pytest.raises(RuntimeError, match="FAILED_NEEDS_ANALYSIS"): + execute_sequential_inventory_buzzer( + Serial(), + _profile(), + exchange, + target_antennas=(0, 1), + ) + + assert sent[-2][4:9] == b"\x33\x00\x00\x80\x01" + assert sent[-1][4:7] == b"\x43\x00\x00" diff --git a/tests/test_usm02_safe_controls.py b/tests/test_usm02_safe_controls.py new file mode 100644 index 0000000..036de85 --- /dev/null +++ b/tests/test_usm02_safe_controls.py @@ -0,0 +1,155 @@ +from src.utr_antenna import RomVersionInfo +from src.utr_commands import build_frame +from src.utr_device_profile import DeviceProfile +from src.utr_reader_settings import FrequencySetting +from src.utr_usm02_safe_controls import choose_temporary_channel, execute_safe_controls +import pytest + + +def _profile(): + return DeviceProfile( + model_key="UTR-SUN02-4CH", + rom=RomVersionInfo("2052USM02", "2", "052", "USM02"), + antenna_capacity=4, + configured_antennas=(0,), + connected_antennas=(0, 1), + antenna_id_output_enabled=True, + ) + + +def _frequency_read(start=26, current=26): + return build_frame( + 0x30, + bytes([0x43, 0x02, 0x00, start, current, 0xC0, 0x1F, 0x00, 0, 0, 0, 0]), + ) + + +def _frequency_write_ack(start): + return build_frame( + 0x30, + bytes([0x33, 0x02, 0x00, start, 0x00, 0xC0, 0x1F, 0x00, 0, 0, 0, 0]), + ) + + +def test_choose_temporary_channel_prefers_permitted_26_to_32(): + setting = FrequencySetting(0, 26, 26, (5, 26, 27), b"\x41\x00\x00", b"\x00" * 4, b"") + assert choose_temporary_channel(setting) == 27 + + +def test_execute_safe_controls_changes_and_restores_without_rf_or_flash(): + responses = [ + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + _frequency_read(26, 26), + build_frame(0x30, b""), + build_frame(0x30, b"\x33\x00\x00\x80\x02\x00\x00\x00"), + build_frame(0x30, b"\x43\x00\x00\x80\x02\x00\x00\x00"), + _frequency_write_ack(27), + _frequency_read(27, 26), + _frequency_write_ack(26), + _frequency_read(26, 26), + build_frame(0x30, b"\x33\x00\x00\x80\x01\x00\x00\x00"), + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + ] + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return responses[len(sent) - 1] + + class Serial: + def reset_input_buffer(self): + pass + + result = execute_safe_controls(Serial(), _profile(), exchange, target_antenna=1) + + assert result.buzzer_ack_verified + assert result.antenna_restored + assert result.frequency_restored + assert result.original_starting_channel == 26 + assert result.temporary_starting_channel == 27 + assert not result.rf_transmission_executed + assert not result.flash_write_executed + + sent_data = [frame[4:4 + frame[3]] for frame in sent] + assert all(not (data and data[0] in {0x10, 0x17, 0x18, 0x1E, 0x6F}) for data in sent_data) + frequency_writes = [data for data in sent_data if data[:2] == b"\x33\x02"] + assert len(frequency_writes) == 2 + assert all(data[2] == 0x00 for data in frequency_writes) + + +def test_unexpected_nack_still_restores_frequency_and_antenna(): + nack = build_frame(0x31, b"\x33\x44\x00\x00\x00\x00\x00\x00\x00\x00") + responses = [ + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + _frequency_read(26, 26), + build_frame(0x30, b""), + build_frame(0x30, b"\x33\x00\x00\x80\x02\x00\x00\x00"), + build_frame(0x30, b"\x43\x00\x00\x80\x02\x00\x00\x00"), + nack, + _frequency_write_ack(26), + _frequency_read(26, 26), + build_frame(0x30, b"\x33\x00\x00\x80\x01\x00\x00\x00"), + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + ] + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return responses[len(sent) - 1] + + class Serial: + def reset_input_buffer(self): + pass + + with pytest.raises(RuntimeError, match="NACK_OBSERVED"): + execute_safe_controls(Serial(), _profile(), exchange, target_antenna=1) + + assert len(sent) == 10 + assert sent[-4][4:7] == b"\x33\x02\x00" + assert sent[-2][4:7] == b"\x33\x00\x00" + + +def test_optional_inventory_reports_only_count_antenna_and_channel(): + inventory = ( + build_frame(0x6C, b"\x10\x00\x00\x00\x00", address=1) + + build_frame(0x30, b"\x10\x00\x01\x00\x1B") + ) + responses = [ + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + _frequency_read(26, 26), + build_frame(0x30, b""), + build_frame(0x30, b"\x33\x00\x00\x80\x02\x00\x00\x00"), + build_frame(0x30, b"\x43\x00\x00\x80\x02\x00\x00\x00"), + _frequency_write_ack(27), + _frequency_read(27, 26), + inventory, + _frequency_write_ack(26), + _frequency_read(26, 27), + build_frame(0x30, b"\x33\x00\x00\x80\x01\x00\x00\x00"), + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + ] + sent: list[bytes] = [] + + def exchange(_ser, command): + sent.append(command) + return responses[len(sent) - 1] + + class Serial: + def reset_input_buffer(self): + pass + + result = execute_safe_controls( + Serial(), + _profile(), + exchange, + target_antenna=1, + verify_rf_channel=True, + ) + + assert result.rf_transmission_executed + assert result.inventory_tag_response_count == 1 + assert result.observed_tag_antennas == (1,) + assert result.observed_rf_channel == 27 + assert result.frequency_restored + assert result.antenna_restored + assert [frame[4:5] for frame in sent].count(b"\x10") == 1 diff --git a/tests/test_usm02_v117_validation_cli.py b/tests/test_usm02_v117_validation_cli.py new file mode 100644 index 0000000..504a9e5 --- /dev/null +++ b/tests/test_usm02_v117_validation_cli.py @@ -0,0 +1,59 @@ +import sys +from types import SimpleNamespace + +from src.utr_commands import build_frame +from src import utr_usb_sample_legacy +from src.utr_usm02_v117_validation_cli import _execute_bootstrap + + +class _FakeSerial: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def reset_input_buffer(self): + pass + + def reset_output_buffer(self): + pass + + +def test_bootstrap_only_sends_prerequisite_commands_and_builds_profile(monkeypatch): + responses = [ + build_frame(0x30, b"\x90" + b"2052USM02"), + build_frame(0x30, b""), + build_frame(0x30, b"\x43\x00\x00\x80\x01\x00\x00\x00"), + build_frame(0x30, b"\x44\x00\x00"), + build_frame(0x30, b"\x44\x01\x01"), + build_frame(0x30, b"\x44\x02\x00"), + build_frame(0x30, b"\x44\x03\x01"), + build_frame(0x30, b"\x41\x00\x1F\xDC\x81\x06\x00\x00\x00\x00\x02"), + build_frame(0x30, b"\x43\x05\x00\x01"), + build_frame(0x30, b"\x43\x05\x01\x0E"), + ] + sent: list[bytes] = [] + + def fake_communicate(_ser, command, timeout=1.0): + sent.append(command) + return responses[len(sent) - 1] + + monkeypatch.setattr(utr_usb_sample_legacy, "communicate", fake_communicate) + monkeypatch.setitem(sys.modules, "serial", SimpleNamespace(Serial=lambda **kwargs: _FakeSerial())) + + profile = _execute_bootstrap("COM6", 115200) + + assert len(sent) == 10 + assert profile.rom.raw_text == "2052USM02" + assert profile.configured_antennas == (0,) + assert profile.connected_antennas == (0, 2) + assert profile.inventory_tid_enabled + assert profile.epc_buffering_enabled + assert profile.read_cycle_completion_enabled + assert profile.antenna_switch_completion_enabled + assert profile.carrier_detect_response_enabled + + sent_data = [frame[4:4 + frame[3]] for frame in sent] + assert b"\x6F" not in sent_data + assert all(not (data and data[0] in {0x17, 0x18, 0x1E}) for data in sent_data) diff --git a/tests/test_v117_catalog.py b/tests/test_v117_catalog.py new file mode 100644 index 0000000..f16bf0d --- /dev/null +++ b/tests/test_v117_catalog.py @@ -0,0 +1,49 @@ +from collections import Counter + +from src.utr_usm02_v117_validation import build_validation_plan, summarize_plan +from src.utr_v117_catalog import COMMAND_SPECS, PlanStatus, classify_for_usm02, get_command_spec + + +def test_catalog_is_mece_54_commands_by_manual_group(): + assert len(COMMAND_SPECS) == 54 + assert len({spec.section for spec in COMMAND_SPECS}) == 54 + assert Counter(spec.section.rsplit(".", 1)[0] for spec in COMMAND_SPECS) == { + "7.3": 12, + "7.4": 31, + "7.5": 11, + } + + +def test_usm02_device_exclusions_are_explicit(): + excluded = { + spec.section + for spec in COMMAND_SPECS + if classify_for_usm02(spec, 2100) == PlanStatus.NOT_SUPPORTED_BY_DEVICE + } + assert excluded == {"7.3.6", "7.3.7", "7.4.10", "7.4.12", "7.4.26", "7.4.28"} + + +def test_rom_gates_are_explicit(): + assert classify_for_usm02(get_command_spec("7.3.12"), 2049) == PlanStatus.NOT_SUPPORTED_BY_ROM + assert classify_for_usm02(get_command_spec("7.3.12"), 2050) == PlanStatus.READY + assert classify_for_usm02(get_command_spec("7.4.14"), 2052) == PlanStatus.NOT_SUPPORTED_BY_ROM + assert classify_for_usm02(get_command_spec("7.4.14"), 2100) == PlanStatus.READY + assert classify_for_usm02(get_command_spec("7.5.11"), None) == PlanStatus.ROM_CHECK_REQUIRED + + +def test_hard_holds_and_through_command_condition(): + held = { + spec.section + for spec in COMMAND_SPECS + if classify_for_usm02(spec, 2100) == PlanStatus.NOT_EXECUTED_BY_USER_DECISION + } + assert held == {"7.3.11", "7.5.5", "7.5.6", "7.5.10"} + through = get_command_spec("7.5.11") + assert classify_for_usm02(through, 2100) == PlanStatus.CONDITION_NOT_AVAILABLE + assert classify_for_usm02(through, 2100, conditions_available=frozenset({"7.5.11"})) == PlanStatus.READY + + +def test_validation_plan_always_has_54_entries_and_summary_totals_54(): + plan = build_validation_plan(2052) + assert len(plan) == 54 + assert sum(summarize_plan(plan).values()) == 54