orchestrata_ultimate.refactor.codemod._fast_fallback

Reference › orchestrata_ultimate › refactor › codemod › _fast_fallback

Pure-Python implementations of the functions that the codemod_fast C extension provides, used when the native extension is not available. No other reference page imports it. The functions are csr_build (build forward and reverse compressed-sparse-row arrays from an edge list), csr_dependencies and csr_dependents (read a node's neighbours), csr_bfs (reachable nodes with an optional depth limit), xxhash3_64 (a 16-character hex digest of bytes, using the xxhash package, which is a hard dependency and raises ImportError if missing) and mmap_load (reads a file's bytes; the native version uses mmap). The _CSRIndex and the parse cache are the kinds of caller these serve. The daemon's identity reply reports whether the native extension or this fallback is in use.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/refactor/codemod/_fast_fallback.py

"""Pure-Python fallbacks for codemod_fast C extension functions.

This module provides the same API as ``codemod_fast`` but without requiring
the C extension to be built. Import from here when the C extension is not
available or when testing parity.
"""

from __future__ import annotations

import array
from collections import deque
from pathlib import Path


def csr_build(
    node_count: int, edges: list[tuple[int, int]]
) -> tuple[array.array, array.array]:
    """Build CSR adjacency arrays from edge list.

    Args:
        node_count: Number of nodes in the graph.
        edges: List of (source, destination) tuples.

    Returns:
        Tuple of (offsets, neighbors) array.array('i') objects.
    """
    out_degree = [0] * node_count
    for src, _ in edges:
        out_degree[src] += 1

    offsets = array.array("i", [0] * (node_count + 1))
    for i in range(node_count):
        offsets[i + 1] = offsets[i] + out_degree[i]

    neighbors = array.array("i", [0] * max(len(edges), 1))
    cursor = list(offsets[:-1])
    for src, dst in edges:
        neighbors[cursor[src]] = dst
        cursor[src] += 1

    return offsets, neighbors


def csr_dependencies(
    offsets: array.array,
    neighbors: array.array,
    node_id: int,
) -> list[int]:
    """Get forward dependencies (outgoing edges) for a node.

    Args:
        offsets: CSR offsets array.
        neighbors: CSR neighbors array.
        node_id: Node to query.

    Returns:
        List of neighbor node IDs.
    """
    start = offsets[node_id]
    end = offsets[node_id + 1]
    return list(neighbors[start:end])


def csr_dependents(
    transpose_offsets: array.array,
    transpose_neighbors: array.array,
    node_id: int,
) -> list[int]:
    """Get reverse dependents (incoming edges) for a node.

    Args:
        transpose_offsets: Transpose CSR offsets array.
        transpose_neighbors: Transpose CSR neighbors array.
        node_id: Node to query.

    Returns:
        List of dependent node IDs.
    """
    start = transpose_offsets[node_id]
    end = transpose_offsets[node_id + 1]
    return list(transpose_neighbors[start:end])


def csr_bfs(
    offsets: array.array,
    neighbors: array.array,
    start: int,
    max_depth: int = -1,
) -> set[int]:
    """BFS transitive closure on CSR index.

    Args:
        offsets: CSR offsets array.
        neighbors: CSR neighbors array.
        start: Starting node ID.
        max_depth: Maximum BFS depth (-1 for unlimited).

    Returns:
        Set of all reachable node IDs (including start).
    """
    visited: set[int] = set()
    queue: deque[tuple[int, int]] = deque([(start, 0)])

    while queue:
        node, depth = queue.popleft()
        if node in visited:
            continue
        visited.add(node)
        if 0 <= max_depth <= depth:
            continue
        lo, hi = offsets[node], offsets[node + 1]
        for i in range(lo, hi):
            nb = neighbors[i]
            if nb not in visited:
                queue.append((nb, depth + 1))

    return visited


def xxhash3_64(data: bytes) -> str:
    """Compute 64-bit hash of data using xxhash.xxh3_64.

    This is the single implementation. xxhash is a hard dependency.

    Args:
        data: Bytes to hash.

    Returns:
        16-character hex digest string.

    Raises:
        ImportError: If xxhash is not installed.
    """
    import xxhash

    return xxhash.xxh3_64(data).hexdigest()


def mmap_load(path: str | Path) -> bytes:
    """Load file contents.

    In the pure-Python fallback, this simply reads the file. The C extension
    uses mmap() for zero-copy loading.

    Args:
        path: Path to the file.

    Returns:
        File contents as bytes.
    """
    return Path(path).read_bytes()