Pythonで設備CSVの状態変化だけを抽出する
10秒周期の「同じ状態の繰り返し」を、観測された変化だけのCSVへ変換する。標準ライブラリのみ・模擬データの自動テスト20件。
入力と出力
設備監視では、定周期に状態を記録すると同じ値の行が増えます。状態の変化を追うだけなら、最初の観測行と、直前の行から値が変わった行を残す方法があります。
入力（input.csv）
timestamp,status 2026-10-18T09:00:00+09:00,RUNNING 2026-10-18T09:00:10+09:00,RUNNING 2026-10-18T09:00:20+09:00,STOPPED 2026-10-18T09:00:30+09:00,STOPPED 2026-10-18T09:00:40+09:00,UNKNOWN
出力（events.csv）：UTCに統一し、最初の観測と状態の切り替わりだけを残します。
timestamp,status 2026-10-18T00:00:00+00:00,RUNNING 2026-10-18T00:00:20+00:00,STOPPED 2026-10-18T00:00:40+00:00,UNKNOWN
5件の観測から3件を残す例です。これはイベントの確定記録ではなく観測の要約です。特に最初の行は「その時点で観測した状態」であって「その時刻に切り替わった」という意味ではありません。
ダウンロード・実行方法
Python 3の標準ライブラリだけを使用します。実装コード（state_change_extract.py） と 自動テスト（test_state_change_extract.py） を同じフォルダに保存してください。
python3 state_change_extract.py input.csv events.csv python3 -B -m unittest -v test_state_change_extract.py
出力ファイルが既にある場合は上書きせず FileExistsError で終了します。元CSVも書き換えません。出力時刻はUTCのISO形式です。Excelで扱う際は表示時差に注意してください。
実装の要点
CSVのヘッダーを timestamp,status に固定し、状態値を RUNNING、STOPPED、UNKNOWN の3種類に限定します。datetime.fromisoformat() で日時を読み、UTCオフセットを必須にしたうえでUTCに正規化します。
moment = datetime.fromisoformat(stamp)
if moment.tzinfo is None or moment.utcoffset() is None:
raise ValueError("UTC offset required")
moment = moment.astimezone(timezone.utc)
if previous_time is not None and moment <= previous_time:
raise ValueError("timestamps must strictly increase")
if previous_status is None or status != previous_status:
changes.append((moment, status))異なるオフセットの同一瞬間を正しく比較し、重複時刻・時刻逆行は例外にします。csv.reader(..., strict=True)、newline=""、utf-8-sig を使用し、引用符エラー・列数不一致・BOM付きUTF-8も扱います。Path.open("x") は既存の出力を保護します。
処理は簡潔さを優先し、抽出結果をメモリに保持してから新規CSVに書き出します。大量ログには逐次書き出しと、別途バックアップ・容量制限・原子的保存の検討が必要です。
検証結果（2026年10月19日）
MacのPython unittest で20件すべて成功しました。状態の変化・同一状態・UNKNOWN保持・UTCオフセット差・1行だけの入力、重複/逆順時刻、タイムゾーン欠落、不正日時・状態・ヘッダー・列数・空欄・空データ・BOM・引用符不正・ファイルサイズ制限・新規出力・既存ファイル上書き防止・同一パス防止を模擬CSVで確認しました。
ただし、実PLC・実センサ・現場ネットワークを使った動作検証はしていません。実環境での性能・耐障害性を保証するものではありません。
この方式が適さないケース
- 欠測がある：通信断の間に状態が何度も変わっても、この方式では復元できません。欠測候補の検出を別途行い、要約後も元データを保管してください。
- 正確な切替時刻が必要：10秒ごとのポーリングなら、観測された変化の実際の発生時刻には最大でサンプリング周期程度の不確かさがあります（遅延や時計ずれがあればさらに拡大）。イベント発生源側での記録が必要です。
- 稼働率を算出したい：最後の観測より後の状態は不明です。観測期間、UNKNOWN、欠測区間を定義しないまま要約CSVから停止時間を確定しないでください。
- 信頼性の高い長期運用：事前の1MiBサイズ確認はファイルの同時更新時に厳密な上限保証になりません。途中で出力に失敗すると部分ファイルが残る可能性があります。CSVの空行は不正行として拒否しますが、入力CSVが読み込み中に変更される場合の整合性は保証しません。
- 機密ログ：設備名・作業者名などを含むファイルを無断で外部公開しないでください。出力内容も元データ由来の機密情報になり得ます。