tkinter で作る現場用監視アプリ — 24 時間運用に耐える設計パターン

1. はじめに — 「動くデモ」と「現場で動くアプリ」の差

tkinter で監視 UI を作ることは、それほど難しくありません。Web で「Python tkinter リアルタイム」と検索すれば、Label と after() でセンサ値を更新するシンプルな例が無数に見つかります。動かしてみると、ちゃんと数値が更新されて、デモとしては成立します。

しかし、それを実際の製造現場に持っていって 24 時間連続稼働させると、ほぼ確実に問題が起きます。

• 翌朝には UI が固まっている
• 1 週間後にはメモリ使用量が 10 倍になっている
• たまに通信切断が起きると、それ以降ずっと値が更新されない
• ログファイルが GB サイズに膨張している
• 夜間に PC がスリープして、朝になっても通信が再開しない

これらの問題は、最初の 1 時間では見えません。設計段階で対処を組み込んでおかないと、現場で「なんか動かない」と言われた時に原因究明と再現が極めて困難になります。

本記事では、tkinter で作る監視アプリを「24 時間連続稼働できる」レベルまで設計するためのパターンを、アーキテクチャ・実装・運用の 3 層に分けて解説します。記事末尾には、自宅環境で再現可能な完全動作サンプルコードも掲載します。

2. 設計要件の整理

まず「現場用監視アプリ」に求められる要件を整理しておきます。要件を曖昧にしたまま実装に入ると、後で「あれが足りない」「これが過剰」となって手戻りします。

機能要件

• 設備（PLC、センサー、計測器）からのデータを定期取得
• 取得値を画面表示（数値・グラフ・ステータス）
• しきい値超過の検知と警告（画面、ログ、可能であればメール / Slack）
• 取得値のログ記録（CSV、データベース、または専用ロガー）
• 現場オペレーターによる基本操作（開始 / 停止、設定値変更）

非機能要件（こちらが本題）

• 連続稼働: 24 時間 × 365 日（少なくとも数週間〜数ヶ月）動き続ける
• 自動復旧: 通信切断・例外発生でも、人手なしに復旧する
• UI フリーズ防止: 通信遅延や例外で画面が固まらない
• メモリ安定: 1 日経っても 1 週間経っても、メモリ使用量が一定範囲に収まる
• ログ管理: ログファイルが無限に肥大化しない
• 誤操作耐性: 現場オペレーターが意図せず重要な設定を変えてしまわない
• 復旧の容易さ: 何かあった時、ログを見れば原因と復旧方法が分かる

要件の優先順位

機能を増やすことよりも、非機能の確実性を優先するのが鉄則です。

• 落ちないこと > 機能の豊富さ
• 復旧できること > 完全に防ぐこと（防ぎきれない事象がある前提で復旧設計）
• ログを残すこと > リアルタイム性（ログがあれば後で原因が追える）
• 操作ミス耐性 > 機能性（現場では「とりあえずクリック」が起きる）

3. 全体アーキテクチャ

監視アプリの内部構造を、3 つの層に分けて整理します。これは tkinter に限らず、長時間動くアプリの基本パターンです。

3 層構造

┌──────────────────────────────────────┐
│  UI 層 (tkinter メインスレッド)        │
│  - ウィジェット描画                     │
│  - ユーザー入力                         │
│  - after() で定期的にキューを取り出し  │
└────────────┬─────────────────────────┘
             │ queue.Queue (UI へ)
             │
┌────────────┴─────────────────────────┐
│  通信層 (バックグラウンドスレッド)     │
│  - PLC / センサーとの通信              │
│  - 切断時の再接続処理                   │
│  - 取得値を queue に投入                │
└────────────┬─────────────────────────┘
             │
┌────────────┴─────────────────────────┐
│  ロギング層 (共通)                     │
│  - logging モジュール                  │
│  - ローテーション                      │
│  - 例外の捕捉と記録                    │
└──────────────────────────────────────┘

原則

