Extract output power IO helpers

README.md

@@ -143,6 +143,7 @@ UTR_USB_Python_CodeX/
 │  ├─ utr_output_power.py
 │  ├─ utr_output_power_change.py
 │  ├─ utr_output_power_inventory_integration.py
+│  ├─ utr_output_power_io.py
 │  ├─ utr_output_power_readout.py
 │  ├─ utr_output_power_temporary_change.py
 │  ├─ utr_output_power_temporary_change_cli.py
@@ -175,9 +176,10 @@ UTR_USB_Python_CodeX/
 | `src/utr_output_power.py` | 送信出力値の dBm / raw値 / little-endian bytes 変換補助です。 |
 | `src/utr_output_power_change.py` | 送信出力設定書き込みフレームの計画補助です。 |
 | `src/utr_output_power_inventory_integration.py` | Inventory前に送信出力一時変更を提案できるか判定します。 |
+| `src/utr_output_power_io.py` | 送信出力設定の読み取りと、作成済み送信出力フレームの送信確認を行う共通I/Oです。 |
 | `src/utr_output_power_readout.py` | 送信出力読み取りレスポンスからキャリア関連時間を解析します。 |
 | `src/utr_output_power_temporary_change.py` | 一時変更用フレームと復元用フレームをセットで作成します。 |
-| `src/utr_output_power_temporary_change_cli.py` | 送信出力一時変更単体確認CLIです。現状は共通ヘルパーも含むため削除しません。 |
+| `src/utr_output_power_temporary_change_cli.py` | 送信出力一時変更単体確認CLIです。共通I/Oは `utr_output_power_io.py` に分離済みです。 |
 | `src/utr_privacy.py` | PC+UIIの画面表示・保存時マスク補助です。 |
 | `src/utr_protocol.py` | フレーム、SUM、NACKなどのプロトコル補助です。 |
 | `src/utr_reader_settings.py` | リーダ設定表示補助です。 |
@@ -256,7 +258,7 @@ inventory_results.json
 
 - 8CH Ctrl+C時の最終復元成功ログを取得する。
 - 送信出力一時変更中のCtrl+C時の復元確認を行う。
-- `src/utr_output_power_temporary_change_cli.py` から共通ヘルパーを分離し、CLI整理を検討する。
+- `src/utr_output_power_temporary_change_cli.py` を単体確認CLIとして残すか、標準入口へ統合して削除するか検討する。
 - `src/utr_usb_sample_legacy.py` の依存を段階的に解消し、将来的な削除を検討する。
 - READMEが再び肥大化した場合は、詳細説明をdocsへ分離する。
 

src/utr_output_power_io.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""送信出力設定の実機I/O共通処理。
+
+このモジュールは、送信出力の読み取りと、送信出力設定フレームの送信確認だけを扱います。
+
+重要:
+- FLASHは変更しません。
+- フレームの生成は行いません。
+- 送信するフレームは呼び出し元で作成済みのものだけです。
+- ACK/NACK判定と表示を共通化し、CLIファイルへ共通ヘルパーを置かないための分離です。
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+import serial
+
+try:
+    from src.utr_reader_settings import format_output_power_setting, parse_output_power_setting_response
+    from src.utr_usb_sample import ACK, COMMANDS, NACK, STX, communicate, print_nack_message
+except ModuleNotFoundError:
+    from utr_reader_settings import format_output_power_setting, parse_output_power_setting_response
+    from utr_usb_sample import ACK, COMMANDS, NACK, STX, communicate, print_nack_message
+
+
+def _is_ack(response: bytes) -> bool:
+    """ACKレスポンスかどうかを確認します。"""
+    return bool(re.match(STX + b"." + ACK, response))
+
+
+def _is_nack(response: bytes) -> bool:
+    """NACKレスポンスかどうかを確認します。"""
+    return bool(re.match(STX + b"." + NACK, response))
+
+
+def _read_current_output_power_setting(ser: serial.Serial) -> dict[str, Any]:
+    """現在の送信出力設定を読み取り、画面表示用に整形して出力します。"""
+    response = communicate(ser, COMMANDS["UHF_READ_OUTPUT_POWER"])
+    if _is_ack(response):
+        parsed = parse_output_power_setting_response(response)
+        print("")
+        print("=== 現在の送信出力設定 ===")
+        for line in format_output_power_setting(parsed):
+            print(line)
+        return parsed
+
+    if _is_nack(response):
+        print_nack_message(response)
+        raise RuntimeError("送信出力設定の読み取りでNACKを受信しました。")
+
+    print("Raw:", response.hex().upper())
+    raise RuntimeError("送信出力設定の読み取りでACK/NACKがありません。")
+
+
+def _send_output_power_frame(ser: serial.Serial, frame: bytes, label: str) -> None:
+    """送信出力設定フレームを送信し、ACK/NACKを確認します。"""
+    print("")
+    print(f"{label}を送信します。")
+    print("FLASHは変更しません。")
+    print("送信フレーム:", frame.hex(" ").upper())
+
+    response = communicate(ser, frame)
+    if _is_ack(response):
+        print(f"{label}がACKで完了しました。")
+        return
+
+    if _is_nack(response):
+        print_nack_message(response)
+        raise RuntimeError(f"{label}でNACKを受信しました。")
+
+    print("Raw:", response.hex().upper())
+    raise RuntimeError(f"{label}でACK/NACKがありません。")

