Skip to content

backends

🤖 AI-Generated Content

This documentation was generated with AI assistance and is still being audited. Some, or potentially a lot, of this information may be inaccurate. Learn more.

flavor.psp.format_2025.backends

Classes

Backend

Bases: ABC

Abstract base class for PSPF bundle access backends.

Functions
close abstractmethod 
close() -> None

Close the bundle file.

Source code in flavor/psp/format_2025/backends.py 
39
40
41
@abstractmethod
def close(self) -> None:
    """Close the bundle file."""
open abstractmethod 
open(path: Path) -> None

Open the bundle file.

Source code in flavor/psp/format_2025/backends.py 
35
36
37
@abstractmethod
def open(self, path: Path) -> None:
    """Open the bundle file."""
read_at abstractmethod 
read_at(offset: int, size: int) -> bytes | memoryview

Read data at specific offset.

Source code in flavor/psp/format_2025/backends.py 
43
44
45
@abstractmethod
def read_at(self, offset: int, size: int) -> bytes | memoryview:
    """Read data at specific offset."""
read_slot abstractmethod 
read_slot(descriptor: SlotDescriptor) -> bytes | memoryview

Read slot data based on descriptor.

Source code in flavor/psp/format_2025/backends.py 
47
48
49
@abstractmethod
def read_slot(self, descriptor: SlotDescriptor) -> bytes | memoryview:
    """Read slot data based on descriptor."""
stream_slot 
stream_slot(
    descriptor: SlotDescriptor,
    chunk_size: int = DEFAULT_CHUNK_SIZE,
) -> Generator[bytes | memoryview, None, None]

Stream slot data in chunks.

Source code in flavor/psp/format_2025/backends.py 
51
52
53
54
55
56
57
58
59
60
61
62
63
def stream_slot(
    self, descriptor: SlotDescriptor, chunk_size: int = DEFAULT_CHUNK_SIZE
) -> Generator[bytes | memoryview, None, None]:
    """Stream slot data in chunks."""
    offset = descriptor.offset
    remaining = descriptor.size

    while remaining > 0:
        to_read = min(chunk_size, remaining)
        chunk = self.read_at(offset, to_read)
        yield chunk
        offset += to_read
        remaining -= to_read

FileBackend 

FileBackend()

Bases: Backend

Traditional file I/O backend.

Source code in flavor/psp/format_2025/backends.py 
213
214
215
216
def __init__(self) -> None:
    self.file: BinaryIO | None = None
    self.path: Path | None = None
    self._cache: dict[tuple[int, int], bytes] = {}  # Simple cache for frequently accessed regions
Functions
close 
close() -> None

Close the file.

Source code in flavor/psp/format_2025/backends.py 
235
236
237
238
239
240
241
242
243
244
245
246
def close(self) -> None:
    """Close the file."""
    logger.debug(
        "🔒 Closing file backend",
        path=str(self.path) if self.path else None,
        cache_entries=len(self._cache),
    )

    if self.file:
        self.file.close()
        self.file = None
    self._cache.clear()
open 
open(path: Path) -> None

Open file with buffered I/O.

Source code in flavor/psp/format_2025/backends.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def open(self, path: Path) -> None:
    """Open file with buffered I/O."""
    start_time = time.perf_counter()
    self.path = path
    file_size = path.stat().st_size
    logger.debug(
        "📖 Opening buffered file backend",
        path=str(path),
        size_bytes=file_size,
        buffer_size=64 * 1024,
    )

    # Use buffered I/O for better performance
    self.file = path.open("rb", buffering=64 * 1024)

    time.perf_counter() - start_time
read_at 
read_at(offset: int, size: int) -> bytes

Read data at specific offset.

Source code in flavor/psp/format_2025/backends.py 
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
def read_at(self, offset: int, size: int) -> bytes:
    """Read data at specific offset."""
    start_time = time.perf_counter()

    if not self.file:
        logger.error("❌ Backend not opened")
        raise RuntimeError("Backend not opened")

    # Check cache first
    cache_key = (offset, size)
    if cache_key in self._cache:
        logger.debug("⚡ Cache hit", offset=offset, size=size)
        return self._cache[cache_key]

    # Read from file
    self.file.seek(offset)
    data = self.file.read(size)

    # Cache small reads
    if size <= 4096:  # Cache small reads
        self._cache[cache_key] = data
        # Limit cache size
        if len(self._cache) > 100:
            # Remove oldest entries (simple FIFO)
            evicted = 0
            for _ in range(20):
                self._cache.pop(next(iter(self._cache)))
                evicted += 1
            logger.debug("🗑️ Cache eviction", evicted=evicted, remaining=len(self._cache))

    elapsed = time.perf_counter() - start_time
    logger.debug(
        "📂 File read_at",
        offset=offset,
        size=size,
        elapsed_us=elapsed * 1000000,
        cached=size <= 4096,
    )

    return data
