quoracle/quorums/expr.py

from typing import Dict, Iterator, Generic, List, Optional, Set, TypeVar
import itertools
import pulp


T = TypeVar('T')


def _min_hitting_set(sets: Iterator[Set[T]]) -> int:
    x_vars: Dict[T, pulp.LpVariable] = dict()
    next_id = itertools.count()

    problem = pulp.LpProblem("min_hitting_set", pulp.LpMinimize)
    for (i, xs) in enumerate(sets):
        for x in xs:
            if x not in x_vars:
                id = next(next_id)
                x_vars[x] = pulp.LpVariable(f'x{id}', cat=pulp.LpBinary)
        problem += sum(x_vars[x] for x in xs) >= 1

    problem += sum(x_vars.values())
    problem.solve(pulp.apis.PULP_CBC_CMD(msg=False))
    return int(sum(v.varValue for v in x_vars.values()))


class Expr(Generic[T]):
    def __add__(self, rhs: 'Expr[T]') -> 'Expr[T]':
        def _or(lhs: Expr[T], rhs: Expr[T]) -> 'Or[T]':
            if isinstance(lhs, Or) and isinstance(rhs, Or):
                return Or(lhs.es + rhs.es)
            elif isinstance(lhs, Or):
                return Or(lhs.es + [rhs])
            elif isinstance(rhs, Or):
                return Or([lhs] + rhs.es)
            else:
                return Or([lhs, rhs])


        return _or(self, rhs)

    def __mul__(self, rhs: 'Expr[T]') -> 'Expr[T]':
        def _and(lhs: Expr[T], rhs: Expr[T]) -> 'And[T]':
            if isinstance(lhs, And) and isinstance(rhs, And):
                return And(lhs.es + rhs.es)
            elif isinstance(lhs, And):
                return And(lhs.es + [rhs])
            elif isinstance(rhs, And):
                return And([lhs] + rhs.es)
            else:
                return And([lhs, rhs])

        return _and(self, rhs)

    def quorums(self) -> Iterator[Set[T]]:
        raise NotImplementedError

    def is_quorum(self, xs: Set[T]) -> bool:
        raise NotImplementedError

    def elements(self) -> Set[T]:
        return {node.x for node in self.nodes()}

    def nodes(self) -> Set['Node[T]']:
        raise NotImplementedError

    def resilience(self) -> int:
        if self.dup_free():
            return self._dup_free_min_failures() - 1
        else:
            return _min_hitting_set(self.quorums()) - 1

    def dual(self) -> 'Expr[T]':
        raise NotImplementedError

    def dup_free(self) -> bool:
        return len(self.nodes()) == self._num_leaves()

    def _num_leaves(self) -> int:
        raise NotImplementedError

    def _dup_free_min_failures(self) -> int:
        raise NotImplementedError


class Node(Expr[T]):
    def __init__(self,
                 x: T,
                 capacity: Optional[float] = None,
                 read_capacity: Optional[float] = None,
                 write_capacity: Optional[float] = None) -> None:
        self.x = x

        # A user either specifies capacity or (read_capacity and
        # write_capacity), but not both.
        if (capacity is None and
            read_capacity is None and
            write_capacity is None):
            self.read_capacity = 1.0
            self.write_capacity = 1.0
        elif (capacity is not None and
              read_capacity is None and
              write_capacity is None):
            self.read_capacity = capacity
            self.write_capacity = capacity
        elif (capacity is None and
              read_capacity is not None and
              write_capacity is not None):
            self.read_capacity = read_capacity
            self.write_capacity = write_capacity
        else:
            raise ValueError('You must specify capacity or (read_capacity '
                             'and write_capacity)')

    def __str__(self) -> str:
        return str(self.x)

    def __repr__(self) -> str:
        return f'Node({self.x})'

    def quorums(self) -> Iterator[Set[T]]:
        yield {self.x}

    def is_quorum(self, xs: Set[T]) -> bool:
        return self.x in xs

    def nodes(self) -> Set['Node[T]']:
        return {self}

    def dual(self) -> Expr:
        return self

    def _num_leaves(self) -> int:
        return 1

    def _dup_free_min_failures(self) -> int:
        return 1


