Coverage for flask/src/rasp_shutter/metrics/collector.py: 74%
88 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-08-23 19:38 +0900
1"""
2シャッター操作メトリクス収集モジュール
4このモジュールは以下のメトリクスを収集し、SQLiteデータベースに保存します：
5- その日に最後に開けた時刻と最後に閉じた時刻
6- 前項の際の照度、日射、太陽高度
7- その日に手動で開けた回数、手動で閉じた回数
8- シャッター制御に失敗した回数
9"""
11from __future__ import annotations
13import datetime
14import logging
15import sqlite3
16import threading
17from pathlib import Path
19import my_lib.sqlite_util
20import my_lib.time
23class MetricsCollector:
24 """シャッターメトリクス収集クラス"""
26 def __init__(self, db_path: Path):
27 """
28 コンストラクタ
30 Args:
31 ----
32 db_path: SQLiteデータベースファイルパス
34 """
35 self.db_path = db_path
36 self.lock = threading.Lock()
37 self._init_database()
39 def _init_database(self):
40 """データベース初期化"""
41 with my_lib.sqlite_util.connect(self.db_path) as conn:
42 conn.execute("""
43 CREATE TABLE IF NOT EXISTS operation_metrics (
44 id INTEGER PRIMARY KEY AUTOINCREMENT,
45 timestamp TIMESTAMP NOT NULL,
46 date TEXT NOT NULL,
47 action TEXT NOT NULL CHECK (action IN ('open', 'close')),
48 operation_type TEXT NOT NULL CHECK (operation_type IN ('manual', 'schedule', 'auto')),
49 lux REAL,
50 solar_rad REAL,
51 altitude REAL,
52 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
53 )
54 """)
56 conn.execute("""
57 CREATE TABLE IF NOT EXISTS daily_failures (
58 id INTEGER PRIMARY KEY AUTOINCREMENT,
59 date TEXT NOT NULL,
60 failure_count INTEGER DEFAULT 1,
61 timestamp TIMESTAMP NOT NULL,
62 created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
63 )
64 """)
66 conn.execute("""
67 CREATE INDEX IF NOT EXISTS idx_operation_metrics_date
68 ON operation_metrics(date)
69 """)
71 conn.execute("""
72 CREATE INDEX IF NOT EXISTS idx_operation_metrics_type
73 ON operation_metrics(operation_type, action)
74 """)
76 conn.execute("""
77 CREATE INDEX IF NOT EXISTS idx_daily_failures_date
78 ON daily_failures(date)
79 """)
81 conn.commit()
83 def _get_today_date(self) -> str:
84 """今日の日付を文字列で取得"""
85 return my_lib.time.now().date().isoformat()
87 def record_shutter_operation(
88 self,
89 action: str,
90 mode: str,
91 sensor_data: dict | None = None,
92 timestamp: datetime.datetime | None = None,
93 ):
94 """
95 シャッター操作を記録
97 Args:
98 ----
99 action: "open" または "close"
100 mode: "manual", "schedule", "auto"
101 sensor_data: センサーデータ（照度、日射、太陽高度など）
102 timestamp: 操作時刻（指定しない場合は現在時刻）
104 """
105 if timestamp is None: 105 ↛ 108line 105 didn't jump to line 108 because the condition on line 105 was always true
106 timestamp = my_lib.time.now()
108 date = timestamp.date().isoformat()
110 # センサーデータを準備
111 lux = None
112 solar_rad = None
113 altitude = None
115 if sensor_data: 115 ↛ 123line 115 didn't jump to line 123 because the condition on line 115 was always true
116 if sensor_data.get("lux", {}).get("valid"):
117 lux = sensor_data["lux"]["value"]
118 if sensor_data.get("solar_rad", {}).get("valid"):
119 solar_rad = sensor_data["solar_rad"]["value"]
120 if sensor_data.get("altitude", {}).get("valid"): 120 ↛ 123line 120 didn't jump to line 123 because the condition on line 120 was always true
121 altitude = sensor_data["altitude"]["value"]
123 with self.lock, sqlite3.connect(self.db_path) as conn:
124 # Python 3.12+: datetimeアダプターを明示的に設定
125 conn.execute("BEGIN")
126 # 個別操作として記録
127 conn.execute(
128 """
129 INSERT INTO operation_metrics
130 (timestamp, date, action, operation_type, lux, solar_rad, altitude)
131 VALUES (?, ?, ?, ?, ?, ?, ?)
132 """,
133 (timestamp.isoformat(), date, action, mode, lux, solar_rad, altitude),
134 )
135 conn.execute("COMMIT")
137 def record_failure(self, timestamp: datetime.datetime | None = None):
138 """
139 シャッター制御失敗を記録
141 Args:
142 ----
143 timestamp: 失敗時刻（指定しない場合は現在時刻）
145 """
146 if timestamp is None: 146 ↛ 149line 146 didn't jump to line 149 because the condition on line 146 was always true
147 timestamp = my_lib.time.now()
149 date = timestamp.date().isoformat()
151 with self.lock, sqlite3.connect(self.db_path) as conn:
152 conn.execute(
153 """
154 INSERT INTO daily_failures (date, timestamp)
155 VALUES (?, ?)
156 """,
157 (date, timestamp.isoformat()),
158 )
160 def get_operation_metrics(self, start_date: str, end_date: str) -> list:
161 """
162 指定期間の操作メトリクスを取得
164 Args:
165 ----
166 start_date: 開始日（YYYY-MM-DD形式）
167 end_date: 終了日（YYYY-MM-DD形式）
169 Returns:
170 -------
171 操作メトリクスデータのリスト
173 """
174 with sqlite3.connect(self.db_path) as conn:
175 conn.row_factory = sqlite3.Row
176 cursor = conn.execute(
177 """
178 SELECT * FROM operation_metrics
179 WHERE date BETWEEN ? AND ?
180 ORDER BY timestamp
181 """,
182 (start_date, end_date),
183 )
184 return [dict(row) for row in cursor.fetchall()]
186 def get_failure_metrics(self, start_date: str, end_date: str) -> list:
187 """
188 指定期間の失敗メトリクスを取得
190 Args:
191 ----
192 start_date: 開始日（YYYY-MM-DD形式）
193 end_date: 終了日（YYYY-MM-DD形式）
195 Returns:
196 -------
197 失敗メトリクスデータのリスト
199 """
200 with sqlite3.connect(self.db_path) as conn:
201 conn.row_factory = sqlite3.Row
202 cursor = conn.execute(
203 """
204 SELECT * FROM daily_failures
205 WHERE date BETWEEN ? AND ?
206 ORDER BY timestamp
207 """,
208 (start_date, end_date),
209 )
210 return [dict(row) for row in cursor.fetchall()]
212 def get_all_operation_metrics(self) -> list:
213 """
214 全期間の操作メトリクスを取得
216 Returns
217 -------
218 操作メトリクスデータのリスト
220 """
221 with sqlite3.connect(self.db_path) as conn:
222 conn.row_factory = sqlite3.Row
223 cursor = conn.execute(
224 """
225 SELECT * FROM operation_metrics
226 ORDER BY timestamp
227 """
228 )
229 return [dict(row) for row in cursor.fetchall()]
231 def get_all_failure_metrics(self) -> list:
232 """
233 全期間の失敗メトリクスを取得
235 Returns
236 -------
237 失敗メトリクスデータのリスト
239 """
240 with sqlite3.connect(self.db_path) as conn:
241 conn.row_factory = sqlite3.Row
242 cursor = conn.execute(
243 """
244 SELECT * FROM daily_failures
245 ORDER BY timestamp
246 """
247 )
248 return [dict(row) for row in cursor.fetchall()]
250 def get_recent_operation_metrics(self, days: int = 30) -> list:
251 """
252 最近N日間の操作メトリクスを取得
254 Args:
255 ----
256 days: 取得する日数
258 Returns:
259 -------
260 操作メトリクスデータのリスト
262 """
263 end_date = my_lib.time.now().date()
264 start_date = end_date - datetime.timedelta(days=days)
266 return self.get_operation_metrics(start_date.isoformat(), end_date.isoformat())
268 def get_recent_failure_metrics(self, days: int = 30) -> list:
269 """
270 最近N日間の失敗メトリクスを取得
272 Args:
273 ----
274 days: 取得する日数
276 Returns:
277 -------
278 失敗メトリクスデータのリスト
280 """
281 end_date = my_lib.time.now().date()
282 start_date = end_date - datetime.timedelta(days=days)
284 return self.get_failure_metrics(start_date.isoformat(), end_date.isoformat())
287# グローバルインスタンス
288_collector_instance: MetricsCollector | None = None
291def get_collector(metrics_data_path) -> MetricsCollector:
292 """メトリクス収集インスタンスを取得"""
293 global _collector_instance # noqa: PLW0603
295 if _collector_instance is None:
296 db_path = Path(metrics_data_path)
297 _collector_instance = MetricsCollector(db_path)
298 logging.info("Metrics collector initialized: %s", db_path)
300 return _collector_instance
303def reset_collector():
304 """グローバルコレクタインスタンスをリセット (テスト用)"""
305 global _collector_instance # noqa: PLW0603
306 _collector_instance = None
309def record_shutter_operation(
310 action: str,
311 mode: str,
312 metrics_data_path,
313 sensor_data: dict | None = None,
314 timestamp: datetime.datetime | None = None,
315):
316 """シャッター操作を記録（便利関数）"""
317 get_collector(metrics_data_path).record_shutter_operation(action, mode, sensor_data, timestamp)
320def record_failure(metrics_data_path, timestamp: datetime.datetime | None = None):
321 """シャッター制御失敗を記録（便利関数）"""
322 get_collector(metrics_data_path).record_failure(timestamp)