• tkinter ウィジェットへのアクセスはメインスレッドからのみ: 別スレッドから直接ウィジェットを操作すると、未定義動作（クラッシュ、ハング）が起きる
• スレッド間通信は queue.Queue で: 共有変数を直接やり取りすると競合状態（race condition）の温床に
• 共有状態は最小限: 複数スレッドから読み書きする変数は、できるだけ少なく、できるだけ queue 経由で
• 例外は各スレッドの最上位で捕捉: 通信スレッドで未捕捉例外が出るとスレッドが死に、UI からは「動いているように見えるが実は止まっている」状態に

4. tkinter の基本と after() パターン

tkinter で「定期的に画面を更新する」基本パターンを押さえます。多くの監視アプリは、これだけで実用に耐えます。

after() の基本

widget.after(ms, callback) は、指定したミリ秒後にコールバックをメインスレッドから呼び出します。コールバックの中で再度 after() を呼ぶことで、定期実行ループが作れます。

import tkinter as tk
from datetime import datetime

root = tk.Tk()
root.title('時計')

label = tk.Label(root, font=('Arial', 24))
label.pack(padx=20, pady=20)

def update_clock():
    label.config(text=datetime.now().strftime('%H:%M:%S'))
    root.after(1000, update_clock)  # 1 秒後に再実行

update_clock()
root.mainloop()

この方式は threading を使わないため、スレッド間競合が原理的に起きません。シンプルな監視アプリならこれで十分です。

after() の落とし穴

• コールバック内で時間のかかる処理をすると UI がフリーズする: after() はメインスレッドで実行されるので、3 秒かかる処理を入れると 3 秒間 UI が無反応になる
• 例外を捕捉しないとループが止まる: コールバック内で例外が発生し再帰呼び出しに到達しないと、二度と実行されない
• 大量の after() 予約: 短い間隔で別の after() を予約しすぎると処理が滞る

after() と例外処理の定型

def update_loop():
    try:
        # メイン処理（短時間で終わるもの）
        update_display()
    except Exception as e:
        logger.exception('update_loop error')
    finally:
        root.after(1000, update_loop)  # 例外があっても次回を予約

finally で必ず次回を予約するのがポイント。これで何があってもループは継続します。

5. threading との使い分け — キューによるスレッド間通信

通信処理はネットワーク待ち・タイムアウトで時間がかかります。これを after() の中で直接やると UI が固まるので、別スレッドに切り出します。

通信スレッド + キューの基本構造

import threading
import queue
import time
import logging

logger = logging.getLogger(__name__)
data_queue = queue.Queue(maxsize=100)
stop_event = threading.Event()

def communication_worker():
    while not stop_event.is_set():
        try:
            value = read_from_plc()  # 時間のかかる通信処理
            try:
                data_queue.put_nowait(value)
            except queue.Full:
                # キューが満杯なら古いデータを捨てて入れ直す
                try:
                    data_queue.get_nowait()
                except queue.Empty:
                    pass
                data_queue.put_nowait(value)
        except Exception as e:
            logger.exception('communication error')
            time.sleep(5)  # エラー時は少し待つ
        else:
            time.sleep(1)  # 正常時は 1 秒間隔

# 起動
worker = threading.Thread(target=communication_worker, daemon=True)
worker.start()

# UI 側で キューを取り出す
def update_display():
    try:
        while True:
            value = data_queue.get_nowait()
            label.config(text=str(value))
    except queue.Empty:
        pass
    root.after(100, update_display)

キュー設計のポイント

• maxsize を必ず設定: 無制限だと、UI が遅延した時にキューが膨らみメモリリークの温床に
• 満杯時の処理を決めておく: 古いデータを捨てる、新しいデータを捨てる、ブロックする——監視アプリでは「最新だけが大事」なので古いデータを捨てる方が望ましい
• UI 側は get_nowait でブロックしない: get でブロックすると UI が固まる
• UI 側で複数取り出す: 1 回の after() でキュー内の溜まったデータをまとめて取り出すと、UI 側の処理頻度を下げられる

