Coverage for health / cli / query.py: 0%

302 statements  

« prev     ^ index     » next       coverage.py v7.13.4, created at 2026-03-02 17:44 +0800

1""" 

2CLI commands for querying health data. 

3 

4Provides command-line interface for viewing and analyzing health metrics. 

5""" 

6 

7import sys 

8from datetime import date, datetime, timedelta 

9from pathlib import Path 

10from typing import Optional, List, Union 

11 

12import click 

13 

14# Add parent directory to path 

15sys.path.insert(0, str(Path(__file__).parent.parent.parent)) 

16 

17from health import config 

18from health.services.query import HealthDataQuery 

19from health.utils.logging_config import setup_logger 

20 

21logger = setup_logger(__name__) 

22 

23 

24def format_date(date_obj: date) -> str: 

25 """Format date for display. 

26 

27 Args: 

28 date_obj: Date to format 

29 

30 Returns: 

31 Formatted date string 

32 """ 

33 return date_obj.strftime("%Y-%m-%d (%A)") 

34 

35 

36def format_seconds_to_hms(seconds: Optional[Union[int, float]]) -> str: 

37 """Convert seconds to HH:MM:SS format. 

38 

39 Args: 

40 seconds: Number of seconds (int or float) 

41 

42 Returns: 

43 Formatted time string 

44 """ 

45 if seconds is None: 

46 return "N/A" 

47 

48 # Convert to int if it's a float 

49 seconds = int(seconds) 

50 

51 hours = seconds // 3600 

52 minutes = (seconds % 3600) // 60 

53 secs = seconds % 60 

54 

55 return f"{hours:02d}:{minutes:02d}:{secs:02d}" 

56 

57 

58def format_number(value: Optional[Union[int, float]], decimals: int = 0) -> str: 

59 """Safely format a number with thousand separators. 

60 

61 Args: 

62 value: Number to format 

63 decimals: Number of decimal places 

64 

65 Returns: 

66 Formatted number string or 'N/A' 

67 """ 

68 if value is None: 

69 return "N/A" 

70 

71 if decimals > 0: 

72 return f"{value:,.{decimals}f}" 

73 else: 

74 return f"{int(value):,}" 

75 

76 

77def print_daily_summary(summary: dict) -> None: 

78 """Print a formatted daily summary. 

79 

80 Args: 

81 summary: Daily summary dictionary 

82 """ 

83 target_date = datetime.fromisoformat(summary["date"]).date() 

84 

85 print(f"\n{'='*80}") 

86 print(f"📊 Health Data Summary for {format_date(target_date)}") 

87 print(f"{'='*80}\n") 

88 

89 metrics = summary.get("metrics", {}) 

90 

91 # Steps 

92 if "steps" in metrics: 

93 steps = metrics["steps"] 

94 print(f"🚶 Steps:") 

95 print(f" Total Steps: {format_number(steps.get('total_steps'))}") 

96 if steps.get('total_distance_meters'): 

97 print(f" Distance: {steps.get('total_distance_meters', 0) / 1000:.2f} km") 

98 if steps.get('calories_burned'): 

99 print(f" Calories: {format_number(steps.get('calories_burned'))} kcal") 

100 print() 

101 

102 # Sleep 

103 if "sleep" in metrics: 

104 sleep = metrics["sleep"] 

105 print(f"😴 Sleep:") 

106 if sleep.get('total_sleep_seconds'): 

107 print(f" Total Sleep: {format_seconds_to_hms(sleep.get('total_sleep_seconds'))}") 

108 if sleep.get('deep_sleep_seconds'): 

109 print(f" Deep Sleep: {format_seconds_to_hms(sleep.get('deep_sleep_seconds'))}") 

110 if sleep.get('light_sleep_seconds'): 

111 print(f" Light Sleep: {format_seconds_to_hms(sleep.get('light_sleep_seconds'))}") 

112 if sleep.get('rem_sleep_seconds'): 

