# sim_device.py
from pymodbus.server import ModbusTcpServer
from pymodbus.simulator import DataType, SimData, SimDevice
 
TEMP_ADDR = 0        # 温度(0.1度単位。215 = 21.5度)
SETPOINT_ADDR = 10   # 設定値(書き込み対象)
 
 
def build_device() -> SimDevice:
    return SimDevice(
        id=1,
        simdata=[
            SimData(address=TEMP_ADDR, values=[215, 480],
                    datatype=DataType.REGISTERS),
            SimData(address=SETPOINT_ADDR, values=[200],
                    datatype=DataType.REGISTERS),
        ],
    )
 
 
async def start(port: int = 5020) -> ModbusTcpServer:
    server = ModbusTcpServer(build_device(), address=("127.0.0.1", port))
    await server.serve_forever(background=True)
    return server

# safe_client.py
from pymodbus.client import AsyncModbusTcpClient
 
ALLOWED_HOSTS = {"127.0.0.1", "localhost"}
 
 
class WriteBlocked(RuntimeError):
    pass
 
 
class SafeClient:
    def __init__(self, host, port=502, *, allow_write=False):
        if host not in ALLOWED_HOSTS:
            raise ValueError(f"接続先 {host} は許可リストにありません")
        self._client = AsyncModbusTcpClient(
            host, port=port, timeout=1, retries=1)
        self._allow_write = allow_write
 
    async def connect(self):
        return await self._client.connect()
 
    def close(self):
        self._client.close()
 
    async def read_holding(self, address, count=1, device_id=1):
        rr = await self._client.read_holding_registers(
            address, count=count, device_id=device_id)
        if rr.isError():
            raise IOError(f"read 失敗: {rr}")
        return rr.registers
 
    async def write_register(self, address, value, device_id=1):
        if not self._allow_write:
            raise WriteBlocked("allow_write=True のときだけ書き込める")
        wr = await self._client.write_register(
            address, value, device_id=device_id)
        if wr.isError():
            raise IOError(f"write 失敗: {wr}")

# test_safe_client.py
import pytest
import pytest_asyncio
from safe_client import SafeClient, WriteBlocked
from sim_device import SETPOINT_ADDR, start
 
 
@pytest_asyncio.fixture
async def server():
    s = await start(5021)
    yield s
    await s.shutdown()
 
 
@pytest.mark.asyncio
async def test_read(server):
    c = SafeClient("127.0.0.1", 5021)
    await c.connect()
    assert await c.read_holding(0, 2) == [215, 480]
    c.close()
 
 
@pytest.mark.asyncio
async def test_write_is_blocked_by_default(server):
    c = SafeClient("127.0.0.1", 5021)
    await c.connect()
    with pytest.raises(WriteBlocked):
        await c.write_register(SETPOINT_ADDR, 250)
    c.close()
 
 
@pytest.mark.asyncio
async def test_write_roundtrip_on_simulator(server):
    c = SafeClient("127.0.0.1", 5021, allow_write=True)
    await c.connect()
    await c.write_register(SETPOINT_ADDR, 250)
    assert await c.read_holding(SETPOINT_ADDR) == [250]
    c.close()
 
 
@pytest.mark.asyncio
async def test_illegal_address_is_reported(server):
    c = SafeClient("127.0.0.1", 5021)
    await c.connect()
    with pytest.raises(IOError, match="exception_code=2"):
        await c.read_holding(500)
    c.close()
 
 
def test_real_host_is_rejected():
    with pytest.raises(ValueError):
        SafeClient("192.168.0.10")

## Modbus通信のルール
- 接続先は `127.0.0.1` の模擬機器(`sim_device.py`)のみ。実機のIPを書かない
- 通信は必ず `SafeClient` 経由。`allow_write=True` はテストの中だけ
- 変更後は `python -m pytest -q` を実行し、結果を読んでから完了と報告する
- 実機への接続は人間が手元で行う。Claudeは実機向けのスクリプトを実行しない

{
  "permissions": {
    "deny": [
      "Read(./.env*)",
      "Read(./config/real_device.toml)",
      "Bash(python scripts/poll_real_device.py *)"
    ]
  }
}