"""DPO rule composition.
The ``rcCommon`` / ``rcSuper`` / ``rcSub`` / ``rcParallel`` operators build a
composition expression with the infix ``*op*`` syntax, and
``RCEvaluator(...).eval(exp)`` evaluates it into the resulting rules.
rc = RCEvaluator(inputRules)
products = rc.eval(r1 *rcSuper* r2)
products = rc.eval(r1 *rcCommon(maximum=True)* r2)
The operators map onto shared/src/mod's ``RuleComposition.compose`` (via the
``RuleComposition`` bridge class). ``maximum`` / ``connected`` / ``includeEmpty``
apply to ``rcCommon``; ``allowPartial`` to ``rcSuper`` / ``rcSub``. The bare
operators use ``allowPartial=True`` and ``connected=True``; pass the call form
(e.g. ``rcSuper(allowPartial=False)``) to override.
"""
from typing import Iterable, List, Union
from .core import TSObject
from .rule import Rule
from .exceptions import LogicError
# Operator strings matching shared/src/mod CompositionOperator.
_COMMON = "common"
_SUPER = "super"
_SUB = "sub"
_PARALLEL = "parallel"
_class_ref = TSObject.get_class_ref("RuleComposition")
[docs]
class RCExp:
"""A rule-composition expression.
A :class:`Rule` is itself a valid leaf expression; the composition operators
combine two sub-expressions into a larger one. Evaluate with
:meth:`RCEvaluator.eval`.
"""
class _RCExpCompose(RCExp):
def __init__(self, first, second, operator: str,
maximum: bool, connected: bool, includeEmpty: bool, allowPartial: bool) -> None:
self.first = first
self.second = second
self.operator = operator
self.maximum = maximum
self.connected = connected
self.includeEmpty = includeEmpty
self.allowPartial = allowPartial
def __repr__(self) -> str:
return f"({self.first!r} *rc{self.operator.capitalize()}* {self.second!r})"
# ---------------------------------------------------------------------------
# rcCommon
# ---------------------------------------------------------------------------
class _RCCommonOpFirstBound:
def __init__(self, maximum: bool, connected: bool, includeEmpty: bool, first) -> None:
self.maximum = maximum
self.connected = connected
self.includeEmpty = includeEmpty
self.first = first
def __mul__(self, second) -> _RCExpCompose:
return _RCExpCompose(self.first, second, _COMMON,
self.maximum, self.connected, self.includeEmpty, False)
class _RCCommonOpArgsBound:
def __init__(self, maximum: bool, connected: bool, includeEmpty: bool) -> None:
self.maximum = maximum
self.connected = connected
self.includeEmpty = includeEmpty
def __rmul__(self, first) -> _RCCommonOpFirstBound:
return _RCCommonOpFirstBound(self.maximum, self.connected, self.includeEmpty, first)
class _RCCommonOp:
def __call__(self, maximum: bool = False, connected: bool = True,
includeEmpty: bool = False) -> _RCCommonOpArgsBound:
return _RCCommonOpArgsBound(maximum, connected, includeEmpty)
def __rmul__(self, first) -> _RCCommonOpFirstBound:
return first * self()
# ---------------------------------------------------------------------------
# rcParallel
# ---------------------------------------------------------------------------
class _RCParallelOpFirstBound:
def __init__(self, first) -> None:
self.first = first
def __mul__(self, second) -> _RCExpCompose:
return _RCExpCompose(self.first, second, _PARALLEL, False, True, False, False)
class _RCParallelOp:
def __rmul__(self, first) -> _RCParallelOpFirstBound:
return _RCParallelOpFirstBound(first)
# ---------------------------------------------------------------------------
# rcSub
# ---------------------------------------------------------------------------
class _RCSubOpFirstBound:
def __init__(self, allowPartial: bool, first) -> None:
self.allowPartial = allowPartial
self.first = first
def __mul__(self, second) -> _RCExpCompose:
return _RCExpCompose(self.first, second, _SUB, False, True, False, self.allowPartial)
class _RCSubOpArgsBound:
def __init__(self, allowPartial: bool) -> None:
self.allowPartial = allowPartial
def __rmul__(self, first) -> _RCSubOpFirstBound:
return _RCSubOpFirstBound(self.allowPartial, first)
class _RCSubOp:
def __call__(self, allowPartial: bool = True) -> _RCSubOpArgsBound:
return _RCSubOpArgsBound(allowPartial)
def __rmul__(self, first) -> _RCSubOpFirstBound:
return first * self()
# ---------------------------------------------------------------------------
# rcSuper
# ---------------------------------------------------------------------------
class _RCSuperOpFirstBound:
def __init__(self, allowPartial: bool, first) -> None:
self.allowPartial = allowPartial
self.first = first
def __mul__(self, second) -> _RCExpCompose:
return _RCExpCompose(self.first, second, _SUPER, False, True, False, self.allowPartial)
class _RCSuperOpArgsBound:
def __init__(self, allowPartial: bool) -> None:
self.allowPartial = allowPartial
def __rmul__(self, first) -> _RCSuperOpFirstBound:
return _RCSuperOpFirstBound(self.allowPartial, first)
class _RCSuperOp:
def __call__(self, allowPartial: bool = True) -> _RCSuperOpArgsBound:
return _RCSuperOpArgsBound(allowPartial)
def __rmul__(self, first) -> _RCSuperOpFirstBound:
return first * self()
# The operator singletons, used as ``r1 *rcSuper* r2`` (bare, default flags) or
# ``r1 *rcSuper(allowPartial=False)* r2``.
rcCommon = _RCCommonOp()
rcParallel = _RCParallelOp()
rcSub = _RCSubOp()
rcSuper = _RCSuperOp()
_RCExpType = Union[Rule, RCExp, Iterable["_RCExpType"]]
[docs]
class RCEvaluator:
"""Evaluates rule-composition expressions.
Constructed with an optional rule database, exposed as :attr:`ruleDatabase`;
the expressions carry their own concrete rules, so it is not otherwise needed.
Evaluated rules are collected in :attr:`createdRules`.
"""
def __init__(self, rules: Iterable[Rule] = ()) -> None:
self.ruleDatabase: List[Rule] = list(rules)
self.createdRules: List[Rule] = []
[docs]
def eval(self, exp: _RCExpType, *, onlyUnique: bool = True) -> List[Rule]:
"""Evaluate ``exp`` into the list of composed rules.
With ``onlyUnique`` (default) duplicate results are removed by canonical
form — the same deduplication ``RuleComposition.compose`` applies within a
single composition, extended across every composed pair in the expression.
"""
results = self._eval(exp)
if onlyUnique:
results = self._unique(results)
self.createdRules.extend(results)
return results
def _eval(self, exp: _RCExpType) -> List[Rule]:
if isinstance(exp, Rule):
return [exp]
if isinstance(exp, _RCExpCompose):
firsts = self._eval(exp.first)
seconds = self._eval(exp.second)
out: List[Rule] = []
for a in firsts:
for b in seconds:
out.extend(self._compose_pair(a, b, exp))
return out
if isinstance(exp, (list, tuple)):
out = []
for e in exp:
out.extend(self._eval(e))
return out
raise LogicError(
"Cannot evaluate rule-composition expression of type '%s'." % type(exp).__name__)
@staticmethod
def _compose_pair(first: Rule, second: Rule, exp: _RCExpCompose) -> List[Rule]:
js_rules = TSObject.wrap_exception(lambda: TSObject.call_static(
_class_ref, "compose", first._ref, second._ref, exp.operator,
exp.maximum, exp.connected, exp.includeEmpty, exp.allowPartial))
return [Rule.wrap(r) for r in js_rules]
@staticmethod
def _unique(rules: List[Rule]) -> List[Rule]:
seen = set()
unique: List[Rule] = []
for r in rules:
key = r._ref.getCanonicalKey()
if key not in seen:
seen.add(key)
unique.append(r)
return unique