read_slot 
read_slot(descriptor: SlotDescriptor) -> bytes

Read slot data.

Source code in flavor/psp/format_2025/backends.py 
289
290
291
def read_slot(self, descriptor: SlotDescriptor) -> bytes:
    """Read slot data."""
    return self.read_at(descriptor.offset, descriptor.size)

HybridBackend 

HybridBackend(header_size: int = 1024 * 1024)

Bases: Backend

Hybrid backend - uses mmap for index/metadata, file I/O for slots.

Source code in flavor/psp/format_2025/backends.py 
362
363
364
365
366
367
def __init__(self, header_size: int = 1024 * 1024) -> None:  # 1MB default
    self.header_size = header_size
    self.file: BinaryIO | None = None
    self.header_mmap: mmap.mmap | None = None
    self.path: Path | None = None
    self._views: list[memoryview] = []  # Track memory views
Functions
close 
close() -> None

Close mappings and file.

Source code in flavor/psp/format_2025/backends.py 
381
382
383
384
385
386
387
388
389
390
391
392
393
def close(self) -> None:
    """Close mappings and file."""
    # Release all memory views first
    self._views.clear()

    if self.header_mmap:
        with suppress(BufferError):
            # If views still exist, just clear our reference
            self.header_mmap.close()
        self.header_mmap = None
    if self.file:
        self.file.close()
        self.file = None
open 
open(path: Path) -> None

Open with partial memory mapping.

Source code in flavor/psp/format_2025/backends.py 
369
370
371
372
373
374
375
376
377
378
379
def open(self, path: Path) -> None:
    """Open with partial memory mapping."""
    self.path = path
    self.file = path.open("rb")

    # Get file size
    file_size = path.stat().st_size

    # Memory-map just the header region
    map_size = min(self.header_size, file_size)
    self.header_mmap = mmap.mmap(self.file.fileno(), map_size, access=mmap.ACCESS_READ)
read_at 
read_at(offset: int, size: int) -> bytes | memoryview

Read using mmap for header, file I/O for rest.

Source code in flavor/psp/format_2025/backends.py 
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def read_at(self, offset: int, size: int) -> bytes | memoryview:
    """Read using mmap for header, file I/O for rest."""
    if not self.file:
        raise RuntimeError("Backend not opened")

    # Use mmap for header region
    if self.header_mmap and offset + size <= len(self.header_mmap):
        view = memoryview(self.header_mmap)[offset : offset + size]
        self._views.append(view)  # Track for cleanup
        return view

    # Use file I/O for slot data
    self.file.seek(offset)
    return self.file.read(size)
read_slot 
read_slot(descriptor: SlotDescriptor) -> bytes | memoryview

Read slot using appropriate method.

Source code in flavor/psp/format_2025/backends.py 
410
411
412
def read_slot(self, descriptor: SlotDescriptor) -> bytes | memoryview:
    """Read slot using appropriate method."""
    return self.read_at(descriptor.offset, descriptor.size)

MMapBackend 

MMapBackend()

Bases: Backend

Memory-mapped file access backend.

Source code in flavor/psp/format_2025/backends.py 
69
70
71
72
73
def __init__(self) -> None:
    self.file: BinaryIO | None = None
    self.mmap: mmap.mmap | None = None
    self.path: Path | None = None
    self._views: list[memoryview] = []  # Track memory views for cleanup
Functions
close 
close() -> None

Close memory map and file.

Source code in flavor/psp/format_2025/backends.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
def close(self) -> None:
    """Close memory map and file."""
    logger.debug(
        "🔒 Closing mmap backend",
        path=str(self.path) if self.path else None,
        tracked_views=len(self._views),
    )

    # Release all memory views first
    self._views.clear()

    if self.mmap:
        with suppress(BufferError):
            # BufferError expected if external code holds memoryview references
            # The mmap will be cleaned up by Python's GC when all references are released
            self.mmap.close()
        self.mmap = None
    if self.file:
        self.file.close()
        self.file = None
