Skip to content

Stream

stream

T = TypeVar('T') module-attribute

TCov = TypeVar('TCov', covariant=True) module-attribute

TCon = TypeVar('TCon', contravariant=True) module-attribute

ReadableStream

Bases: ABC, Generic[TCov]

Abstract base class for readable stream interpreter model objects.

Source code in xdsl/interpreters/utils/stream.py 
12
13
14
15
16
17
18
19
class ReadableStream(abc.ABC, Generic[TCov]):
    """
    Abstract base class for readable stream interpreter model objects.
    """

    @abc.abstractmethod
    def read(self) -> TCov:
        raise NotImplementedError()

read() -> TCov abstractmethod

Source code in xdsl/interpreters/utils/stream.py 
17
18
19
@abc.abstractmethod
def read(self) -> TCov:
    raise NotImplementedError()

WritableStream

Bases: ABC, Generic[TCon]

Abstract base class for readable stream interpreter model objects.

Source code in xdsl/interpreters/utils/stream.py 
22
23
24
25
26
27
28
29
class WritableStream(abc.ABC, Generic[TCon]):
    """
    Abstract base class for readable stream interpreter model objects.
    """

    @abc.abstractmethod
    def write(self, value: TCon) -> None:
        raise NotImplementedError()

write(value: TCon) -> None abstractmethod

Source code in xdsl/interpreters/utils/stream.py 
27
28
29
@abc.abstractmethod
def write(self, value: TCon) -> None:
    raise NotImplementedError()

Nats dataclass

Bases: ReadableStream[int]

A stream designed for testing, outputs the next natural number each time it's read.

Source code in xdsl/interpreters/utils/stream.py 
32
33
34
35
36
37
38
39
40
41
42
@dataclass
class Nats(ReadableStream[int]):
    """
    A stream designed for testing, outputs the next natural number each time it's read.
    """

    index = 0

    def read(self) -> int:
        self.index += 1
        return self.index

index = 0 class-attribute instance-attribute

__init__() -> None

read() -> int

Source code in xdsl/interpreters/utils/stream.py 
40
41
42
def read(self) -> int:
    self.index += 1
    return self.index

Acc dataclass

Bases: WritableStream[int]

A stream designed for testing, appends the next natural number written.

Source code in xdsl/interpreters/utils/stream.py 
45
46
47
48
49
50
51
52
53
54
@dataclass
class Acc(WritableStream[int]):
    """
    A stream designed for testing, appends the next natural number written.
    """

    values: list[int] = field(default_factory=list[int])

    def write(self, value: int) -> None:
        return self.values.append(value)

values: list[int] = field(default_factory=(list[int])) class-attribute instance-attribute

__init__(values: list[int] = list[int]()) -> None

write(value: int) -> None

Source code in xdsl/interpreters/utils/stream.py 
53
54
def write(self, value: int) -> None:
    return self.values.append(value)