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

1""" 

2Query service for health data analysis. 

3 

4Provides methods to query and analyze health data from storage. 

5""" 

6 

7import json 

8from datetime import date, timedelta 

9from pathlib import Path 

10from typing import Dict, List, Optional, Any, Union 

11 

12from health import config 

13from health.db.repository import HealthRepository 

14from health.services.storage import HealthStorage 

15from health.utils.logging_config import setup_logger 

16 

17logger = setup_logger(__name__) 

18 

19 

20class HealthDataQuery: 

21 """Service for querying and analyzing health data.""" 

22 

23 def __init__( 

24 self, storage: Optional[HealthStorage] = None, repo: Optional[HealthRepository] = None 

25 ) -> None: 

26 """Initialize query service. 

27 

28 Args: 

29 storage: Optional storage service instance 

30 repo: Optional repository instance 

31 """ 

32 self.storage = storage or HealthStorage() 

33 self.repo = repo or HealthRepository() 

34 

35 def get_daily_summary(self, target_date: date) -> Dict[str, Any]: 

36 """Get a summary of all health metrics for a specific day. 

37 

38 Args: 

39 target_date: Date to query 

40 

41 Returns: 

42 Dictionary containing all available metrics for the day 

43 """ 

44 logger.info(f"Querying health data summary for {target_date}") 

45 

46 summary = { 

47 "date": target_date.isoformat(), 

48 "metrics": {}, 

49 "activities": [], 

50 } 

51 

52 # Query all daily metrics 

53 for metric_type in config.DATA_TYPE_CONFIG.keys(): 

54 # Skip activities (handled separately) and weight (body metric) 

55 if metric_type in ["activities", "weight"]: 

56 continue 

57 

58 data = self.storage.load_daily_metric(metric_type, target_date) 

59 if data: 

60 summary["metrics"][metric_type] = data 

61 

62 # Query activities for the day 

63 activities = self.repo.get_activities_by_date_range(target_date, target_date) 

64 for activity in activities: 

65 activity_data = self.storage.load_activity(activity["activity_id"]) 

66 if activity_data: 

67 summary["activities"].append(activity_data) 

68 

69 # Query weight (if available) 

70 weight_data = self.storage.load_daily_metric("weight", target_date) 

71 if weight_data: 

72 summary["metrics"]["weight"] = weight_data 

73 

74 logger.info( 

75 f"Found {len(summary['metrics'])} metrics and {len(summary['activities'])} activities" 

76 ) 

77 

78 return summary 

79 

80 def get_metric_range( 

81 self, metric_type: str, start_date: date, end_date: date 

82 ) -> List[Dict[str, Any]]: 

83 """Get a specific metric's data over a date range. 

84 

85 Args: 

86 metric_type: Type of metric (e.g., "steps", "sleep", "heart_rate") 

87 start_date: Start date of range 

88 end_date: End date of range 

89 

90 Returns: 

91 List of metric data dictionaries, ordered by date 

92 

93 Raises: 

94 ValueError: If metric_type is invalid 

95 """ 

96 if metric_type not in config.DATA_TYPE_CONFIG: 

97 valid_types = ", ".join(config.DATA_TYPE_CONFIG.keys()) 

98 raise ValueError( 

99 f"Invalid metric type: {metric_type}. Valid types: {valid_types}" 

100 ) 

101 

102 logger.info(f"Querying {metric_type} from {start_date} to {end_date}") 

103 

104 results = [] 

105 current_date = start_date 

106 

107 while current_date <= end_date: 

108 data = self.storage.load_daily_metric(metric_type, current_date) 

109 if data: 

110 results.append(data) 

111 current_date += timedelta(days=1) 

112 

113 logger.info(f"Found {len(results)} records for {metric_type}") 

114 

115 return results 

116 

117 def get_activities_range( 

118 self, start_date: date, end_date: date, activity_type: Optional[str] = None 

119 ) -> List[Dict[str, Any]]: 

