Cache a kernel and its metadata.
(self, model_name: str, operation: str, kernel_data: Dict[str, Any])
| 119 | return hashlib.sha256(content.encode()).hexdigest()[:16] |
| 120 | |
| 121 | def cache_kernel(self, model_name: str, operation: str, kernel_data: Dict[str, Any]): |
| 122 | """Cache a kernel and its metadata.""" |
| 123 | kernel_code = kernel_data.get("code", "") |
| 124 | kernel_id = self._generate_kernel_id(model_name, operation, kernel_code) |
| 125 | |
| 126 | kernel_file = self.kernels_dir / f"{kernel_id}.pkl" |
| 127 | with open(kernel_file, 'wb') as f: |
| 128 | pickle.dump(kernel_data, f) |
| 129 | |
| 130 | with sqlite3.connect(self.db_path) as conn: |
| 131 | cursor = conn.cursor() |
| 132 | cursor.execute(""" |
| 133 | INSERT OR REPLACE INTO kernel_metadata |
| 134 | (id, model_name, operation, created_at, last_used, |
| 135 | performance_metrics, constraints, file_path, checksum) |
| 136 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 137 | """, ( |
| 138 | kernel_id, |
| 139 | model_name, |
| 140 | operation, |
| 141 | datetime.now().isoformat(), |
| 142 | datetime.now().isoformat(), |
| 143 | json.dumps(kernel_data.get("benchmark", {})), |
| 144 | json.dumps(kernel_data.get("constraints", {})), |
| 145 | str(kernel_file), |
| 146 | hashlib.sha256(kernel_code.encode()).hexdigest() |
| 147 | )) |
| 148 | conn.commit() |
| 149 | |
| 150 | return kernel_id |
| 151 | |
| 152 | def get_cached_kernel(self, model_name: str, operation: str) -> Optional[Dict[str, Any]]: |
| 153 | """Retrieve a cached kernel if available.""" |
nothing calls this directly
no test coverage detected