113 print(f" REM Sleep: {format_seconds_to_hms(sleep.get('rem_sleep_seconds'))}") 

114 if sleep.get('sleep_score') is not None: 

115 print(f" Sleep Score: {sleep.get('sleep_score')}/100") 

116 print() 

117 

118 # Heart Rate 

119 if "heart_rate" in metrics: 

120 hr = metrics["heart_rate"] 

121 print(f"💓 Heart Rate:") 

122 if hr.get('resting_heart_rate'): 

123 print(f" Resting HR: {hr.get('resting_heart_rate')} bpm") 

124 if hr.get('min_heart_rate'): 

125 print(f" Min HR: {hr.get('min_heart_rate')} bpm") 

126 if hr.get('max_heart_rate'): 

127 print(f" Max HR: {hr.get('max_heart_rate')} bpm") 

128 if hr.get('average_heart_rate'): 

129 print(f" Average HR: {hr.get('average_heart_rate')} bpm") 

130 print() 

131 

132 # HRV 

133 if "hrv" in metrics: 

134 hrv = metrics["hrv"] 

135 print(f"📈 HRV (Heart Rate Variability):") 

136 if hrv.get('hrv_value'): 

137 print(f" HRV: {hrv.get('hrv_value'):.1f} ms") 

138 if hrv.get('status'): 

139 print(f" Status: {hrv.get('status')}") 

140 print() 

141 

142 # Stress 

143 if "stress" in metrics: 

144 stress = metrics["stress"] 

145 print(f"😰 Stress:") 

146 if stress.get('average_stress_level') is not None: 

147 print(f" Average Stress: {stress.get('average_stress_level')}/100") 

148 if stress.get('max_stress_level') is not None: 

149 print(f" Max Stress: {stress.get('max_stress_level')}/100") 

150 print() 

151 

152 # Body Battery 

153 if "body_battery" in metrics: 

154 bb = metrics["body_battery"] 

155 print(f"🔋 Body Battery:") 

156 if bb.get('charged'): 

157 print(f" Charged: +{bb.get('charged')}") 

158 if bb.get('drained'): 

159 print(f" Drained: -{bb.get('drained')}") 

160 if bb.get('highest_value') is not None: 

161 print(f" Highest: {bb.get('highest_value')}/100") 

162 if bb.get('lowest_value') is not None: 

163 print(f" Lowest: {bb.get('lowest_value')}/100") 

164 print() 

165 

166 # Weight 

167 if "weight" in metrics: 

168 weight = metrics["weight"] 

169 print(f"⚖️ Weight:") 

170 if weight.get('weight_kg'): 

171 print(f" Weight: {weight.get('weight_kg')} kg") 

172 if weight.get('bmi'): 

173 print(f" BMI: {weight.get('bmi'):.1f}") 

174 if weight.get('body_fat_percentage'): 

175 print(f" Body Fat: {weight.get('body_fat_percentage'):.1f}%") 

176 print() 

177 

178 # Respiration 

179 if "respiration" in metrics: 

180 resp = metrics["respiration"] 

181 print(f"🫁 Respiration:") 

182 if resp.get('average_respiration_rate'): 

183 print(f" Average: {resp.get('average_respiration_rate'):.1f} bpm") 

184 if resp.get('min_respiration_rate'): 

185 print(f" Min: {resp.get('min_respiration_rate'):.1f} bpm") 

186 if resp.get('max_respiration_rate'): 

187 print(f" Max: {resp.get('max_respiration_rate'):.1f} bpm") 

188 print() 

189 

190 # SPO2 

191 if "spo2" in metrics: 

192 spo2 = metrics["spo2"] 

193 if spo2.get('average_spo2'): 

194 print(f"🩸 Blood Oxygen (SpO2):") 

195 print(f" Average: {spo2.get('average_spo2'):.1f}%") 

196 if spo2.get('min_spo2'): 

197 print(f" Min: {spo2.get('min_spo2'):.1f}%") 

198 if spo2.get('max_spo2'): 

