(self, path: Path)
| 12411 | return all(metadata.get(key) == value for key, value in self._metadata.items()) |
| 12412 | |
| 12413 | def _read_entry_header(self, path: Path) -> Optional["SequenceDiskCache.Header"]: |
| 12414 | with self._safe_open(path) as tensors: |
| 12415 | if not self._metadata_compatible(tensors.metadata()): |
| 12416 | return None |
| 12417 | tokens_array = tensors.get_tensor(self.TENSOR_TOKENS) |
| 12418 | has_prompt_logits = self.TENSOR_PROMPT_LOGITS in tensors.keys() |
| 12419 | tokens = tuple(int(token) for token in tokens_array.tolist()) |
| 12420 | if len(tokens) < self.min_tokens: |
| 12421 | return None |
| 12422 | return SequenceDiskCache.Header( |
| 12423 | tokens=tokens, |
| 12424 | has_prompt_logits=has_prompt_logits, |
| 12425 | ) |
| 12426 | |
| 12427 | def _read_entry_payload( |
| 12428 | self, |
no test coverage detected