Coverage for app \ storage \ logger.py: 22%

1""" 

2Usage logger for tracking SLM and automata usage 

3 

4Logs all requests to SQLite for pattern analysis and plasticity 

5""" 

6import logging 

7import sqlite3 

8from datetime import datetime 

9from typing import Dict, Any, Optional 

10from pathlib import Path 

11import json 

12 

13from app.config import settings 

14 

15logger = logging.getLogger(__name__) 

16 

17 

18def get_logger(name: str) -> logging.Logger: 

19 """ 

20 Get a logger instance for the given module name. 

21  

22 Args: 

23 name: Module name (usually __name__) 

24  

25 Returns: 

26 Logger instance 

27 """ 

28 return logging.getLogger(name) 

29 

30 

31class UsageLogger: 

32 """Logs all code processing requests for analysis""" 

33 

34 def __init__(self, db_path: Optional[Path] = None): 

35 """ 

36 Initialize usage logger 

37 

38 Args: 

39 db_path: Path to SQLite database 

40 """ 

41 self.db_path = db_path or settings.db_path 

42 self._ensure_db() 

43 

44 def _ensure_db(self): 

45 """Create database and tables if they don't exist""" 

46 try: 

47 self.db_path.parent.mkdir(parents=True, exist_ok=True) 

48 

49 conn = sqlite3.connect(self.db_path) 

50 cursor = conn.cursor() 

51 

52 # Main usage log table 

53 cursor.execute(""" 

54 CREATE TABLE IF NOT EXISTS usage_log ( 

55 id INTEGER PRIMARY KEY AUTOINCREMENT, 

56 timestamp DATETIME DEFAULT CURRENT_TIMESTAMP, 

57 task TEXT NOT NULL, 

58 language TEXT NOT NULL, 

59 code_hash TEXT, 

60 used_automata BOOLEAN, 

61 used_slm BOOLEAN, 

62 automata_components TEXT, 

63 slm_component TEXT, 

64 success BOOLEAN, 

65 duration_ms REAL, 

66 error TEXT 

67 ) 

68 """) 

69 

70 # Pattern frequency table 

71 cursor.execute(""" 

72 CREATE TABLE IF NOT EXISTS pattern_frequency ( 

73 id INTEGER PRIMARY KEY AUTOINCREMENT, 

74 pattern_type TEXT NOT NULL, 

75 pattern_signature TEXT NOT NULL, 

76 frequency INTEGER DEFAULT 1, 

77 last_seen DATETIME DEFAULT CURRENT_TIMESTAMP, 

78 automatable BOOLEAN DEFAULT 0, 

79 UNIQUE(pattern_type, pattern_signature) 

80 ) 

81 """) 

82 

83 # Automata conversion candidates 

84 cursor.execute(""" 

85 CREATE TABLE IF NOT EXISTS automata_candidates ( 

86 id INTEGER PRIMARY KEY AUTOINCREMENT, 

87 pattern_type TEXT NOT NULL, 

88 pattern_signature TEXT NOT NULL, 

89 frequency INTEGER, 

90 estimated_speedup REAL, 

91 priority_score REAL, 

92 created_at DATETIME DEFAULT CURRENT_TIMESTAMP, 

93 converted BOOLEAN DEFAULT 0 

94 ) 

95 """) 

96 

97 conn.commit() 

98 conn.close() 

99 

100 logger.info(f"Database initialized at {self.db_path}") 

101 

102 except Exception as e: 

103 logger.error(f"Failed to initialize database: {e}") 

104 raise 

105 

106 def log_request( 

107 self, 

108 task: str, 

109 language: str, 

110 code_hash: str, 

111 used_automata: bool, 

112 used_slm: bool, 

113 pipeline: list, 

114 success: bool, 

115 duration_ms: float, 

116 error: Optional[str] = None 

117 ): 

118 """Log a code processing request""" 

119 try: 

120 conn = sqlite3.connect(self.db_path) 

121 cursor = conn.cursor() 

122 

123 # Extract components from pipeline 

124 automata_components = [ 

125 step['component'] for step in pipeline 

126 if step['step_type'] == 'automata' 

127 ] 

128 slm_component = next( 

129 (step['component'] for step in pipeline if step['step_type'] == 'slm'), 

130 None 

131 ) 

132 

133 cursor.execute(""" 

134 INSERT INTO usage_log ( 

135 task, language, code_hash, used_automata, used_slm, 

136 automata_components, slm_component, success, duration_ms, error 

137 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 

138 """, ( 

139 task, 

140 language, 

141 code_hash, 

142 used_automata, 

143 used_slm, 

144 json.dumps(automata_components) if automata_components else None, 

145 slm_component, 

146 success, 

147 duration_ms, 

148 error 

149 )) 

150 

151 conn.commit() 

152 conn.close() 

153 

154 logger.debug(f"Logged request: {task} ({language}) - automata:{used_automata} slm:{used_slm}") 

155 

156 except Exception as e: 

157 logger.error(f"Failed to log request: {e}") 

158 

159 def update_pattern_frequency( 

160 self, 

161 pattern_type: str, 

162 pattern_signature: str 

163 ): 

164 """Update frequency counter for a pattern""" 

165 try: 

166 conn = sqlite3.connect(self.db_path) 

167 cursor = conn.cursor() 

168 

169 cursor.execute(""" 

170 INSERT INTO pattern_frequency (pattern_type, pattern_signature, frequency) 

171 VALUES (?, ?, 1) 

172 ON CONFLICT(pattern_type, pattern_signature) 

173 DO UPDATE SET 

174 frequency = frequency + 1, 

175 last_seen = CURRENT_TIMESTAMP 

176 """, (pattern_type, pattern_signature)) 

177 

178 conn.commit() 

179 conn.close() 

180 

181 except Exception as e: 

182 logger.error(f"Failed to update pattern frequency: {e}") 

183 

184 def get_frequent_patterns(self, min_frequency: int = 10) -> list: 

185 """Get patterns that occur frequently""" 

186 try: 

187 conn = sqlite3.connect(self.db_path) 

188 conn.row_factory = sqlite3.Row 

189 cursor = conn.cursor() 

190 

191 cursor.execute(""" 

192 SELECT * 

193 FROM pattern_frequency 

194 WHERE frequency >= ? AND automatable = 0 

195 ORDER BY frequency DESC 

196 """, (min_frequency,)) 

197 

198 patterns = [dict(row) for row in cursor.fetchall()] 

199 

200 conn.close() 

201 return patterns 

202 

203 except Exception as e: 

204 logger.error(f"Failed to get frequent patterns: {e}") 

205 return [] 

206 

207 def get_stats(self) -> Dict[str, Any]: 

208 """Get usage statistics""" 

209 try: 

210 conn = sqlite3.connect(self.db_path) 

211 conn.row_factory = sqlite3.Row 

212 cursor = conn.cursor() 

213 

214 # Total requests 

215 cursor.execute("SELECT COUNT(*) as total FROM usage_log") 

216 total = cursor.fetchone()['total'] 

217 

218 # Automata vs SLM usage 

219 cursor.execute(""" 

220 SELECT 

221 SUM(CASE WHEN used_automata AND NOT used_slm THEN 1 ELSE 0 END) as automata_only, 

222 SUM(CASE WHEN used_slm AND NOT used_automata THEN 1 ELSE 0 END) as slm_only, 

223 SUM(CASE WHEN used_automata AND used_slm THEN 1 ELSE 0 END) as both 

224 FROM usage_log 

225 """) 

226 usage = dict(cursor.fetchone()) 

227 

228 # Average duration 

229 cursor.execute(""" 

230 SELECT 

231 AVG(CASE WHEN used_automata AND NOT used_slm THEN duration_ms END) as avg_automata_ms, 

232 AVG(CASE WHEN used_slm THEN duration_ms END) as avg_slm_ms 

233 FROM usage_log 

234 WHERE success = 1 

235 """) 

236 durations = dict(cursor.fetchone()) 

237 

238 # Automation rate 

239 automation_rate = (usage['automata_only'] / total * 100) if total > 0 else 0 

240 

241 conn.close() 

242 

243 return { 

244 "total_requests": total, 

245 "automation_rate": automation_rate, 

246 "automata_only": usage['automata_only'], 

247 "slm_only": usage['slm_only'], 

248 "both": usage['both'], 

249 "avg_automata_duration_ms": durations['avg_automata_ms'] or 0, 

250 "avg_slm_duration_ms": durations['avg_slm_ms'] or 0 

251 } 

252 

253 except Exception as e: 

254 logger.error(f"Failed to get stats: {e}") 

255 return {}