open 
open(path: Path) -> None

Open file and create memory mapping.

Source code in flavor/psp/format_2025/backends.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def open(self, path: Path) -> None:
    """Open file and create memory mapping."""
    start_time = time.perf_counter()
    self.path = path
    file_size = path.stat().st_size
    logger.debug(
        "🗺️ Opening mmap backend",
        path=str(path),
        size_bytes=file_size,
        size_mb=file_size / 1024 / 1024,
    )

    self.file = path.open("rb")

    # Create read-only memory map
    self.mmap = mmap.mmap(
        self.file.fileno(),
        0,  # Map entire file
        access=mmap.ACCESS_READ,
    )

    # Platform-specific optimizations
    if hasattr(mmap, "MADV_SEQUENTIAL"):
        # Hint for sequential access on Unix
        self.mmap.madvise(mmap.MADV_SEQUENTIAL)

    elapsed = time.perf_counter() - start_time
    logger.debug(
        "🚀 Preloading complete",
        elapsed_ms=elapsed * 1000,
        pages=file_size // DEFAULT_PAGE_SIZE,
    )
prefetch 
prefetch(offset: int, size: int) -> None

Hint to OS to prefetch pages.

Source code in flavor/psp/format_2025/backends.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def prefetch(self, offset: int, size: int) -> None:
    """Hint to OS to prefetch pages."""
    logger.debug(
        "📥 Prefetching pages",
        offset=offset,
        size=size,
        pages=size // DEFAULT_PAGE_SIZE,
    )

    if hasattr(os, "posix_fadvise") and hasattr(os, "POSIX_FADV_WILLNEED") and self.file:
        # Linux: hint that we'll need this data soon
        os.posix_fadvise(self.file.fileno(), offset, size, os.POSIX_FADV_WILLNEED)  # type: ignore[attr-defined]
    elif sys.platform == "win32" and self.mmap:
        # Windows: touch pages to load them
        # This is less efficient but works
        view = memoryview(self.mmap)[offset : offset + 1]
        _ = view[0]  # Touch first byte to trigger page load
    else:
        logger.debug("⚠️ Prefetch not available on this platform")
read_at 
read_at(offset: int, size: int) -> memoryview

Return a memory view without copying data.

Source code in flavor/psp/format_2025/backends.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
def read_at(self, offset: int, size: int) -> memoryview:
    """Return a memory view without copying data."""
    start_time = time.perf_counter()

    if not self.mmap:
        logger.error("❌ Backend not opened")
        raise RuntimeError("Backend not opened")

    # Validate bounds
    if offset < 0:
        logger.error("❌ Invalid offset", offset=offset)
        raise ValueError(f"Negative offset not allowed: {offset}")
    if size < 0:
        logger.error("❌ Invalid size", size=size)
        raise ValueError(f"Negative size not allowed: {size}")
    if offset + size > len(self.mmap):
        logger.error(
            "❌ Read beyond bounds",
            offset=offset,
            size=size,
            file_size=len(self.mmap),
        )
        raise ValueError(
            f"Read beyond file bounds: offset={offset}, size={size}, file_size={len(self.mmap)}"
        )

    # Return a view into the mapped memory (zero-copy)
    view = memoryview(self.mmap)[offset : offset + size]
    self._views.append(view)  # Track for cleanup

    elapsed = time.perf_counter() - start_time
    logger.debug(
        "🔍 MMap read_at",
        offset=offset,
        size=size,
        elapsed_us=elapsed * 1000000,
        zero_copy=True,
    )

    return view
read_slot 
read_slot(descriptor: SlotDescriptor) -> memoryview

Read slot as memory view.

Source code in flavor/psp/format_2025/backends.py 
170
171
172
def read_slot(self, descriptor: SlotDescriptor) -> memoryview:
    """Read slot as memory view."""
    return self.read_at(descriptor.offset, descriptor.size)
view_at 
view_at(offset: int, size: int) -> memoryview

Get a zero-copy view of data at offset (same as read_at for mmap).

