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