"""A consistent-hashing ring, and the machinery to measure what it actually
buys you.

Nodes are placed at positions on a circle of 2**32 slots. A key is owned by
the first node position clockwise from (or exactly at) the key's own
position. Adding a node inserts new positions; only the keys in the arcs
those positions cut off change owner. Nothing else moves.

There is NO random number generator anywhere in this file. Every position
comes from SHA-1 truncated to 32 bits, so every number this toy prints is
reproducible byte-for-byte on any machine. The "randomness" you see in the
output is the hash function's own, which is exactly the point: a uniform
hash does not produce a uniform *partition*.

    Ring(nodes, vnodes)      build a ring with `vnodes` points per node
    .add(node) / .remove(node)
    .route(key) -> node      the successor lookup, one bisect
    .route_all(keys) -> {key: node}
    .distribution(keys) -> {node: count}
    .arcs() -> {node: fraction of the ring owned}
    moved(before, after) -> {key: (old_owner, new_owner)}
"""

import hashlib
from bisect import bisect_left, bisect_right

HASH_BITS = 32
HASH_SPACE = 1 << HASH_BITS


def ring_hash(label):
    """Map a label onto the ring: SHA-1, keep the top 4 bytes as a big-endian
    unsigned int. Truncation is safe here because every bit of a SHA-1 digest
    is equally good; we just need 32 of them."""
    digest = hashlib.sha1(label.encode("utf-8")).digest()
    return int.from_bytes(digest[:4], "big")


class Ring:
    """A hash ring. `vnodes` is how many points each node occupies."""

    def __init__(self, nodes=(), vnodes=1):
        if vnodes < 1:
            raise ValueError("vnodes must be >= 1")
        self.vnodes = vnodes
        self.nodes = []
        self._points = []  # ring positions, kept sorted
        self._owners = []  # _owners[i] is the node owning _points[i]
        for node in nodes:
            self.add(node)

    # -- membership ------------------------------------------------------

    def _labels(self, node):
        """The labels whose hashes become this node's ring positions. The
        label, not the node name, is what gets hashed — that is the only
        reason one node can sit in many places."""
        return [f"{node}#{i}" for i in range(self.vnodes)]

    def add(self, node):
        if node in self.nodes:
            raise ValueError(f"{node!r} is already on the ring")
        self.nodes.append(node)
        for label in self._labels(node):
            pos = ring_hash(label)
            i = bisect_right(self._points, pos)
            self._points.insert(i, pos)
            self._owners.insert(i, node)

    def remove(self, node):
        if node not in self.nodes:
            raise ValueError(f"{node!r} is not on the ring")
        self.nodes.remove(node)
        kept = [(p, o) for p, o in zip(self._points, self._owners) if o != node]
        self._points = [p for p, _ in kept]
        self._owners = [o for _, o in kept]

    # -- routing ---------------------------------------------------------

    def route(self, key):
        """The successor lookup: first ring position >= hash(key), wrapping
        past the top of the ring back to the first position."""
        if not self._points:
            raise ValueError("the ring is empty")
        i = bisect_left(self._points, ring_hash(key))
        if i == len(self._points):
            i = 0  # past the last point: wrap to the first
        return self._owners[i]

    def route_all(self, keys):
        return {key: self.route(key) for key in keys}

    # -- measurement -----------------------------------------------------

    def distribution(self, keys):
        """How many of `keys` each node owns. Nodes owning none still
        appear, with 0 — a silently missing node would hide the worst case."""
        counts = {node: 0 for node in self.nodes}
        for key in keys:
            counts[self.route(key)] += 1
        return counts

    def arcs(self):
        """Fraction of the ring's 2**32 slots each node owns. This is the
        thing that actually determines load; the key counts merely sample
        it. A point at p owns the half-open arc (previous point, p]."""
        share = {node: 0 for node in self.nodes}
        for i, pos in enumerate(self._points):
            if len(self._points) == 1:
                span = HASH_SPACE
            else:
                span = (pos - self._points[i - 1]) % HASH_SPACE
            share[self._owners[i]] += span
        return {node: span / HASH_SPACE for node, span in share.items()}

    def points(self):
        """(position, node) pairs in ring order — for printing small rings."""
        return list(zip(self._points, self._owners))


def moved(before, after):
    """Keys whose owner changed between two routings of the same key set."""
    return {
        key: (owner, after[key])
        for key, owner in before.items()
        if after[key] != owner
    }


def balance(counts):
    """Summarise a distribution: (min, max, max/fair-share, max/min).

    `fair` is what a perfectly even partition would give each node, so
    max/fair is "how much more than its share the busiest node holds" — the
    number an operator actually has to provision for."""
    values = list(counts.values())
    total = sum(values)
    fair = total / len(values)
    lo, hi = min(values), max(values)
    return lo, hi, hi / fair, (hi / lo if lo else float("inf"))