停止の仕組み

threading.Event をスレッド間で共有し、停止フラグとして使います。daemon=True にしておくと、メインプロセスが終了した時にスレッドも自動で終わりますが、明示的に stop_event.set() してから join() する方が確実です。

6. メモリリーク対策

1 時間動かしただけでは気づかず、24 時間動かして初めて顕在化するのがメモリリークです。事前に対策を組み込んでおかないと、運用開始後に発覚し、原因究明にかなりの時間を要します。

典型的な原因

• リストへの追加し続け: ログ表示用のリストに append() し続けて削除しない
• Matplotlib の figure を close() していない: グラフを再描画するたびにメモリ消費
• tkinter ウィジェットを destroy() せず追加し続ける: 動的にラベルを増やす UI で起きやすい
• 循環参照: ガーベジコレクタの対象になりにくいパターン
• ロガーがオブジェクトを保持: logger.exception() が例外オブジェクトを保持し続けるケース
• キャッシュの肥大化: functools.lru_cache を maxsize 無制限で使う、自前のキャッシュ辞書が無限に増える

検知方法

• tracemalloc モジュール: Python 標準。プログラム中の特定タイミングでメモリ確保のスナップショットを取得し、差分を分析できる
• memory_profiler パッケージ: 関数単位でメモリ使用量を記録
• OS のリソースモニタ: タスクマネージャや psutil で RSS（実メモリ使用量）を継続観測
• 24 時間ベンチマーク: 実際に 24 時間動かして、開始時と 24 時間後のメモリ使用量を比較。これが最終確認

予防のパターン

• 履歴データは collections.deque(maxlen=N) で上限を持つ: 古いものから自動削除
• 表示用リストは定期的にクリア: 「直近 100 件のみ表示」のような上限設計
• Matplotlib は plt.close(fig) を必ず呼ぶ
• 使い終わったオブジェクトは del または None 代入
• ログはローテーション必須（後述）
• 定期再起動の許容: どうしてもリークが避けられない場合、夜間に自動再起動するのも一つの解

具体例: deque で履歴を持つ

from collections import deque

# 直近 1000 件だけ保持
history = deque(maxlen=1000)

def on_new_data(value):
    history.append(value)
    # history の長さは自動で 1000 を超えない

7. 通信切断 → 自動復旧

製造現場では、何らかの理由でネットワークが瞬間的に切れることが日常的にあります。スイッチの再起動、ケーブル抜き差し、PLC 側の保守、PC のスリープ復帰——いずれも通信エラーの原因です。

切断の典型シナリオ

• ネットワークケーブルの瞬間的な抜き差し（数秒）
• L2 / L3 スイッチの再起動（数十秒〜数分）
• PLC 側の TCP セッションタイムアウト（数分〜数十分）
• 機器側の再起動（数分）
• PC のスリープ・休止（数分〜数時間）
• Windows Update による再起動（数十分）

自動復旧の設計原則

• 切断を例外として捉える: 通信処理を try / except で囲み、エラー時に再接続フローへ
• 指数バックオフ: 連続失敗時、再試行間隔を徐々に長くする（1 秒 → 2 秒 → 4 秒 → 上限 60 秒）。即時連打すると相手機器に負荷をかけ、ログも肥大化する
• サーキットブレーカー: 連続失敗が一定回数に達したら、しばらく試行を停止して状態を「異常」に。一定時間後に「半開」状態で 1 回だけ試行
• 復旧時の通知: オペレーターに「通信が復旧しました」を画面表示する。これがないと、復旧したのか手動操作が必要なのか判断できない

再接続フローのコード例

import time
import logging

logger = logging.getLogger(__name__)

