Algorithm Search⌘K

Forward star in Python

Edges sorted by tail vertex with a per-vertex index of first outgoing edge.

File
forward_star.py, 88 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
forward_star.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/forward-star/. The authors are listed in prov/83VY7T.jsonl.

from collections.abc import Iterable


def _prefix_counts(n: int, keys: Iterable[int]) -> list[int]:
    """p[i] = the number of keys less than i, for i in 0 .. n."""
    p = [0] * (n + 1)
    for k in keys:
        p[k + 1] += 1
    for i in range(1, n + 1):
        p[i] += p[i - 1]
    return p


class ForwardStar:
    """A static directed network: arcs sorted by tail, named 0 .. m - 1.

    point[i] is the name of the first arc out of i; trace[rpoint[j]:rpoint[j + 1]]
    lists the names of the arcs into j. Building takes Theta(n + m) time,
    out_arcs and in_arcs take time linear in their output, degrees O(1).
    Parallel arcs and self-loops are allowed.
    """

    def __init__(self, n: int, arcs: Iterable[tuple[int, int, float]]) -> None:
        if n < 0:
            raise ValueError("n must be at least 0")
        given = list(arcs)
        for t, h, _ in given:
            if not (0 <= t < n and 0 <= h < n):
                raise IndexError("vertex out of range")
        self.n, self.m = n, len(given)
        start = _prefix_counts(n, (t for t, _, _ in given))
        pos = start[:n]
        order: list[tuple[int, int, float]] = [(0, 0, 0)] * self.m
        for a in given:  # stable counting sort by tail
            order[pos[a[0]]] = a  # @line L1
            pos[a[0]] += 1
        self.tail = [t for t, _, _ in order]
        self.head = [h for _, h, _ in order]
        self.cost = [c for _, _, c in order]
        self.flow: list[float] = [0] * self.m
        self.point = start  # @line L2
        self.rpoint = _prefix_counts(n, self.head)  # @line L3
        rpos = self.rpoint[:n]  # @line L4
        self.trace = [0] * self.m
        for e in range(self.m):  # @line L5
            j = self.head[e]  # @line L6
            self.trace[rpos[j]] = e
            rpos[j] += 1

    def _check(self, i: int) -> None:
        if not 0 <= i < self.n:
            raise IndexError("vertex out of range")

    def out_arcs(self, i: int) -> list[int]:
        self._check(i)
        return list(range(self.point[i], self.point[i + 1]))  # @line L7

    def in_arcs(self, j: int) -> list[int]:
        self._check(j)
        return [self.trace[k] for k in range(self.rpoint[j], self.rpoint[j + 1])]  # @line L8

    def out_degree(self, i: int) -> int:
        self._check(i)
        return self.point[i + 1] - self.point[i]  # @line L9

    def in_degree(self, j: int) -> int:
        self._check(j)
        return self.rpoint[j + 1] - self.rpoint[j]  # @line L10

    def arcs(self) -> list[tuple[int, int, float]]:
        return list(zip(self.tail, self.head, self.cost))

    def check_invariant(self) -> None:
        n, m = self.n, self.m
        assert self.point[0] == 0 and self.point[n] == m
        assert self.rpoint[0] == 0 and self.rpoint[n] == m
        for i in range(n):
            assert self.point[i] <= self.point[i + 1]
            assert self.rpoint[i] <= self.rpoint[i + 1]
            for e in range(self.point[i], self.point[i + 1]):
                assert self.tail[e] == i
            seg = self.trace[self.rpoint[i] : self.rpoint[i + 1]]
            assert all(seg[k] < seg[k + 1] for k in range(len(seg) - 1))
            assert all(self.head[e] == i for e in seg)
        assert sorted(self.trace) == list(range(m))
test_forward_star.py 56 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/forward-star/. The authors are listed in prov/83VY7T.jsonl.

import random
from collections import Counter

import pytest

from forward_star import ForwardStar


def test_example() -> None:
    g = ForwardStar(5, [(3, 4, 2), (0, 2, 5), (2, 3, 1), (3, 1, 4), (0, 1, 3), (1, 2, 1)])
    g.check_invariant()
    assert g.arcs() == [(0, 2, 5), (0, 1, 3), (1, 2, 1), (2, 3, 1), (3, 4, 2), (3, 1, 4)]
    assert g.point == [0, 2, 3, 4, 6, 6]
    assert g.rpoint == [0, 0, 2, 4, 5, 6]
    assert g.trace == [1, 5, 0, 2, 3, 4]
    assert g.out_arcs(3) == [4, 5]
    assert g.in_arcs(1) == [1, 5]
    g.flow[5] = 7
    assert [g.flow[e] for e in g.out_arcs(3)] == [0, 7]
    assert [g.flow[e] for e in g.in_arcs(1)] == [0, 7]
    assert [g.out_degree(i) for i in range(5)] == [2, 1, 1, 2, 0]
    assert [g.in_degree(i) for i in range(5)] == [0, 2, 2, 1, 1]


def test_edge_cases() -> None:
    g = ForwardStar(0, [])
    assert g.point == [0] and g.trace == []
    h = ForwardStar(2, [(0, 1, 1), (0, 1, 2), (1, 1, 0)])
    h.check_invariant()
    assert h.out_arcs(0) == [0, 1] and h.in_arcs(1) == [0, 1, 2]
    assert h.out_arcs(1) == [2] and h.in_arcs(0) == []
    with pytest.raises(IndexError):
        ForwardStar(2, [(0, 2, 1)])
    with pytest.raises(ValueError):
        ForwardStar(-1, [])
    with pytest.raises(IndexError):
        h.out_arcs(2)


def test_random_against_multiset_model() -> None:
    rng = random.Random(8377)
    for _ in range(600):
        n = rng.randint(1, 7)
        arcs = [(rng.randrange(n), rng.randrange(n), rng.randint(0, 9)) for _ in range(rng.randint(0, 20))]
        g = ForwardStar(n, arcs)
        g.check_invariant()
        assert Counter(g.arcs()) == Counter(arcs)
        # stability: arcs with the same tail keep the input order
        for i in range(n):
            assert [g.arcs()[e] for e in g.out_arcs(i)] == [a for a in arcs if a[0] == i]
            assert Counter(g.arcs()[e] for e in g.in_arcs(i)) == Counter(a for a in arcs if a[1] == i)
            assert g.out_degree(i) == sum(a[0] == i for a in arcs)
            assert g.in_degree(i) == sum(a[1] == i for a in arcs)

What the next level needs

  • the tests did not run in this build

Read the full page about Forward star