psutil × lm-sensors × NVMe SMARTで完全把握！NUCBOX(ミニPC)のシステム情報をInfluxDBに丸ごと記録する

pip install psutil

sudo apt install lm-sensors
sudo sensors-detect   # センサーの自動検出（基本的にEnterを押し続ける）
sensors               # 動作確認

sudo apt install nvme-cli
sudo nvme smart-log /dev/nvme0   # 動作確認

# -*- coding: utf-8 -*-
import os
import socket
import time
import subprocess
import psutil
from influx.v2 import writer as influxwriter
import smarthomelog

def get_cpu_temp():
    """sensorsコマンドからCPUパッケージ温度を取得"""
    try:
        output = subprocess.check_output(["sensors"]).decode()
        for line in output.splitlines():
            if "Package id 0" in line:
                return float(line.split("+")[1].split("°")[0])
    except:
        return -1
    return -1

def get_nvme_temps():
    """sensorsコマンドからNVMe温度3種（Composite/Sensor1/Sensor2）を取得"""
    temps = {
        "nvme_composite": -1,
        "nvme_sensor1": -1,
        "nvme_sensor2": -1
    }
    try:
        output = subprocess.check_output(["sensors"]).decode()
        in_nvme = False
        for line in output.splitlines():
            if "nvme-pci" in line:
                in_nvme = True
            elif in_nvme:
                if "Composite" in line:
                    temps["nvme_composite"] = float(line.split("+")[1].split("°")[0])
                elif "Sensor 1" in line:
                    temps["nvme_sensor1"] = float(line.split("+")[1].split("°")[0])
                elif "Sensor 2" in line:
                    temps["nvme_sensor2"] = float(line.split("+")[1].split("°")[0])
                elif line.strip() == "":
                    break
    except:
        pass
    return temps

def get_nvme_smart():
    """nvme smart-logコマンドからSMART情報を取得（sudo必要）"""
    stats = {
        "data_read_tb": -1, "data_written_tb": -1,
        "power_cycles": -1, "power_on_hours": -1, "unsafe_shutdowns": -1,
        "percentage_used": -1, "media_errors": -1, "error_log_entries": -1,
        "warning_temp_time": -1, "critical_temp_time": -1,
        "thermal_t1_count": -1, "thermal_t1_time": -1,
        "controller_busy_time": -1
    }
    try:
        output = subprocess.check_output(
            ["sudo", "nvme", "smart-log", "/dev/nvme0"]
        ).decode()
        for line in output.splitlines():
            if "Data Units Read" in line:
                val = int(line.split(":")[1].split("(")[0].strip())
                stats["data_read_tb"] = val * 512000 / (1024**4)
            elif "Data Units Written" in line:
                val = int(line.split(":")[1].split("(")[0].strip())
                stats["data_written_tb"] = val * 512000 / (1024**4)
            elif "power_cycles" in line:
                stats["power_cycles"] = int(line.split(":")[1].strip())
            elif "power_on_hours" in line:
                stats["power_on_hours"] = int(line.split(":")[1].strip())
            elif "unsafe_shutdowns" in line:
                stats["unsafe_shutdowns"] = int(line.split(":")[1].strip())
            elif "percentage_used" in line:
                stats["percentage_used"] = int(line.split(":")[1].strip().replace("%",""))
            elif "media_errors" in line:
                stats["media_errors"] = int(line.split(":")[1].strip())
            elif "num_err_log_entries" in line:
                stats["error_log_entries"] = int(line.split(":")[1].strip())
            elif "Warning Temperature Time" in line:
                stats["warning_temp_time"] = int(line.split(":")[1].strip())
            elif "Critical Composite Temperature Time" in line:
                stats["critical_temp_time"] = int(line.split(":")[1].strip())
            elif "Thermal Management T1 Trans Count" in line:
                stats["thermal_t1_count"] = int(line.split(":")[1].strip())
            elif "Thermal Management T1 Total Time" in line:
                stats["thermal_t1_time"] = int(line.split(":")[1].strip())
            elif "controller_busy_time" in line:
                stats["controller_busy_time"] = int(line.split(":")[1].strip())
    except:
        pass
    return stats

def read_int_safe(path):
    """sysファイルシステムから整数値を安全に読み取る"""
    try:
        with open(path, "r") as f:
            return int(f.read().strip())
    except:
        return -1

def get_thermal_throttle():
    """CPUサーマルスロットリング発生回数を取得"""
    return {
        "core_throttle_count":
            read_int_safe("/sys/devices/system/cpu/cpu0/thermal_throttle/core_throttle_count"),
        "package_throttle_count":
            read_int_safe("/sys/devices/system/cpu/cpu0/thermal_throttle/package_throttle_count")
    }

def get_turbo_status(cpu_freq):
    """Turbo Boost有効状態と現在のBoost動作状態を取得"""
    turbo_disabled = read_int_safe(
        "/sys/devices/system/cpu/intel_pstate/no_turbo"
    )
    turbo_enabled = 0 if turbo_disabled == 1 else 1
    cpu_freq_obj = psutil.cpu_freq()
    max_freq = cpu_freq_obj.max if cpu_freq_obj else -1
    min_freq = cpu_freq_obj.min if cpu_freq_obj else -1
    is_boosting = 1 if cpu_freq > 1200 else 0   # 1.2GHz超えでBoost中と判定
    return {
        "turbo_enabled": turbo_enabled,
        "cpu_is_boosting": is_boosting,
        "cpu_max_mhz": max_freq,
        "cpu_min_mhz": min_freq
    }

def record_nucbox_status():
    hostname = socket.gethostname()

    # CPU
    cpu_percent = psutil.cpu_percent(interval=1)
    load1, load5, load15 = os.getloadavg()
    cpu_freq = psutil.cpu_freq().current
    cpu_temp = get_cpu_temp()

    # コア別使用率・周波数・温度
    per_cpu_percent = psutil.cpu_percent(percpu=True)
    per_cpu_freq = psutil.cpu_freq(percpu=True)
    core_temps = {}
    try:
        temps = psutil.sensors_temperatures()
        if "coretemp" in temps:
            for entry in temps["coretemp"]:
                if entry.label.startswith("Core"):
                    core_name = entry.label.lower().replace(" ", "")
                    core_temps[f"{core_name}_temp"] = entry.current
    except:
        pass
    core_data = {}
    for i, usage in enumerate(per_cpu_percent):
        core_data[f"cpu_core{i}_percent"] = usage
        if per_cpu_freq and i < len(per_cpu_freq):
            core_data[f"cpu_core{i}_mhz"] = per_cpu_freq[i].current

    # メモリ
    mem = psutil.virtual_memory()
    swap = psutil.swap_memory()

    # ディスク
    disk = psutil.disk_usage("/")
    io = psutil.disk_io_counters()

    # ネットワーク
    net = psutil.net_io_counters()

    # NVMe
    nvme_temps = get_nvme_temps()
    nvme_smart = get_nvme_smart()

    # サーマルスロットリング・Turbo Boost
    throttle = get_thermal_throttle()
    turbo = get_turbo_status(cpu_freq)

    data = {
        # CPU
        "cpu_percent": cpu_percent,
        "cpu_temp": cpu_temp,
        "cpu_freq_mhz": cpu_freq,
        "cpu_load1": load1,
        "cpu_load5": load5,
        "cpu_load15": load15,
        # メモリ
        "mem_percent": mem.percent,
        "mem_used_mb": mem.used / 1024 / 1024,
        "mem_available_mb": mem.available / 1024 / 1024,
        "mem_swap_used_mb": swap.used / 1024 / 1024,
        # ディスク
        "disk_percent": disk.percent,
        "disk_read_mb": io.read_bytes / 1024 / 1024,
        "disk_write_mb": io.write_bytes / 1024 / 1024,
        "disk_read_count": io.read_count,
        "disk_write_count": io.write_count,
        # ネットワーク
        "net_bytes_sent_mb": net.bytes_sent / 1024 / 1024,
        "net_bytes_recv_mb": net.bytes_recv / 1024 / 1024,
        "net_packets_sent": net.packets_sent,
        "net_packets_recv": net.packets_recv,
        # NVMe温度・SMART・サーマル・Turbo・コア別
        **nvme_temps,
        **nvme_smart,
        **throttle,
        **turbo,
        **core_data,
        **core_temps,
    }

    influx.insert_nucbox_system_log(hostname, data)

# ===== メインループ =====
influx = influxwriter.Writer()
success_log = smarthomelog.Log(__file__, smarthomelog.LogType.SUCCESS)

while True:
    record_nucbox_status()
    success_log.write('NUCBOX処理正常終了', __file__)
    time.sleep(5)

def insert_nucbox_system_log(self, hostname: str, fields):
    fields['timestamp'] = datetime.datetime.now().strftime('%Y/%m/%d %H:%M:%S')
    measurement = 'nucbox_system_log'
    tags = {
        'hostname': hostname
    }
    try:
        self.conn.write_dict(measurement, tags, fields)
    except Exception as e:
        print(e)

output = subprocess.check_output(["sensors"]).decode()
for line in output.splitlines():
    if "Package id 0" in line:
        return float(line.split("+")[1].split("°")[0])

if "Data Units Read" in line:
    val = int(line.split(":")[1].split("(")[0].strip())
    stats["data_read_tb"] = val * 512000 / (1024**4)

for i, usage in enumerate(per_cpu_percent):
    core_data[f"cpu_core{i}_percent"] = usage
    if per_cpu_freq and i < len(per_cpu_freq):
        core_data[f"cpu_core{i}_mhz"] = per_cpu_freq[i].current

[Unit]
Description=NUCBOX System Record to InfluxDB
After=network.target influxd.service

[Service]
ExecStart=/usr/bin/python3 /home/<username>/projects/smarthome/core/record/raspi/nucbox_system_record.py
Restart=always
RestartSec=10
User=root

[Install]
WantedBy=multi-user.target

sudo cp nucbox_system_record.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable nucbox_system_record
sudo systemctl start nucbox_system_record

sudo python3 nucbox_system_record.py

from(bucket: "your-bucket")
  |> range(start: -1m)
  |> filter(fn: (r) => r._measurement == "nucbox_system_log")
  |> filter(fn: (r) => r._field == "cpu_percent" or r._field == "cpu_temp")

from(bucket: "your-bucket")
  |> range(start: -1m)
  |> filter(fn: (r) => r._measurement == "nucbox_system_log")
  |> filter(fn: (r) => r._field == "percentage_used" or r._field == "power_on_hours")