List all cached kernels, optionally filtered by model.
(self, model_name: Optional[str] = None)
| 179 | return None |
| 180 | |
| 181 | def list_cached_kernels(self, model_name: Optional[str] = None) -> List[Dict[str, Any]]: |
| 182 | """List all cached kernels, optionally filtered by model.""" |
| 183 | with sqlite3.connect(self.db_path) as conn: |
| 184 | cursor = conn.cursor() |
| 185 | |
| 186 | if model_name: |
| 187 | query = """ |
| 188 | SELECT model_name, operation, created_at, performance_metrics, file_path |
| 189 | FROM kernel_metadata |
| 190 | WHERE model_name = ? |
| 191 | ORDER BY created_at DESC |
| 192 | """ |
| 193 | cursor.execute(query, (model_name,)) |
| 194 | else: |
| 195 | query = """ |
| 196 | SELECT model_name, operation, created_at, performance_metrics, file_path |
| 197 | FROM kernel_metadata |
| 198 | ORDER BY created_at DESC |
| 199 | """ |
| 200 | cursor.execute(query) |
| 201 | |
| 202 | results = [] |
| 203 | for row in cursor.fetchall(): |
| 204 | model, op, created, perf_json, file_path = row |
| 205 | file_size = os.path.getsize(file_path) / 1024 if os.path.exists(file_path) else 0 |
| 206 | |
| 207 | results.append({ |
| 208 | "model": model, |
| 209 | "operation": op, |
| 210 | "created": created, |
| 211 | "performance": json.loads(perf_json) if perf_json else {}, |
| 212 | "size_kb": file_size |
| 213 | }) |
| 214 | |
| 215 | return results |
| 216 | |
| 217 | def count_cached_kernels(self) -> int: |
| 218 | """Count total number of cached kernels.""" |