199 print(f" Max: {spo2.get('max_spo2'):.1f}%") 

200 print() 

201 

202 # Floors 

203 if "floors" in metrics: 

204 floors = metrics["floors"] 

205 if floors.get('floors_climbed'): 

206 print(f"🪜 Floors:") 

207 print(f" Climbed: {floors.get('floors_climbed')}") 

208 if floors.get('floors_descended'): 

209 print(f" Descended: {floors.get('floors_descended')}") 

210 print() 

211 

212 # Intensity Minutes 

213 if "intensity_minutes" in metrics: 

214 im = metrics["intensity_minutes"] 

215 if im.get('moderate_minutes') is not None or im.get('vigorous_minutes') is not None: 

216 print(f"⏱️ Intensity Minutes:") 

217 if im.get('moderate_minutes') is not None: 

218 print(f" Moderate: {im.get('moderate_minutes')} min") 

219 if im.get('vigorous_minutes') is not None: 

220 print(f" Vigorous: {im.get('vigorous_minutes')} min") 

221 if im.get('total_minutes') is not None: 

222 print(f" Total: {im.get('total_minutes')} min") 

223 print() 

224 

225 # Hydration 

226 if "hydration" in metrics: 

227 hydration = metrics["hydration"] 

228 if hydration.get('total_intake_ml'): 

229 print(f"💧 Hydration:") 

230 print(f" Intake: {hydration.get('total_intake_ml')} ml") 

231 if hydration.get('goal_ml'): 

232 print(f" Goal: {hydration.get('goal_ml')} ml") 

233 print() 

234 

235 # RHR (Resting Heart Rate) - if not already shown in heart_rate 

236 if "rhr" in metrics and "heart_rate" not in metrics: 

237 rhr = metrics["rhr"] 

238 if rhr.get('resting_heart_rate'): 

239 print(f"❤️ Resting Heart Rate:") 

240 print(f" RHR: {rhr.get('resting_heart_rate')} bpm") 

241 print() 

242 

243 # Activities 

244 activities = summary.get("activities", []) 

245 if activities: 

246 print(f"🏃 Activities ({len(activities)}):") 

247 for activity in activities: 

248 activity_type = activity.get("activity_type", "Unknown") 

249 duration = format_seconds_to_hms(activity.get("duration_seconds")) 

250 distance = activity.get("distance_meters") 

251 

252 activity_str = f" • {activity_type}: {duration}" 

253 if distance: 

254 activity_str += f", {distance / 1000:.2f} km" 

255 

256 if activity.get("average_heart_rate"): 

257 activity_str += f", Avg HR: {format_number(activity.get('average_heart_rate'))} bpm" 

258 

259 print(activity_str) 

260 print() 

261 

262 print(f"{'='*80}\n") 

263 

264 

265def print_metric_range(metric_type: str, data_list: List[dict]) -> None: 

266 """Print a formatted metric range. 

267 

268 Args: 

269 metric_type: Type of metric 

270 data_list: List of metric data 

271 """ 

272 if not data_list: 

273 print(f"\n❌ No data found for {metric_type}\n") 

274 return 

275 

276 print(f"\n{'='*80}") 

277 print(f"📊 {metric_type.upper().replace('_', ' ')} - {len(data_list)} records") 

278 print(f"{'='*80}\n") 

279 

280 # Define what fields to display for each metric type 

281 display_configs = { 

282 "steps": ["date", "total_steps", "total_distance_meters", "calories_burned"], 

283 "sleep": ["date", "total_sleep_seconds", "deep_sleep_seconds", "sleep_score"], 

284 "heart_rate": ["date", "resting_heart_rate", "min_heart_rate", "max_heart_rate"], 

285 "hrv": ["date", "hrv_value", "status"], 

286 "stress": ["date", "average_stress_level", "max_stress_level"], 

287 "body_battery": ["date", "charged", "drained", "highest_value", "lowest_value"], 

288 "weight": ["date", "weight_kg", "bmi", "body_fat_percentage"], 

289 "rhr": ["date", "resting_heart_rate"], 

290 } 

