电厂厂区环境监测方案：POE以太网温湿度变送器长距离布线稳定性测试

Layer 1 供电层：末端电压、功耗、协商档位、温升
Layer 2 物理层：链路 up/down、协商速率、CRC 错误、丢包
Layer 3 网络层：Ping 延迟与丢包、跨区路由可达性
Layer 4 应用层：Modbus TCP 可读率、响应时间、数据准确性

import subprocess, concurrent.futures, time, csv

def ping_one(ip, count=60, timeout=2):
    try:
        r = subprocess.run(
            ["ping", "-c", str(count), "-W", str(timeout), ip],
            capture_output=True, text=True, timeout=count*(timeout+1)
        )
        out = r.stdout
        loss = out.split("% packet loss")[0].split()[-1]
        # 提取 avg rtt
        avg = out.split("min/avg/max")[1].split("/")[1] if "min/avg/max" in out else ""
        return (ip, r.returncode == 0, loss, avg)
    except Exception as e:
        return (ip, False, "err", "")

def run(hosts, out_csv="link_test.csv"):
    with concurrent.futures.ThreadPoolExecutor(max_workers=20) as ex:
        futs = {ex.submit(ping_one, h): h for h in hosts}
        rows = []
        for f in concurrent.futures.as_completed(futs):
            rows.append(f.result())
    rows.sort()
    with open(out_csv, "w", newline="") as f:
        w = csv.writer(f); w.writerow(["ip","reachable","loss_pct","avg_rtt_ms"])
        w.writerows(rows)
    for ip, ok, loss, avg in rows:
        print(f"{ip:<18} {'OK' if ok else 'FAIL':<5} loss={loss:<6} avg_rtt={avg}")

# 示例：按距离分组的厂区点位
hosts = ["10.20.10.50", "10.20.10.51", "10.20.11.60", "10.20.11.61"]
run(hosts)

import asyncio, time, csv
from pymodbus.client import AsyncModbusTcpClient

async def probe(ip, port=502, slave=1, dur=3600, period=10):
    client = AsyncModbusTcpClient(ip, port=port, timeout=3)
    await client.connect()
    log = []
    for _ in range(dur//period):
        t0 = time.monotonic()
        try:
            r = await client.read_holding_registers(0, 2, slave=slave)
            if r.isError():
                log.append([time.time(), "ERR", "", ""])
            else:
                log.append([time.time(), "OK", r.registers[0]*0.1, r.registers[1]*0.1])
        except Exception as e:
            log.append([time.time(), "EXC_"+type(e).__name__, "", ""])
        await asyncio.sleep(period)
    await client.close()
    with open(f"probe_{ip.replace('.','_')}.csv","w",newline="") as f:
        w=csv.writer(f); w.writerow(["ts","status","temp","hum"]); w.writerows(log)

# 长链路点位重点测试
asyncio.run(probe("10.20.11.60", dur=86400, period=10))