Stream#
- class discopy.stream.Stream(now, dom=None, cod=None, mem=None, _later=None)[source]#
Bases:
Composable,Whiskerable,NamedGeneric['category']Monoidal streams over an underlying category.
- Parameters:
now (category.ar) – 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.Category] >>> 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
- property later: Ty#
The tail of a stream, or self if
is_constant().
- property is_constant: bool#
Whether a stream of type is constant.
- property head: Ty#
The
singleton()over the first time step.
- property tail: Ty#
The tail of a stream, or self if
is_constant().
- classmethod singleton(arg)[source]#
Construct the stream with a given arrow now and the empty stream later.
- Parameters:
arg (category.ar) –
- Return type:
- classmethod sequence(name, dom, cod, mem=None, n_steps=0, box_factory=<class 'discopy.symmetric.Box'>)[source]#
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
- unroll()[source]#
Unrolling a stream for n_steps.
Example
>>> from discopy.drawing 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), path="docs/_static/stream/unroll.png")
- Return type:
- classmethod id(x=None)[source]#
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)
- then(other)[source]#
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(path="docs/_static/stream/stream-then.png")
- tensor(other)[source]#
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(path="docs/_static/stream/stream-tensor.png")
- feedback(dom=None, cod=None, mem=None, _first_call=True)[source]#
The delayed feedback of a monoidal stream.
- Parameters:
- Return type:
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.drawing import Equation >>> Equation(f.unroll(2).now, fb.unroll(2).now, symbol="$\\mapsto$" ... ).draw(path="docs/_static/stream/feedback-unrolling.png")