291 

292 fields = display_configs.get(metric_type, ["date"]) 

293 

294 # Print header 

295 header = " | ".join([f"{field:20}" for field in fields]) 

296 print(header) 

297 print("-" * len(header)) 

298 

299 # Print data rows 

300 for data in data_list: 

301 row = [] 

302 for field in fields: 

303 value = data.get(field, None) 

304 

305 # Format specific fields 

306 if field == "date": 

307 value = str(value) if value else "N/A" 

308 elif field.endswith("_seconds"): 

309 value = format_seconds_to_hms(value) 

310 elif isinstance(value, float): 

311 value = format_number(value, decimals=2) 

312 elif isinstance(value, int): 

313 value = format_number(value) 

314 else: 

315 value = str(value) if value is not None else "N/A" 

316 

317 row.append(f"{value:20}") 

318 

319 print(" | ".join(row)) 

320 

321 print(f"\n{'='*80}\n") 

322 

323 

324@click.group() 

325def cli() -> None: 

326 """Health data query CLI.""" 

327 pass 

328 

329 

330@cli.command() 

331@click.argument("date_str", required=False) 

332def daily(date_str: Optional[str] = None) -> None: 

333 """Show health data summary for a specific day. 

334 

335 DATE_STR: Date in YYYY-MM-DD format (default: today) 

336 

337 Examples: 

338 python -m health.cli.query daily 

339 python -m health.cli.query daily 2024-01-15 

340 """ 

341 try: 

342 # Parse date 

343 if date_str: 

344 target_date = datetime.strptime(date_str, "%Y-%m-%d").date() 

345 else: 

346 target_date = date.today() 

347 

348 # Query data 

349 query_service = HealthDataQuery() 

350 summary = query_service.get_daily_summary(target_date) 

351 

352 # Display 

353 print_daily_summary(summary) 

354 

355 except ValueError as e: 

356 click.echo(f"❌ Error: Invalid date format. Use YYYY-MM-DD. {e}", err=True) 

357 sys.exit(1) 

358 except Exception as e: 

359 click.echo(f"❌ Error: {e}", err=True) 

360 logger.exception("Failed to get daily summary") 

361 sys.exit(1) 

362 

363 

364@cli.command() 

365@click.argument("metric_type") 

366@click.argument("start_date") 

367@click.argument("end_date") 

368@click.option("--stats", is_flag=True, help="Show statistics summary") 

369def range(metric_type: str, start_date: str, end_date: str, stats: bool) -> None: 

370 """Show metric data over a date range. 

371 

372 METRIC_TYPE: Type of metric (e.g., steps, sleep, heart_rate) 

373 START_DATE: Start date in YYYY-MM-DD format 

374 END_DATE: End date in YYYY-MM-DD format 

375 

376 Examples: 

377 python -m health.cli.query range steps 2024-01-01 2024-01-07 

378 python -m health.cli.query range sleep 2024-01-01 2024-01-07 --stats 

379 """ 

380 try: 

381 # Parse dates 

382 start = datetime.strptime(start_date, "%Y-%m-%d").date() 

383 end = datetime.strptime(end_date, "%Y-%m-%d").date() 

384 

385 if start > end: 

386 raise ValueError("Start date must be before or equal to end date") 

387 

388 # Query data 

389 query_service = HealthDataQuery() 

390 data_list = query_service.get_metric_range(metric_type, start, end) 

391 

392 # Display 

393 print_metric_range(metric_type, data_list) 

394 

395 # Show statistics if requested 

396 if stats and data_list: 

397 # Determine which fields to analyze 

398 field_configs = { 

399 "steps": ["total_steps", "total_distance_meters", "calories_burned"], 

400 "sleep": ["total_sleep_seconds", "deep_sleep_seconds", "sleep_score"], 

401 "heart_rate": ["resting_heart_rate", "min_heart_rate", "max_heart_rate"], 

402 "hrv": ["hrv_value"], 

403 "stress": ["average_stress_level", "max_stress_level"], 

404 "body_battery": ["charged", "drained", "highest_value"], 

405 "weight": ["weight", "bmi", "body_fat_percentage"], 

406 } 