def communication_worker(client, stop_event, ui_queue):
    backoff = 1
    while not stop_event.is_set():
        try:
            if not client.is_connected():
                logger.info('reconnecting...')
                client.connect()
                logger.info('reconnected')
                ui_queue.put({'type': 'status', 'value': 'connected'})
                backoff = 1  # 復旧したらバックオフをリセット

            value = client.read()
            ui_queue.put({'type': 'data', 'value': value})
            time.sleep(1)

        except (ConnectionError, TimeoutError) as e:
            logger.warning(f'communication error: {e}, retry in {backoff}s')
            ui_queue.put({'type': 'status', 'value': 'disconnected'})
            try:
                client.close()
            except Exception:
                pass
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)  # 指数バックオフ、上限 60 秒

        except Exception as e:
            logger.exception('unexpected error')
            time.sleep(5)

8. ロギング設計

ログは、トラブル発生時の唯一の手がかりです。「なんとなく動かない」を「16:23 に通信が切れて、16:25 に復旧した」と言えるようにするのがロギングの役割です。

logging モジュールの基本設定

import logging
from logging.handlers import TimedRotatingFileHandler

def setup_logger():
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)

    formatter = logging.Formatter(
        '%(asctime)s [%(levelname)s] %(name)s: %(message)s'
    )

    # 日次ローテーション、30 日分保持
    file_handler = TimedRotatingFileHandler(
        'log/app.log',
        when='midnight',
        backupCount=30,
        encoding='utf-8',
    )
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)

    # 標準エラーにも出す（開発時に便利）
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)

    return logger

ログレベルの使い分け

• DEBUG: 開発時のみ。本番では出さない。詳細な変数値、関数の入出力
• INFO: 正常動作の記録。「起動」「接続成功」「データ取得 N 件」など、後から動作を再構成できる粒度
• WARNING: 異常だが回復可能。「通信切断、再接続中」「しきい値超過」
• ERROR: 個別処理が失敗。「データ書き込み失敗」
• CRITICAL: アプリ全体の継続が困難。基本的に最後の一回だけ

例外ログのコツ

例外を捕捉した時は logger.exception() を使うと、Traceback が自動で記録されます。

try:
    do_something()
except Exception:
    logger.exception('failed to do_something')
    # exception() は ERROR レベルで Traceback 込みで出力

ログのローテーション

• RotatingFileHandler: ファイルサイズベース（例: 10MB ごとに新規）。バースト的にログが増える環境向け
• TimedRotatingFileHandler: 時刻ベース（日次、毎時など）。一定ペースでログが出る環境向け。監視アプリは通常こちら

ローテーションを設定しないと、半年後に「ログファイルが 50GB になっていてディスクが満杯」という事態を招きます。これは複数の現場で実際に起きている事故です。

9. 現場オペレーター向け UI 設計

UI 設計は、エンジニアにとって慣れない領域かもしれません。しかし「現場で使えるか」を分けるのは、機能の数ではなく UI の作りです。

大原則: ユーザーは IT 専門家ではない

現場オペレーターの方々は、設備のプロですが、PC 操作のプロではありません。「ログを開いて確認してください」「コマンドプロンプトを開いて」は通じない世界です。「画面に書いてある通りにすればいい」状態を目指します。

視認性

• フォントサイズ大きめ: 数値は最低 24pt、できれば 36pt 以上。離れた位置から見えるのが要件
• ステータス色: 赤 = 異常、黄 = 注意、緑 = 正常、グレー = 待機。色覚多様性に配慮するなら形やパターンも併用
• 重要度の階層: 「設備が停止しているかどうか」が一目で分かるのが最優先
• 余白: ぎっしり詰め込まない。離して配置する方が見やすい

誤操作防止

• 確認ダイアログ: 重要な操作（停止、設定変更、書き込み）には必ず確認
• 無効化されたボタン: 文脈で押せない操作はグレーアウト。押せるが何も起きないより、押せないことを示す
• 取り消し: 設定変更の Undo を用意
• 権限分離: 「閲覧モード」と「設定モード」を分け、設定モードはパスワード（簡易でも）

状態通知の階層

監視アプリの画面は、上から下へと「重要度の高い情報 → 詳細」の順に並べると見やすくなります。

