"""
The feedback category of monoidal streams over a symmetric monoidal category.
We adapted the definition of intensional streams from :cite:t:`DiLavoreEtAl22`.
## Summary
.. autosummary::
:template: class.rst
:nosignatures:
:toctree:
Ty
Stream
## Note
Monoidal streams form a feedback category as follows:
>>> from discopy import feedback
>>> from discopy.monoidal import Equation
>>> x, y, m = map(feedback.Ty, "xym")
>>> f = feedback.Box('f', x @ m.delay(), y @ m)
>>> fb = f.feedback()
>>> X, Y, M = [Ty.sequence(symmetric.Ty(n)) for n in "xym"]
>>> Ff = Stream.sequence("f", X @ M.delay(), Y @ M)
>>> F = feedback.Functor(ob_map={x: X, y: Y, m: M}, ar_map={f: Ff},
... cod=Stream)
>>> Equation(fb, F(fb).unroll(2).now, symbol="$\\\\mapsto$").draw(
... doctest="docs/_static/stream/feedback-to-stream.svg")
.. image:: /_static/stream/feedback-to-stream.svg
:align: center
## Examples
### Fibonacci
We can define the Fibonacci sequence as a feedback diagram interpreted in the
category of streams of python types and functions.
>>> from discopy import *
>>> from discopy.feedback import *
>>> X = Ty('X')
>>> fby, wait = FollowedBy(X), Swap(X, X.d).feedback()
>>> zero, one = Box('zero', Ty(), X.head), Box('one', Ty(), X.head)
>>> copy, plus = Copy(X), Box('plus', X @ X, X)
>>> @Diagram.feedback
... @Diagram.from_callable(X.d, X @ X)
... def fib(x):
... y = fby(zero(), plus.d(fby.d(one.d(), wait.d(x)), x))
... return (y, y)
>>> fib_ = (copy.d >> one.d @ wait.d @ X.d
... >> fby.d @ X.d
... >> plus.d
... >> zero @ X.d
... >> fby >> copy).feedback()
>>> assert Equation(fib.arg, fib_.arg)
>>> fib_.draw(wire_labels=False, figsize=(5, 5),
... doctest="docs/_static/stream/fibonacci-feedback.svg")
.. image:: /_static/stream/fibonacci-feedback.svg
:align: center
>>> cod = stream.Stream[python.Function]
>>> F = feedback.Functor(
... ob_map={X: int},
... ar_map={zero: cod.singleton(python.Function(lambda: 0, (), int)),
... one: cod.singleton(python.Function(lambda: 1, (), int)),
... plus: lambda x, y: x + y}, cod=cod)
>>> assert F(fib).unroll(9).now()[:10] == (0, 1, 1, 2, 3, 5, 8, 13, 21, 34)
### Random walk
We can define a simple random walk as a feedback diagram interpreted in the
category of streams of python types and probabilistic functions.
>>> from random import choice, seed; seed(420)
>>> rand = Box('rand', Ty(), X)
>>> F.ar_map[rand] = lambda: choice([-1, +1])
>>> @Diagram.feedback
... @Diagram.from_callable(X.d, X @ X)
... def walk(x):
... x = plus.d(rand.d(), x)
... x = fby(zero(), x)
... return (x, x)
>>> walk.draw(wire_labels=False, figsize=(5, 5),
... doctest="docs/_static/stream/random-walk-feedback.svg")
.. image:: /_static/stream/random-walk-feedback.svg
:align: center
>>> assert F(walk).unroll(9).now()[:10] == (0, -1, 0, 1, 2, 1, 0, -1, 0, 1)
>>> assert F(walk).unroll(9).now()[:10] == (0, -1, -2, -1, 0, 1, 0, 1, 2, 1)
>>> assert F(walk).unroll(9).now()[:10] == (0, -1, 0, 1, 0, 1, 0, -1, 0, -1)
## Axioms
Note that we can only check equality of streams up to a finite number of steps.
>>> from discopy.stream import *
>>> all_eq = lambda xs: len(set(xs)) == 1
>>> eq_up_to_n = lambda *xs, n=3: all_eq(x.unroll(n).now for x in xs)
>>> x, y, z, w, m, n, o = map(Ty.sequence, "xyzwmno")
>>> f = Stream.sequence('f', x, y, m)
>>> g = Stream.sequence('g', y, z, n)
>>> h = Stream.sequence('h', z, w, o)
* Unitality and associativity hold on the nose:
>>> _id = Stream.id
>>> assert eq_up_to_n(f @ _id(), f, _id() @ f)
>>> assert eq_up_to_n(f >> _id(f.cod), f, _id(f.dom) >> f)
>>> assert eq_up_to_n((f >> g) >> h), (f >> (g >> h))
>>> ((f >> g) >> h).now.draw(
... doctest="docs/_static/stream/feedback-associativity.svg")
.. image:: /_static/stream/feedback-associativity.svg
:align: center
* Associativity of tensor holds up to interchanger:
>>> Equation(*map(lambda x: x.now, ((f @ g) @ h, f @ (g @ h)))).draw(
... doctest="docs/_static/stream/feedback-tensor-associativity.svg")
.. image:: /_static/stream/feedback-tensor-associativity.svg
:align: center
>>> eq_up_to_interchanger = lambda *xs: all_eq(
... x.now.to_hypergraph() for x in xs)
>>> assert eq_up_to_interchanger((f @ g) @ h, f @ (g @ h))
* Interchanger holds up to permutation of the memories:
>>> x_, y_, z_, m_, n_ = [
... Ty.sequence(symmetric.Ty(name + "'")) for name in "xyzmn"]
>>> f_ = Stream.sequence("f'", x_, y_, m_)
>>> g_ = Stream.sequence("g'", y_, z_, n_)
>>> LHS, RHS = f @ f_ >> g @ g_, (f >> g) @ (f_ >> g_)
>>> Equation(LHS.now, RHS.now, symbol="$\\\\sim$").draw(figsize=(8, 6),
... doctest="docs/_static/stream/feedback-interchanger.svg")
.. image:: /_static/stream/feedback-interchanger.svg
:align: center
>>> pi, id_dom = (0, 1, 2, 4, 3, 5), symmetric.Id(LHS.now.dom)
>>> assert symmetric.Equation(
... LHS.now, id_dom.permute(*pi) >> RHS.now.permute(*pi))
See :mod:`discopy.feedback` for the other axioms for feedback categories.
"""
from __future__ import annotations
from typing import Optional
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from discopy import symmetric
from discopy.abc import MonoidalCategory, NamedGeneric
from discopy.python import finset
from discopy.utils import (
AxiomError,
assert_isinstance, unbiased, inductive, classproperty, factory_name)
[docs]
@dataclass
class Ty(NamedGeneric['base']):
"""
A stream of types from some underlying class `base`.
Parameters:
now (base) : The value of the stream at time step zero.
_later (Optional[Callable[[], Ty[base]]]) :
A thunk for the tail of the stream, constant by default.
"""
base = symmetric.Ty # The underlying class of types.
now: base = None
_later: Callable[[], Ty[base]] = None
ob = classproperty(lambda cls: cls)
def __init__(
self, now: base = None, _later: Callable[[], Ty[base]] = None):
self.now = self.base() if now is None else self.base.cast(now)
self._later = _later
def __repr__(self):
_later = "" if self.is_constant else f", _later={repr(self._later)}"
return factory_name(type(self)) + f"({repr(self.now)}{_later})"
@property
def later(self) -> Ty:
""" The tail of a stream, or `self` if :meth:`is_constant`. """
return self if self.is_constant else self._later()
@property
def head(self) -> Ty:
""" The :meth:`singleton` over the first time step. """
return self.singleton(self.now)
tail = later
@property
def is_constant(self) -> bool:
""" Whether a stream of type is constant. """
return self._later is None
[docs]
@classmethod
def singleton(cls, x: base) -> Ty:
"""
Constructs the stream with `x` now and the empty stream later.
>>> XY = Ty.singleton(symmetric.Ty('x', 'y'))
>>> for x in [XY.now, XY.later.now, XY.later.later.now]: print(x)
x @ y
Ty()
Ty()
"""
return cls(now=x, _later=lambda: cls())
[docs]
@inductive
def delay(self) -> Ty:
"""
Delays a stream of types by pre-pending with the unit.
>>> XY = Ty(symmetric.Ty('x', 'y')).delay()
>>> for x in [XY.now, XY.later.now, XY.later.later.now]: print(x)
Ty()
x @ y
x @ y
"""
return type(self)(self.base(), lambda: self)
d = property(lambda self: self.delay())
[docs]
@classmethod
def sequence(cls, x: base, n_steps: int = 0) -> Ty:
"""
Constructs the stream `x0`, `x1`, etc.
>>> XY = Ty.sequence(symmetric.Ty('x', 'y'))
>>> for x in [XY.now, XY.later.now, XY.later.later.now]: print(x)
x0 @ y0
x1 @ y1
x2 @ y2
"""
now = cls.base().tensor(*(cls.base(f"{obj}{n_steps}") for obj in x))
return cls(now, _later=lambda: cls.sequence(x, n_steps + 1))
[docs]
@inductive
def unroll(self) -> Ty:
"""
Unroll a stream `x0, x1, x2, x3, ...` to `x0 @ x1, x2, x3, ...`.
>>> U = Ty.sequence('x').unroll()
>>> for x in [U.now, U.later.now, U.later.later.now]: print(x)
x0 @ x1
x2
x3
"""
return type(self)(self.now @ self.later.now, lambda: self.later.later)
[docs]
@unbiased
def tensor(self, other: Ty) -> Ty:
"""
The tensor of streams of types is computed pointwise.
>>> X, Y = map(Ty.sequence, "xy")
>>> XY = X @ Y
>>> for x in [XY.now, XY.later.now, XY.later.later.now]: print(x)
x0 @ y0
x1 @ y1
x2 @ y2
"""
if not isinstance(other, Ty):
return NotImplemented
_later = None if self.is_constant and other.is_constant else (
lambda: self.later.tensor(other.later))
return type(self)(self.now @ other.now, _later)
__matmul__ = symmetric.Ty.__matmul__
__pow__ = symmetric.Ty.__pow__
[docs]
@dataclass
class Stream(MonoidalCategory, NamedGeneric['category']):
"""
Monoidal streams over an underlying `category`.
Parameters:
now (category) : The value of the stream at time step zero.
dom (Optional[Ty[category.ob]]) :
The domain of the stream, constant `now.dom` if `_later is None`.
cod (Optional[Ty[category.ob]]) :
The codomain of the stream, constant `now.dom` if `_later is None`.
mem (Optional[Ty[category.ob]]) :
The memory of the stream, the constant empty type by default.
_later (Optional[Callable[[], Stream[category]]]) :
A thunk for the tail of the stream, constant by default.
Example
-------
>>> from discopy import python
>>> T, S = Ty[python.Ty], Stream[python.Function]
>>> x, y, m = int, bool, str
>>> now = python.Function(lambda n: (bool(n % 2), str(n)), x, (y, m))
>>> dom, cod, mem = T(x), T(y), T(m).delay()
>>> later = S(lambda n, s: (bool(n % 2), f"{s} {n}"), dom, cod, mem.later)
>>> f = S(now, dom, cod, mem, lambda: later)
>>> f.unroll(2).now(1, 2, 3)
(True, False, True, '1 2 3')
Note
----
The parameters should satisfy the following conditions:
>>> assert now.dom == dom.now @ mem.now
>>> assert now.cod == cod.now @ mem.later.now
>>> assert dom.later.now == later.dom.now
>>> assert cod.later.now == later.cod.now
>>> assert mem.later.now == later.mem.now
"""
category = symmetric.Diagram
ob = classproperty(lambda cls: Ty[cls.category.ob])
now: category
dom: ob = None
cod: ob = None
mem: ob = None
_later: Callable[[], Stream[category]] = None
later, is_constant = Ty.later, Ty.is_constant
head, tail = Ty.head, Ty.tail
def __init__(
self, now: category,
dom: ob = None,
cod: ob = None,
mem: ob = None,
_later: Callable[[], Stream[category]] = None):
if dom is None or cod is None:
if mem is not None or _later is not None:
raise ValueError(
"Cannot have mem or _later if dom or cod is None.")
dom = Ty[self.category.ob](now.dom) if dom is None else dom
cod = Ty[self.category.ob](now.cod) if cod is None else cod
mem = Ty[self.category.ob]() if mem is None else mem
for typ in (dom, cod, mem):
assert_isinstance(typ, Ty)
if not isinstance(now, self.category):
now = self.category(
now, dom.now @ mem.now, cod.now @ mem.later.now)
if now.dom != dom.now @ mem.now:
raise AxiomError(f"{dom.now @ mem.now} != {now.dom}")
if now.cod != cod.now @ mem.later.now:
raise AxiomError(f"{cod.now @ mem.later.now} != {now.cod}")
if _later is None:
if not all(x.is_constant for x in [dom, cod, mem]):
raise ValueError(
"Constant streams should have constant dom, cod and mem")
self.dom, self.cod, self.mem = dom, cod, mem
self.now, self._later = now, _later
[docs]
def check_later(self):
""" Check that later has consistent domain, codomain and memory. """
later = self.later
assert_isinstance(later, type(self))
assert self.dom.later.now == later.dom.now
assert self.cod.later.now == later.cod.now
assert self.mem.later.now == later.mem.now
mem_dom = property(lambda self: self.mem.now)
mem_cod = property(lambda self: self.mem.later.now)
[docs]
@classmethod
def singleton(cls, arg: category) -> Stream:
"""
Construct the stream with a given arrow now and the empty stream later.
"""
dom, cod = map(
Ty[cls.category.ob].singleton, (arg.dom, arg.cod))
return cls(arg, dom, cod, _later=lambda: cls.id())
[docs]
@classmethod
def sequence(
cls, name: str, dom: Ty, cod: Ty, mem: Ty = None, n_steps: int = 0,
box_factory=symmetric.Box) -> Stream:
"""
Produce a stream of boxes indexed by a time step.
Example
-------
>>> x, y, m = [Ty.sequence(symmetric.Ty(n)) for n in "xym"]
>>> f = Stream.sequence("f", x @ m.delay(), y @ m)
>>> for fi in [f.now, f.later.now, f.later.later.now]:
... print(fi, ":", fi.dom, "->", fi.cod)
f0 : x0 -> y0 @ m0
f1 : x1 @ m0 -> y1 @ m1
f2 : x2 @ m1 -> y2 @ m2
"""
mem = Ty[cls.category.ob]() if mem is None else mem
now = box_factory(
f"{name}{n_steps}", dom.now @ mem.now, cod.now @ mem.later.now)
return cls(now, dom, cod, mem, _later=lambda: cls.sequence(
name, dom.later, cod.later, mem.later, n_steps + 1, box_factory))
[docs]
@inductive
def delay(self) -> Stream:
""" Delay a stream by one time step, shortened to `self.d`. """
dom, cod, mem = [x.delay() for x in (self.dom, self.cod, self.mem)]
now, _later = self.category.id(self.mem.now), lambda: self
return type(self)(now, dom, cod, mem, _later)
d = property(lambda self: self.delay())
[docs]
@inductive
def unroll(self) -> Stream:
"""
Unrolling a stream for `n_steps`.
Example
-------
>>> from discopy.monoidal import Equation
>>> f = Stream.sequence("f", *map(Ty.sequence, "xym"))
>>> Equation(f.now, f.unroll().now, f.unroll(2).now, symbol=',').draw(
... figsize=(8, 4), doctest="docs/_static/stream/unroll.svg")
.. image:: /_static/stream/unroll.svg
:align: center
"""
later = self.later
dom, cod = self.dom.unroll(), self.cod.unroll()
mem = Ty[self.category.ob](
self.mem.now, lambda: self.mem.later.later)
now = self.dom.now @ self.category.swap(later.dom.now, self.mem_dom)
now >>= self.now @ later.dom.now
now >>= self.cod.now @ self.category.swap(
self.mem_cod, later.dom.now) >> self.cod.now @ later.now
return type(self)(now, dom, cod, mem, _later=lambda: later.later)
[docs]
@classmethod
def id(cls, x: Optional[Ty] = None) -> Stream:
"""
Construct a stream of identity arrows.
Example
-------
>>> id_x = Stream.id(Ty.sequence('x'))
>>> print(id_x.now, id_x.later.now, id_x.later.later.now)
Id(x0) Id(x1) Id(x2)
"""
x = Ty[cls.category.ob]() if x is None else x
assert_isinstance(x, Ty)
now, dom, cod = cls.category.id(x.now), x, x
_later = None if x.is_constant else lambda: cls.id(x.later)
return cls(now, dom, cod, _later=_later)
[docs]
@unbiased
def then(self, other: Stream) -> Stream:
"""
Composition of streams is given by swapping the memories as follows:
Example
-------
>>> x, y, z, m, n = map(Ty.sequence, "xyzmn")
>>> f = Stream.sequence("f", x, y, m)
>>> g = Stream.sequence("g", y, z, n)
>>> (f >> g).now.draw(doctest="docs/_static/stream/stream-then.svg")
.. image:: /_static/stream/stream-then.svg
:align: center
"""
swap = self.category.swap
now = self.now @ other.mem_dom
now >>= self.cod.now @ swap(self.mem_cod, other.mem_dom)
now >>= other.now @ self.mem_cod
now >>= other.cod.now @ swap(other.mem_cod, self.mem_cod)
dom, cod, mem = self.dom, other.cod, self.mem @ other.mem
_later = None if self.is_constant and other.is_constant else (
lambda: self.later >> other.later)
return type(self)(now, dom, cod, mem, _later)
[docs]
@unbiased
def tensor(self, other: Stream) -> Stream:
"""
Tensor of streams is given by swapping the memories as follows:
Example
-------
>>> x, y, z, w, m, n = map(Ty.sequence, "xyzwmn")
>>> f = Stream.sequence("f", x, y, m)
>>> g = Stream.sequence("g", z, w, n)
>>> (f @ g).now.draw(doctest="docs/_static/stream/stream-tensor.svg")
.. image:: /_static/stream/stream-tensor.svg
:align: center
"""
assert_isinstance(other, Stream)
swap = self.category.swap
now = self.dom.now @ swap(other.dom.now, self.mem_dom) @ other.mem_dom
now >>= self.now @ other.now
now >>= self.cod.now @ swap(
self.mem_cod, other.cod.now) @ other.mem_cod
dom = self.dom @ other.dom
cod = self.cod @ other.cod
mem = self.mem @ other.mem
_later = None if self.is_constant and other.is_constant else (
lambda: self.later.tensor(other.later))
return type(self)(now, dom, cod, mem, _later)
[docs]
@classmethod
def swap(cls, left: Ty, right: Ty) -> Stream:
""" Construct a stream of swaps. """
now = cls.category.swap(left.now, right.now)
dom, cod = left @ right, right @ left
_later = None if left.is_constant and right.is_constant else (
lambda: cls.swap(left.later, right.later))
return cls(now, dom, cod, _later=_later)
[docs]
@classmethod
def permutation(cls, xs: Sequence[int], doms: Sequence[Ty]) -> Stream:
""" Construct a stream of permutations. """
doms = list(doms)
xs = finset.Permutation(xs, len(doms))
dom = cls.ob().tensor(*doms)
if xs.is_identity:
return cls.id(dom)
now = cls.category.ar.permutation(xs, [dom.now for dom in doms])
_later = None if all(dom.is_constant for dom in doms) else (
lambda: cls.permutation(xs, [dom.later for dom in doms]))
cod = cls.ob().tensor(*(doms[i] for i in xs))
return cls(now, dom, cod, _later=_later)
[docs]
@classmethod
def copy(cls, dom: Ty, n: int = 2) -> Stream:
""" Construct a stream of diagonal morphisms. """
now, cod = cls.category.copy(dom.now, n), dom ** n
_later = None if dom.is_constant else lambda: cls.copy(dom.later, n)
return cls(now, dom, cod, _later=_later)
[docs]
def feedback(
self, dom: Ty = None, cod: Ty = None, mem: Ty = None, _first_call=True
) -> Stream:
"""
The delayed feedback of a monoidal stream.
Parameters:
dom (Ty) : The domain of the result.
cod (Ty) : The domain of the result.
mem (Ty) : The memory over which we are taking a feedback.
Example
-------
>>> x, y, m = [Ty.sequence(symmetric.Ty(n)) for n in "xym"]
>>> f = Stream.sequence("f", x @ m.delay(), y @ m)
>>> fb = f.feedback(x, y, m)
>>> from discopy.monoidal import Equation
>>> Equation(f.unroll(2).now, fb.unroll(2).now, symbol="$\\\\mapsto$"
... ).draw(doctest="docs/_static/stream/feedback-unrolling.svg")
.. image:: /_static/stream/feedback-unrolling.svg
:align: center
"""
if mem is None or dom is None or cod is None:
if not self.is_constant or dom is not None or cod is not None:
raise NotImplementedError
assert self.dom.now == dom.now if _first_call else (
self.dom.now == dom.now @ mem.now)
assert self.cod.now == cod.now @ mem.now if _first_call else (
self.cod.now == cod.now @ mem.later.now)
def _later():
return self.later.feedback(dom.later, cod.later, mem.later, False)
mem = mem.delay() if _first_call else mem
return type(self)(self.now, dom, cod, mem @ self.mem, _later)
followed_by = id