src/utr_output_power_temporary_change_cli.py

@@ -18,18 +18,17 @@
 
 import re
 import sys
-from typing import Any
 
 import serial
 from serial.tools import list_ports
 
 try:
+    from src.utr_output_power_io import _read_current_output_power_setting, _send_output_power_frame
     from src.utr_output_power_temporary_change import (
         OutputPowerTemporaryChangePlans,
         build_output_power_temporary_change_plans,
         format_output_power_temporary_change_plans,
     )
-    from src.utr_reader_settings import format_output_power_setting, parse_output_power_setting_response
     from src.utr_usb_sample import (
         ACK,
         COMMANDS,
@@ -48,12 +47,12 @@
         prompt_for_port_name,
     )
 except ModuleNotFoundError:
+    from utr_output_power_io import _read_current_output_power_setting, _send_output_power_frame
     from utr_output_power_temporary_change import (
         OutputPowerTemporaryChangePlans,
         build_output_power_temporary_change_plans,
         format_output_power_temporary_change_plans,
     )
-    from utr_reader_settings import format_output_power_setting, parse_output_power_setting_response
     from utr_usb_sample import (
         ACK,
         COMMANDS,
@@ -85,45 +84,6 @@ def _is_nack(response: bytes) -> bool:
     return bool(re.match(STX + b"." + NACK, response))
 
 
-def _read_current_output_power_setting(ser: serial.Serial) -> dict[str, Any]:
-    """現在の送信出力設定を読み取ります。"""
-    response = communicate(ser, COMMANDS["UHF_READ_OUTPUT_POWER"])
-    if _is_ack(response):
-        parsed = parse_output_power_setting_response(response)
-        print("")
-        print("=== 現在の送信出力設定 ===")
-        for line in format_output_power_setting(parsed):
-            print(line)
-        return parsed
-
-    if _is_nack(response):
-        print_nack_message(response)
-        raise RuntimeError("送信出力設定の読み取りでNACKを受信しました。")
-
-    print("Raw:", response.hex().upper())
-    raise RuntimeError("送信出力設定の読み取りでACK/NACKがありません。")
-
-
-def _send_output_power_frame(ser: serial.Serial, frame: bytes, label: str) -> None:
-    """送信出力設定フレームを送信し、ACK/NACKを確認します。"""
-    print("")
-    print(f"{label}を送信します。")
-    print("FLASHは変更しません。")
-    print("送信フレーム:", frame.hex(" ").upper())
-
-    response = communicate(ser, frame)
-    if _is_ack(response):
-        print(f"{label}がACKで完了しました。")
-        return
-
-    if _is_nack(response):
-        print_nack_message(response)
-        raise RuntimeError(f"{label}でNACKを受信しました。")
-
-    print("Raw:", response.hex().upper())
-    raise RuntimeError(f"{label}でACK/NACKがありません。")
-
-
 def _verify_supported_reader(ser: serial.Serial) -> None:
     """ROMバージョンを読み取り、対象機種か確認します。"""
     response = communicate(ser, COMMANDS["ROM_VERSION_CHECK"])

src/utr_usb_inventory_with_output_power.py

@@ -36,15 +36,12 @@
         evaluate_output_power_temporary_change_availability,
         format_output_power_temporary_change_offer,
     )
+    from src.utr_output_power_io import _read_current_output_power_setting, _send_output_power_frame
     from src.utr_output_power_temporary_change import (
         OutputPowerTemporaryChangePlans,
         build_output_power_temporary_change_plans,
         format_output_power_temporary_change_plans,
     )
-    from src.utr_output_power_temporary_change_cli import (
-        _read_current_output_power_setting,
-        _send_output_power_frame,
-    )
     from src.utr_privacy import get_or_create_masked_pc_uii_id
     from src.utr_usb_sample import (
         ACK,
@@ -81,15 +78,12 @@
         evaluate_output_power_temporary_change_availability,
         format_output_power_temporary_change_offer,
     )
+    from utr_output_power_io import _read_current_output_power_setting, _send_output_power_frame
     from utr_output_power_temporary_change import (
         OutputPowerTemporaryChangePlans,
         build_output_power_temporary_change_plans,
         format_output_power_temporary_change_plans,
     )
-    from utr_output_power_temporary_change_cli import (
-        _read_current_output_power_setting,
-        _send_output_power_frame,
-    )
     from utr_privacy import get_or_create_masked_pc_uii_id
     from utr_usb_sample import (
         ACK,