1. 最上段: 全体ステータス（正常 / 警告 / 異常）を大きく
2. 中段: 主要な計測値（フォントサイズ大）
3. 下段: ログ表示エリア、過去のグラフ、詳細情報
4. 右下など隅: 通信状態、稼働時間など補助情報

10. 実装サンプル — 自宅で再現できる監視アプリ

これまでのパターンを統合した、動作する監視アプリのコードです。実機がなくても、Modbus シミュレータを使えば自宅環境で完全に再現できます。

事前準備

• Python 3.10+
• pip install pymodbus
• Modbus シミュレータ（ModSim、Modbus Slave、または pymodbus.server）

サンプルコード（main.py）

import tkinter as tk
from tkinter import ttk
import threading
import queue
import time
import logging
from collections import deque
from logging.handlers import TimedRotatingFileHandler
from pymodbus.client import ModbusTcpClient

# ============= ロギング設定 =============
def setup_logger():
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    formatter = logging.Formatter(
        '%(asctime)s [%(levelname)s] %(message)s'
    )
    fh = TimedRotatingFileHandler(
        'app.log', when='midnight', backupCount=30, encoding='utf-8'
    )
    fh.setFormatter(formatter)
    logger.addHandler(fh)
    sh = logging.StreamHandler()
    sh.setFormatter(formatter)
    logger.addHandler(sh)
    return logger

logger = setup_logger()

# ============= 通信スレッド =============
HOST = '127.0.0.1'
PORT = 5020
data_queue = queue.Queue(maxsize=100)
stop_event = threading.Event()

def communication_worker():
    client = ModbusTcpClient(HOST, port=PORT)
    backoff = 1
    while not stop_event.is_set():
        try:
            if not client.connected:
                logger.info(f'connecting to {HOST}:{PORT}')
                if not client.connect():
                    raise ConnectionError('connect failed')
                logger.info('connected')
                data_queue.put({'type': 'status', 'value': 'connected'})
                backoff = 1

            res = client.read_holding_registers(address=0, count=1, slave=1)
            if res.isError():
                raise IOError(f'modbus error: {res}')
            value = res.registers[0]
            data_queue.put({'type': 'data', 'value': value})
            time.sleep(1)

        except (ConnectionError, IOError, TimeoutError) as e:
            logger.warning(f'communication error: {e}')
            data_queue.put({'type': 'status', 'value': 'disconnected'})
            try:
                client.close()
            except Exception:
                pass
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)
        except Exception:
            logger.exception('unexpected error')
            time.sleep(5)

    client.close()
    logger.info('communication worker stopped')

# ============= UI =============
THRESHOLD = 1000
history = deque(maxlen=100)

class MonitorApp:
    def __init__(self, root):
        self.root = root
        self.root.title('現場監視ミニアプリ')
        self.root.geometry('500x400')

        # ステータス表示
        self.status_var = tk.StringVar(value='待機中')
        self.status_label = tk.Label(
            root, textvariable=self.status_var,
            font=('Arial', 28, 'bold'), bg='gray', fg='white',
            width=20, pady=10,
        )
        self.status_label.pack(fill='x', padx=10, pady=10)

        # 計測値表示
        self.value_var = tk.StringVar(value='---')
        tk.Label(root, text='現在値', font=('Arial', 12)).pack()
        tk.Label(
            root, textvariable=self.value_var, font=('Arial', 36)
        ).pack(pady=10)

        # 接続状態
        self.conn_var = tk.StringVar(value='未接続')
        tk.Label(
            root, textvariable=self.conn_var,
            font=('Arial', 10), fg='gray',
        ).pack()

        # 終了処理
        root.protocol('WM_DELETE_WINDOW', self.on_close)

        # 定期更新ループ
        self.update_loop()

    def update_loop(self):
        try:
            # キューに溜まったメッセージをすべて処理
            while True:
                msg = data_queue.get_nowait()
                if msg['type'] == 'data':
                    value = msg['value']
                    history.append(value)
                    self.value_var.set(str(value))
                    if value > THRESHOLD:
                        self.set_status('警告', 'red')
                    else:
                        self.set_status('正常', 'green')
                elif msg['type'] == 'status':
                    if msg['value'] == 'connected':
                        self.conn_var.set('接続中')
                    else:
                        self.conn_var.set('再接続中...')
                        self.set_status('通信断', 'orange')
        except queue.Empty:
            pass
        except Exception:
            logger.exception('update_loop error')
        finally:
            self.root.after(200, self.update_loop)

    def set_status(self, text, color):
        self.status_var.set(text)
        self.status_label.config(bg=color)

    def on_close(self):
        logger.info('closing app')
        stop_event.set()
        self.root.after(500, self.root.destroy)