407 

408 fields = field_configs.get(metric_type, []) 

409 if fields: 

410 statistics = query_service.get_metric_statistics(metric_type, start, end, fields) 

411 

412 print(f"📈 Statistics Summary:\n") 

413 for field, stat in statistics.items(): 

414 if stat["count"] > 0: 

415 print(f"{field}:") 

416 print(f" Min: {stat['min']}") 

417 print(f" Max: {stat['max']}") 

418 print(f" Avg: {stat['avg']:.2f}") 

419 print(f" Count: {stat['count']}") 

420 print() 

421 

422 except ValueError as e: 

423 click.echo(f"❌ Error: {e}", err=True) 

424 sys.exit(1) 

425 except Exception as e: 

426 click.echo(f"❌ Error: {e}", err=True) 

427 logger.exception("Failed to get metric range") 

428 sys.exit(1) 

429 

430 

431@cli.command() 

432@click.argument("start_date") 

433@click.argument("end_date") 

434@click.option("--type", "activity_type", help="Filter by activity type") 

435def activities(start_date: str, end_date: str, activity_type: Optional[str] = None) -> None: 

436 """Show activities within a date range. 

437 

438 START_DATE: Start date in YYYY-MM-DD format 

439 END_DATE: End date in YYYY-MM-DD format 

440 

441 Examples: 

442 python -m health.cli.query activities 2024-01-01 2024-01-07 

443 python -m health.cli.query activities 2024-01-01 2024-01-07 --type running 

444 """ 

445 try: 

446 # Parse dates 

447 start = datetime.strptime(start_date, "%Y-%m-%d").date() 

448 end = datetime.strptime(end_date, "%Y-%m-%d").date() 

449 

450 if start > end: 

451 raise ValueError("Start date must be before or equal to end date") 

452 

453 # Query data 

454 query_service = HealthDataQuery() 

455 activities_list = query_service.get_activities_range(start, end, activity_type) 

456 

457 if not activities_list: 

458 print(f"\n❌ No activities found\n") 

459 return 

460 

461 print(f"\n{'='*80}") 

462 print(f"🏃 Activities ({len(activities_list)} found)") 

463 print(f"{'='*80}\n") 

464 

465 for activity in activities_list: 

466 print(f"📅 {activity.get('date')} - {activity.get('activity_type', 'Unknown')}") 

467 print(f" Duration: {format_seconds_to_hms(activity.get('duration_seconds'))}") 

468 

469 if activity.get('distance_meters'): 

470 print(f" Distance: {activity.get('distance_meters') / 1000:.2f} km") 

471 

472 if activity.get('average_heart_rate'): 

473 print(f" Avg HR: {format_number(activity.get('average_heart_rate'))} bpm") 

474 

475 if activity.get('max_heart_rate'): 

476 print(f" Max HR: {format_number(activity.get('max_heart_rate'))} bpm") 

477 

478 if activity.get('calories'): 

479 print(f" Calories: {format_number(activity.get('calories'))} kcal") 

480 

481 print() 

482 

483 print(f"{'='*80}\n") 

484 

485 except ValueError as e: 

486 click.echo(f"❌ Error: {e}", err=True) 

487 sys.exit(1) 

488 except Exception as e: 

489 click.echo(f"❌ Error: {e}", err=True) 

490 logger.exception("Failed to get activities") 

491 sys.exit(1) 

492 

493 

494@cli.command() 

495def types() -> None: 

496 """List all available metric types.""" 

497 print("\n📋 Available Metric Types:\n") 

498 

499 for metric_type, type_config in config.DATA_TYPE_CONFIG.items(): 

500 print(f" • {metric_type:20} - {type_config['description']}") 

501 

502 print() 

503 

504 

505if __name__ == "__main__": 

506 cli()