CSV設備監視で「欠測」と「設備停止」を混同しない
模擬データによる検証例です。安全制御や実設備の故障診断に使用しないでください。
問題と考え方
設備状態を10秒ごとにCSV記録する場合、00:00:10の次が00:00:40であっても、設備が30秒停止していたとは断定できない。ログが存在しないことは「通信・記録・時刻系のどこかに欠測の疑いがある」ことしか意味しない。
定期間隔を T=10秒、判定しきい値を 1.5T=15秒 とする。隣接する観測時刻の差が15秒を超えた場合、前回観測+T から 今回観測 までを欠測候補とする。00:00:10→00:00:40なら欠測候補は20秒。停止時間に算入しない。
CSV例
timestamp,status
2026-10-18T00:00:00+09:00,RUNNING
2026-10-18T00:00:10+09:00,RUNNING
2026-10-18T00:00:40+09:00,STOPPEDPython標準ライブラリでの最小実装
import csv
from datetime import datetime, timezone, timedelta
def gaps(csv_path, interval=10, factor=1.5):
if interval <= 0 or factor <= 1:
raise ValueError("invalid settings")
result, previous = [], None
with open(csv_path, encoding="utf-8-sig", newline="") as f:
rows = csv.DictReader(f)
if not rows.fieldnames or not {"timestamp", "status"} <= set(rows.fieldnames):
raise ValueError("timestamp,status required")
for line, row in enumerate(rows, 2):
t = datetime.fromisoformat(row["timestamp"])
if t.tzinfo is None or t.utcoffset() is None:
raise ValueError(f"line {line}: UTC offset required")
if row["status"] not in {"RUNNING", "STOPPED", "UNKNOWN"}:
raise ValueError(f"line {line}: invalid status")
t = t.astimezone(timezone.utc)
if previous is not None:
delta = (t - previous).total_seconds()
if delta <= 0:
raise ValueError(f"line {line}: timestamps not increasing")
if delta > interval * factor:
result.append((previous + timedelta(seconds=interval), t, delta - interval))
previous = t
return result実行例： print(gaps("sample_heartbeat.csv"))。この短い実装は説明用で、業務運用には入力サイズ上限や監査ログなどの追加対策が必要です。
検証
2026-10-18、Pythonのunittestで8/8件成功：20秒の欠測、正常間隔、異なるUTCオフセット、重複時刻、オフセットなし、状態値不正、ヘッダー不足、設定値不正。模擬CSVのみで、実機検証はしていない。
注意
これは欠測候補の推定であり、実際の通信断・設備停止・センサ故障の原因判定ではない。最終行より後の欠測は観測終了時刻がないため扱わない。時刻同期と計測仕様を確認し、安全制御には利用しない。機密設備ログを外部サービスに無断送信しない。
一次資料
- Python公式 csv: https://docs.python.org/3/library/csv.html
- Python公式 datetime: https://docs.python.org/3/library/datetime.html
- Python公式 unittest: https://docs.python.org/3/library/unittest.html