Source code in flavor/psp/format_2025/backends.py 
174
175
176
def view_at(self, offset: int, size: int) -> memoryview:
    """Get a zero-copy view of data at offset (same as read_at for mmap)."""
    return self.read_at(offset, size)

StreamBackend 

StreamBackend(chunk_size: int = DEFAULT_CHUNK_SIZE)

Bases: Backend

Streaming backend - never loads full slots into memory.

Source code in flavor/psp/format_2025/backends.py 
308
309
310
311
def __init__(self, chunk_size: int = DEFAULT_CHUNK_SIZE) -> None:
    self.file: BinaryIO | None = None
    self.path: Path | None = None
    self.chunk_size = chunk_size
Functions
close 
close() -> None

Close the file.

Source code in flavor/psp/format_2025/backends.py 
318
319
320
321
322
def close(self) -> None:
    """Close the file."""
    if self.file:
        self.file.close()
        self.file = None
open 
open(path: Path) -> None

Open file for streaming.

Source code in flavor/psp/format_2025/backends.py 
313
314
315
316
def open(self, path: Path) -> None:
    """Open file for streaming."""
    self.path = path
    self.file = path.open("rb", buffering=self.chunk_size)
read_at 
read_at(offset: int, size: int) -> bytes

Read data at specific offset - limited to chunk size.

Source code in flavor/psp/format_2025/backends.py 
324
325
326
327
328
329
330
331
332
333
def read_at(self, offset: int, size: int) -> bytes:
    """Read data at specific offset - limited to chunk size."""
    if not self.file:
        raise RuntimeError("Backend not opened")

    # Limit read size for streaming
    read_size = min(size, self.chunk_size)

    self.file.seek(offset)
    return self.file.read(read_size)
read_slot 
read_slot(descriptor: SlotDescriptor) -> bytes

Read only first chunk of slot for streaming.

Source code in flavor/psp/format_2025/backends.py 
335
336
337
338
def read_slot(self, descriptor: SlotDescriptor) -> bytes:
    """Read only first chunk of slot for streaming."""
    # For streaming, we don't read the whole slot at once
    return self.read_at(descriptor.offset, min(descriptor.size, self.chunk_size))
stream_slot 
stream_slot(
    descriptor: SlotDescriptor,
    chunk_size: int | None = None,
) -> Generator[bytes | memoryview, None, None]

Stream slot data in chunks.

Source code in flavor/psp/format_2025/backends.py 
340
341
342
343
344
345
def stream_slot(
    self, descriptor: SlotDescriptor, chunk_size: int | None = None
) -> Generator[bytes | memoryview, None, None]:
    """Stream slot data in chunks."""
    chunk_size = chunk_size or self.chunk_size
    return super().stream_slot(descriptor, chunk_size)

Functions

create_backend 

create_backend(
    mode: int = ACCESS_AUTO, path: Path | None = None
) -> Backend

Factory function to create the appropriate backend.

Source code in flavor/psp/format_2025/backends.py 
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def create_backend(mode: int = ACCESS_AUTO, path: Path | None = None) -> Backend:
    """Factory function to create the appropriate backend."""

    if mode == ACCESS_AUTO:
        # Auto-select based on file size and platform
        if path and path.exists():
            file_size = path.stat().st_size

            # Use mmap for files over 1MB
            if file_size > 1024 * 1024:
                mode = ACCESS_MMAP
                logger.debug(
                    "🤖 Auto-selected mmap backend",
                    file_size_mb=file_size / 1024 / 1024,
                )
            # Use streaming for very large files on limited memory
            elif file_size > 100 * 1024 * 1024 and sys.platform == "win32":
                mode = ACCESS_STREAM
                logger.debug(
                    "🤖 Auto-selected stream backend",
                    file_size_mb=file_size / 1024 / 1024,
                    platform=sys.platform,
                )
            else:
                mode = ACCESS_FILE
                logger.debug("🤖 Auto-selected file backend", file_size_kb=file_size / 1024)
        else:
            mode = ACCESS_FILE
            logger.debug("🤖 Default to file backend", path_exists=False)

    # Create the appropriate backend
    if mode == ACCESS_MMAP:
        return MMapBackend()
    elif mode == ACCESS_STREAM:
        return StreamBackend()
    elif mode == ACCESS_FILE:
        return FileBackend()
    else:
        # Default to hybrid for unknown modes
        return HybridBackend()