ローカルMQTTブローカーの開発

[Unit]
Description=Set CPU Max Frequency to 2.0GHz for Stability
After=multi-user.target

[Service]
Type=oneshot
ExecStart=/bin/bash -c 'for i in /sys/devices/system/cpu/cpu*/cpufreq/scaling_max_freq; do echo 2000000 > $i; done'
ExecStart=/bin/bash -c 'for i in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do echo performance > $i; done'
RemainAfterExit=yes

[Install]
WantedBy=multi-user.target

#!/bin/bash

while true; do 
  echo "$(date +%s),$(sensors | grep 'Tctl' | awk '{print $2}' | tr -d '+°C'),$(cat /sys/class/powercap/intel-rapl:0/energy_uj),$(cat /sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq)" >> cpu_data.csv
  sleep 600
done

HandleLidSwitch=ignore
HandleLidSwitchExternalPower=ignore
IdleAction=ignore
IdleActionSec=0

HandlePowerKey=poweroff

import time
import csv
import os
import psutil
import subprocess

# 設定値
SLEEP_INTERVAL = 60  # データ取得間隔（秒単位、10分）
CSV_FILE = "cpu_data.csv"  # 出力CSVファイル名
ENERGY_UJ_PATH = "/sys/class/powercap/intel-rapl:0/energy_uj"  # 消費エネルギーファイル
CPU_FREQ_PATH = "/sys/devices/system/cpu/cpu0/cpufreq/scaling_cur_freq"  # CPUクロック周波数ファイル
SENSORS_KEYWORD = "Package id 0"  # `sensors` コマンドで取得する温度のキー
STRESS_DURATION = 600  # 負荷テストの実行時間（秒）
STRESS_CORES = 4  # 使用するCPUコア数

def get_cpu_temp():
    """CPUの温度を取得する"""
    try:
        result = subprocess.run(["sensors"], capture_output=True, text=True, check=True)
        for line in result.stdout.split("\n"):
            if SENSORS_KEYWORD in line:
                parts = line.split()
                if len(parts) >= 4:  # インデックスエラーを防ぐ
                    try:
                        return float(parts[3].replace("+", "").replace("°C", ""))
                    except ValueError:
                        return -1.0  # 異常値として -1.0 を返す
    except subprocess.CalledProcessError:
        return -1.0  # `sensors` コマンドの実行失敗時

    return -1.0  # `SENSORS_KEYWORD` が見つからなかった場合



def get_cpu_freq():
    """CPUのクロック周波数を取得する"""
    with open(CPU_FREQ_PATH, "r") as f:
        return int(f.read().strip()) / 1000  # kHz → MHz

def get_cpu_usage():
    """CPU使用率を取得する"""
    return psutil.cpu_percent(interval=1)

def get_energy_uj():
    """消費エネルギー（microjoules）を取得する"""
    try:
        result = subprocess.run(["sudo", "cat", ENERGY_UJ_PATH], capture_output=True, text=True)
        return int(result.stdout.strip())
    except Exception:
        return None

def run_stress():
    """負荷テストを実行する"""
    print(f"Starting stress test: {STRESS_CORES} cores for {STRESS_DURATION} seconds")
    subprocess.Popen(["stress", "-c", str(STRESS_CORES), "-t", str(STRESS_DURATION)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)

def main():
    """メイン処理：データを定期的に記録し、負荷テストを実行"""
    prev_energy = get_energy_uj()
    prev_time = time.time()

    with open(CSV_FILE, "a", newline="") as csvfile:
        writer = csv.writer(csvfile)
        writer.writerow(["timestamp", "cpu_temp", "power", "cpu_freq", "cpu_usage"])

        while True:
            run_stress()  # 負荷テストを実行

            time.sleep(SLEEP_INTERVAL)  # 設定した間隔でデータ取得

            curr_energy = get_energy_uj()
            curr_time = time.time()
            energy_diff = (curr_energy - prev_energy) / 1_000_000  # µJ → J
            time_diff = curr_time - prev_time
            power = round(energy_diff / time_diff, 2)

            cpu_temp = get_cpu_temp()
            cpu_freq = get_cpu_freq()
            cpu_usage = get_cpu_usage()

            writer.writerow([int(curr_time), cpu_temp, power, cpu_freq, cpu_usage])
            csvfile.flush()

            prev_energy = curr_energy
            prev_time = curr_time

if __name__ == "__main__":
    main()