class Or(Expr[T]):
    def __init__(self, es: List[Expr[T]]) -> None:
        if len(es) == 0:
            raise ValueError(f'Or cannot be constructed with an empty list')

        self.es = es

    def __str__(self) -> str:
        return '(' + ' + '.join(str(e) for e in self.es) + ')'

    def __repr__(self) -> str:
        return f'Or({self.es})'

    def quorums(self) -> Iterator[Set[T]]:
        for e in self.es:
            yield from e.quorums()

    def is_quorum(self, xs: Set[T]) -> bool:
        return any(e.is_quorum(xs) for e in self.es)

    def nodes(self) -> Set[Node[T]]:
        return set.union(*[e.nodes() for e in self.es])

    def dual(self) -> Expr:
        return And([e.dual() for e in self.es])

    def _num_leaves(self) -> int:
        return sum(e._num_leaves() for e in self.es)

    def _dup_free_min_failures(self) -> int:
        return sum(e._dup_free_min_failures() for e in self.es)


class And(Expr[T]):
    def __init__(self, es: List[Expr[T]]) -> None:
        if len(es) == 0:
            raise ValueError(f'And cannot be constructed with an empty list')

        self.es = es

    def __str__(self) -> str:
        return '(' + ' * '.join(str(e) for e in self.es) + ')'

    def __repr__(self) -> str:
        return f'And({self.es})'

    def quorums(self) -> Iterator[Set[T]]:
        for subquorums in itertools.product(*[e.quorums() for e in self.es]):
            yield set.union(*subquorums)

    def is_quorum(self, xs: Set[T]) -> bool:
        return all(e.is_quorum(xs) for e in self.es)

    def nodes(self) -> Set[Node[T]]:
        return set.union(*[e.nodes() for e in self.es])

    def dual(self) -> Expr:
        return Or([e.dual() for e in self.es])

    def _num_leaves(self) -> int:
        return sum(e._num_leaves() for e in self.es)

    def _dup_free_min_failures(self) -> int:
        return min(e._dup_free_min_failures() for e in self.es)

class Choose(Expr[T]):
    def __init__(self, k: int, es: List[Expr[T]]) -> None:
        if k <= 0 or k > len(es):
            raise ValueError(f'k must be in the range [1, {len(es)}]')

        self.k = k
        self.es = es

    def __str__(self) -> str:
        return f'choose{self.k}(' + ', '.join(str(e) for e in self.es) + ')'

    def __repr__(self) -> str:
        return f'Chose({self.k}, {self.es})'

    def quorums(self) -> Iterator[Set[T]]:
        for combo in itertools.combinations(self.es, self.k):
            for subquorums in itertools.product(*[e.quorums() for e in combo]):
                yield set.union(*subquorums)

    def is_quorum(self, xs: Set[T]) -> bool:
        return sum(1 if e.is_quorum(xs) else 0 for e in self.es) >= self.k

    def nodes(self) -> Set[Node[T]]:
        return set.union(*[e.nodes() for e in self.es])

    def dual(self) -> Expr:
        # TODO(mwhittaker): Prove that this is in fact the dual.
        return Choose(len(self.es) - self.k + 1, [e.dual() for e in self.es])

    def _num_leaves(self) -> int:
        return sum(e._num_leaves() for e in self.es)

    def _dup_free_min_failures(self) -> int:
        return sum(sorted(e._dup_free_min_failures() for e in self.es)[:self.k])


def choose(k: int, es: List[Expr[T]]) -> Expr[T]:
    if k == 1:
        return Or(es)
    elif k == len(es):
        return And(es)
    else:
        return Choose(k, es)


def majority(es: List[Expr[T]]) -> Expr[T]:
    return choose(len(es) // 2 + 1, es)