# ============= 起動 =============
if __name__ == '__main__':
    worker = threading.Thread(target=communication_worker, daemon=True)
    worker.start()

    root = tk.Tk()
    app = MonitorApp(root)
    root.mainloop()

動作確認

1. 別ターミナルで Modbus シミュレータを起動（pymodbus.server でも可）
2. python main.py でアプリ起動
3. シミュレータ側でレジスタ 0 番地の値を変更 → アプリの表示が変わる
4. シミュレータを停止 → ステータスが「通信断（オレンジ）」に変わる
5. シミュレータを再起動 → 数秒〜十数秒後に「正常（緑）」に自動復旧
6. レジスタ値を 1000 超に → ステータスが「警告（赤）」に

このサンプルが満たしている要件

• ✅ after() ベースの UI 更新（メインスレッドで安全）
• ✅ 通信は別スレッド、queue で UI とやり取り
• ✅ キューに maxsize あり、メモリ無限増殖を防止
• ✅ 履歴は deque(maxlen=N) で上限あり
• ✅ 通信切断 → 指数バックオフで自動再接続
• ✅ ステータス色（赤・オレンジ・緑・グレー）
• ✅ 大きなフォントサイズで視認性確保
• ✅ ロガー（日次ローテーション、30 日保持）
• ✅ 例外で UI が止まらない（finally で次回予約）
• ✅ 終了時に通信スレッドを停止

11. 24 時間連続稼働チェックリスト

本番投入前に必ず確認すべき項目です。一つでも欠けると、運用開始後に痛い目を見ます。

アプリの実装

• 24 時間動かしてもメモリ使用量（RSS）が増え続けない
• ログがローテートし、ディスクを食い続けない
• 通信切断が発生しても自動で復旧する
• 各スレッドの最上位で例外を捕捉している
• キュー類に maxsize があり、UI 遅延でも溢れない
• 履歴データが deque や上限付きリストで管理されている
• 例外でも UI ループが止まらない（finally で再予約）
• アプリ起動・停止・例外がすべてログに残る

Windows OS 設定

• 電源プランを「高パフォーマンス」または「バランス」（スリープなし）
• スクリーンセーバーを OFF
• Windows Update の自動再起動を抑制（業務時間外に固定）
• USB セレクティブサスペンドを OFF（USB 機器を使う場合）
• ハードディスクの停止時間を「なし」

運用設計

• 定期再起動の方針（必要 / 不要を決める。週次など）
• 監視ログを別 PC でモニタリング（ヘルスチェック）
• 障害時の復旧手順を文書化（誰が、どう対応するか）
• 連絡先（担当者の電話、メールアドレス）が UI に表示されている
• バックアップとリストアの手順
• 本番前に 1 週間以上の連続稼働テスト

12. おわりに

本記事では、tkinter で作る現場監視アプリを 24 時間連続稼働させるための設計パターンを、アーキテクチャ・実装・運用の 3 層に分けて解説しました。

「動くデモ」と「現場で動くアプリ」の差は、最初の 1 時間では見えません。この差を埋めるのは、メモリ管理・スレッド設計・自動復旧・ロギング・UI 配慮——いずれも個別には地味な要素ですが、積み重ねが「壊れないアプリ」を作ります。

本記事のサンプルコードを足がかりに、実際の現場で使える監視アプリへ拡張していってください。