120 """Get all activities within a date range. 

121 

122 Args: 

123 start_date: Start date of range 

124 end_date: End date of range 

125 activity_type: Optional filter by activity type 

126 

127 Returns: 

128 List of activity data dictionaries 

129 """ 

130 logger.info(f"Querying activities from {start_date} to {end_date}") 

131 

132 # Get activity index entries 

133 activity_indices = self.repo.get_activities_by_date_range(start_date, end_date) 

134 

135 # Load full activity data 

136 activities = [] 

137 for idx in activity_indices: 

138 # Filter by activity type if specified 

139 if activity_type and idx.get("activity_type") != activity_type: 

140 continue 

141 

142 activity_data = self.storage.load_activity(idx["activity_id"]) 

143 if activity_data: 

144 activities.append(activity_data) 

145 

146 logger.info(f"Found {len(activities)} activities") 

147 

148 return activities 

149 

150 def get_metric_statistics( 

151 self, metric_type: str, start_date: date, end_date: date, fields: List[str] 

152 ) -> Dict[str, Dict[str, Any]]: 

153 """Calculate statistics for specific metric fields over a date range. 

154 

155 Args: 

156 metric_type: Type of metric 

157 start_date: Start date of range 

158 end_date: End date of range 

159 fields: List of field names to calculate statistics for 

160 

161 Returns: 

162 Dictionary mapping field names to their statistics (min, max, avg, count) 

163 """ 

164 logger.info( 

165 f"Calculating statistics for {metric_type} fields: {fields} from {start_date} to {end_date}" 

166 ) 

167 

168 # Get all data for the range 

169 data_list = self.get_metric_range(metric_type, start_date, end_date) 

170 

171 # Calculate statistics for each field 

172 stats = {} 

173 for field in fields: 

174 values = [] 

175 for data in data_list: 

176 if field in data and data[field] is not None: 

177 values.append(data[field]) 

178 

179 if values: 

180 stats[field] = { 

181 "min": min(values), 

182 "max": max(values), 

183 "avg": sum(values) / len(values), 

184 "count": len(values), 

185 "sum": sum(values), 

186 } 

187 else: 

188 stats[field] = { 

189 "min": None, 

190 "max": None, 

191 "avg": None, 

192 "count": 0, 

193 "sum": None, 

194 } 

195 

196 return stats 

197 

198 def get_available_dates(self, metric_type: str) -> List[str]: 

199 """Get all available dates for a metric type. 

200 

201 Args: 

202 metric_type: Type of metric 

203 

204 Returns: 

205 List of date strings in ISO format 

206 """ 

207 # This is a simplified implementation - in a real scenario, 

208 # you might want to query the database index for better performance 

209 storage_path = config.DATA_TYPE_CONFIG[metric_type]["storage_path"] 

210 base_path = config.DATA_DIR / storage_path 

211 

212 dates = [] 

213 if base_path.exists(): 

214 # Walk through year/month directories 

215 for year_dir in sorted(base_path.iterdir()): 

216 if not year_dir.is_dir(): 

217 continue 

218 for month_dir in sorted(year_dir.iterdir()): 

219 if not month_dir.is_dir(): 

220 continue 

221 for file_path in sorted(month_dir.glob("*.json")): 

222 # Extract date from filename (YYYY-MM-DD.json) 

223 date_str = file_path.stem 

224 dates.append(date_str) 

225 

226 return sorted(dates) 

227 

228 def get_latest_data(self, metric_type: str, days: int = 7) -> List[Dict[str, Any]]: 

229 """Get the most recent data for a metric. 

230 

231 Args: 

232 metric_type: Type of metric 

233 days: Number of days to look back (default: 7) 

234 

235 Returns: 

236 List of metric data dictionaries for the last N days 

237 """ 

238 end_date = date.today() 

239 start_date = end_date - timedelta(days=days - 1) 

240 

241 return self.get_metric_range(metric_type, start_date, end_date)