backends

flavor.psp.format_2025.backends

TODO: Add module docstring.

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

@abstractmethod
def close(self) -> None:
    """Close the bundle file."""
    pass

open abstractmethod

open(path: Path) -> None

Open the bundle file.

Source code in flavor/psp/format_2025/backends.py

@abstractmethod
def open(self, path: Path) -> None:
    """Open the bundle file."""
    pass

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

@abstractmethod
def read_at(self, offset: int, size: int) -> bytes | memoryview:
    """Read data at specific offset."""
    pass

read_slot abstractmethod

read_slot(descriptor: SlotDescriptor) -> bytes | memoryview

Read slot data based on descriptor.

Source code in flavor/psp/format_2025/backends.py

@abstractmethod
def read_slot(self, descriptor: SlotDescriptor) -> bytes | memoryview:
    """Read slot data based on descriptor."""
    pass

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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()