測定したcvsデータを10分毎に読み込みプロットするapp.pyを作成
# -*- coding: utf-8 -*-
from datetime import datetime
import glob
import csv
import subprocessimport os
import dash
import plotly.graph_objects as go
from dash import dcc, html
from dash.dependencies import Input, Output
def get_date():
return datetime.now().strftime("%Y-%m-%d")
def get_path(d):
return f'data/{d}.csv'
def get_csv_dates():
files = glob.glob('data/*.csv')
if not files:
return
files.sort(reverse=True)
dates =
for f in files:
basename = os.path.basename(f)
date_str = os.path.splitext(basename)[0]
dates.append(date_str)
return dates
def read_csv(path):
try:
if not os.path.exists(path):
return
with open(path) as f:
reader = csv.reader(f, delimiter=',')
# ヘッダー（timestamp）を除外して正常な行のみ抽出
rows = [row for row in reader if row and 'timestamp' not in row[0]]
if not rows:
return
# 時間の古い順に並び替え
rows.sort(key=lambda x: x[0])
data_t = [list(x) for x in zip(*rows)]
return data_t
except Exception as e:
print(f"Error reading {path}: {e}")
return []
def sampling():
dates_list = get_csv_dates()
datetime_now = datetime.now().strftime("%Y/%m/%d - %H:%M:%S")
if not dates_list:
return "データがありません", datetime_now
path = get_path(dates_list[0])
try:
with open(path, 'r') as f:
lines = f.readlines()
valid_lines = [l.strip() for l in lines if l.strip() and 'timestamp' not in l]
if not valid_lines:
return "データがありません", datetime_now
last_line = valid_lines[-1].split(',')
count = last_line[1]
b_co2 = last_line[4]
b_sol = last_line[5]
c_co2 = last_line[6]
c_sol = last_line[7]
main_display = f"添加: {count}回 | B機: {b_co2}ppm/{b_sol}W | C機: {c_co2}ppm/{c_sol}W"
return main_display, f"最終更新: {datetime_now}"
except Exception:
return "データ読み取りエラー", datetime_now
def shutdown():
subprocess.Popen(['bash', 'shutdown.sh'])
def safe_float(val):
try:
return float(val)
except (ValueError, TypeError):
return None
def get_combined_fig(path, target_date):
data_t = read_csv(path)
layout = go.Layout(plot_bgcolor='WhiteSmoke', paper_bgcolor='WhiteSmoke')
fig = go.Figure(layout=layout)
# 💡 解決策: グラフのデータ（X軸）には元の「長い日時データ」を100%そのまま渡す
if data_t and len(data_t) >= 8:
timestamps = data_t[0]
co2_b = [safe_float(i) for i in data_t[4]]
sol_b = [safe_float(i) for i in data_t[5]]
co2_c = [safe_float(i) for i in data_t[6]]
sol_c = [safe_float(i) for i in data_t[7]]
# --- B機系統 (赤系) ---
fig.add_trace(go.Scatter(x=timestamps, y=co2_b, name='B機 CO2', yaxis='y1', line=dict(width=3, color='Crimson')))
fig.add_trace(go.Scatter(x=timestamps, y=sol_b, name='B機 日射', yaxis='y2', line=dict(width=1.5, color='LightPink')))
# --- C機系統 (青系) ---
fig.add_trace(go.Scatter(x=timestamps, y=co2_c, name='C機 CO2', yaxis='y1', line=dict(width=3, color='RoyalBlue')))
fig.add_trace(go.Scatter(x=timestamps, y=sol_c, name='C機 日射', yaxis='y2', line=dict(width=1.5, color='LightSkyBlue')))
# 24時間の固定表示枠を「日時型」として裏側で計算する
start_time = f"{target_date} 00:00:00"
end_time = f"{target_date} 23:59:59"
fig.update_layout(
xaxis=dict(
type='date', # X軸の性質を「日付・時間型」に指定
title='時間',
range=[start_time, end_time], # 24時間の幅に固定
tickformat='%H:%M', # 見た目（説明）だけを「時:分」の表示にする
dtick=1800000, # 目盛り間隔は「30分刻み」にする（1800000ミリ秒）
tickangle=45
),
yaxis=dict(title='CO2 (ppm)', range=[400, 1200]),
yaxis2=dict(title='日射 (W/m²)', range=[0, 600], overlaying='y', side='right'),
margin=dict(l=60, r=60, t=40, b=60),
showlegend=True,
legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1),
height=450
)
return fig
# --- Dash App Initial Settings ---
app = dash.Dash(__name__, meta_tags=[{"name": "viewport", "content": "width=device-width, initial-scale=1"}])
app.title = 'Sensor Monitor 2026'
csv_dates = get_csv_dates()
initial_date = csv_dates[0] if csv_dates else get_date()
app.layout = html.Div(children=[
html.Br(),
html.H3('Raspberry Pi Sensor Monitor', style={'fontFamily': 'Arial Black'}),
html.H3(id='container-sample-main', style={'fontFamily': 'Arial Black'}),
html.H6(id='container-sample-sub'),
html.Hr(),
html.Div([
html.Div([
html.Label('Date '),
dcc.Dropdown(csv_dates, value=initial_date, id='dropdown_date')
], className="custom-dropdown"),
], style={'display': 'flex', 'flexWrap': 'wrap', 'justifyContent': 'center'}),
html.Div([
dcc.Graph(id='combined-graph', config={'displayModeBar': False, 'responsive': True}),
], className="custom-figure"),
html.Hr(),
html.Div([
html.Button('Shutdown', id='shutdown', n_clicks=0),
html.H6(id='shutdown_message', style={'fontSize': 16}),
]),
dcc.Interval(id='interval', interval=600000, n_intervals=0)
], style={'textAlign': 'center', 'backgroundColor': 'WhiteSmoke', 'color': '#2F3F5C'})
@app.callback(
[Output('container-sample-main', 'children'),
Output('container-sample-sub', 'children'),
Output('combined-graph', 'figure'),
Output('dropdown_date', 'options'),
Output('dropdown_date', 'value')],
[Input('interval', 'n_intervals'),
Input('dropdown_date', 'value')]
)
def update_dashboard(n, selected_date):
dates_list = get_csv_dates()
co2_val, time_str = sampling()
target_date = selected_date if selected_date in dates_list else (dates_list[0] if dates_list else get_date())
fig = get_combined_fig(get_path(target_date), target_date)
return co2_val, time_str, fig, dates_list, target_date
@app.callback(Output('shutdown_message', 'children'), Input('shutdown', 'n_clicks'))
def handle_shutdown(n_clicks):
if n_clicks > 0:
shutdown()
return 'Shutdown signal sent...'
return ''
if __name__ == '__main__':
app.run(debug=False, host="***.***.***.***")
PCとスマートホンで見られるように表示を変える
/assets/style.css
/* 全体共通 */
body {
margin: 0;
padding: 0;
font-family: Arial, sans-serif;
}
/* PC向けスタイル */
@media (min-width: 1001px) {
#container-sample-main {
font-size: 32px; /* 少し小さくして一行に収まりやすく */
color: #2F3F5C;
margin: 20px 0;
}
}
/* モバイル向けスタイル */
@media (max-width: 1000px) {
#container-sample-main {
font-size: 18px; /* スマホではさらに小さく */
font-weight: bold;
}
}
source myenv/bin/acivate
cd pi/sensor
python app.py
ラズパイ起動時に自動的に起動する
sudo nano /etc/systemd/system/sensormonitor.service
Description=Run sensor monitor
After=network.target
[Service]
ExecStart=/home/akihiro/myenv/bin/python3 /home/akihiro/pi/sensor/app.py
WorkingDirectory=/home/akihiro/pi/sensor
Restart=always
[Install]
WantedBy=multi-user.target
sudo systemctl daemon-reload
sudo systemctl enable sensormonitor.service
sudo systemctl start sensormonitor.service
sudo systemctl status sensormonitor.service
