Reset cache

spinn_front_end_common/interface/buffer_management/storage_objects/buffer_database.py

@@ -45,6 +45,14 @@ class BufferDatabase(BaseDatabase):
 
     __slots__ = ()
 
+    @classmethod
+    def reset_db(cls, reset_number: int) -> "BufferDatabase":
+        """
+        Retrieves a BufferDatabase for this segment.
+        """
+        database_file = cls.reset_file(reset_number)
+        return BufferDatabase(database_file)
+
     def clear_recording_region(
             self, x: int, y: int, p: int, region: int) -> bool:
         """

spinn_front_end_common/interface/provenance/provenance_reader.py

@@ -76,6 +76,14 @@ def __init__(self, provenance_data_path: Optional[str] = None):
         super().__init__(provenance_data_path, read_only=True,
                          row_factory=None, text_factory=None)
 
+    @classmethod
+    def reset_db(cls, reset_number: int) -> "ProvenanceReader":
+        """
+        Retrieves a ProvenanceReader for this segment.
+        """
+        database_file = cls.reset_file(reset_number)
+        return ProvenanceReader(database_file)
+
     def run_query(self, query: str, params: Iterable[_SqliteTypes] = ()
                   ) -> List[Sequence[_SqliteTypes]]:
         """

spinn_front_end_common/utilities/base_database.py

@@ -15,7 +15,7 @@
 import os
 import sqlite3
 import time
-from typing import Optional, Union
+from typing import Dict, Optional, Union
 from typing_extensions import TypeAlias
 from spinn_front_end_common.data import FecDataView
 from spinn_front_end_common.utilities.sqlite_db import SQLiteDB
@@ -25,6 +25,8 @@
 _SECONDS_TO_MICRO_SECONDS_CONVERSION = 1000
 _SqliteTypes: TypeAlias = Union[str, int, float, bytes, None]
 
+reset_cache: Dict[int, str] = {}
+
 
 def _timestamp():
     return int(time.time() * _SECONDS_TO_MICRO_SECONDS_CONVERSION)
@@ -65,6 +67,17 @@ def __init__(self, database_file: Optional[str] = None, *,
         super().__init__(
             self._database_file, read_only=read_only, row_factory=row_factory,
             text_factory=text_factory, ddl_file=_DDL_FILE)
+        reset = FecDataView.get_reset_number()
+        if (reset not in reset_cache or
+                reset_cache[reset] != self._database_file):
+            reset_cache[reset] = self._database_file
+
+    @classmethod
+    def reset_file(cls, reset_number: int) -> str:
+        """
+        Provides the database file uses during this reset
+        """
+        return reset_cache[reset_number]
 
     @classmethod
     def default_database_file(cls) -> str: