Skip to content

Stream

Module for managing a stream of builder nodes.

This module provides classes to represent builder nodes and builder streams. A builder node is a single element in the stream, representing a step in a construction process. A builder stream is a sequence of builder nodes, allowing traversal and manipulation of the construction steps.

BuilderNode dataclass 

BuilderNode(
    node: Builder, next: Optional[BuilderNode] = None
)

A node in the builder stream.

__repr__ 

__repr__() -> str

Representation of the BuilderNode.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
22
23
24
def __repr__(self) -> str:
    """Representation of the BuilderNode."""
    return repr(self.node)

BuilderStream dataclass 

BuilderStream(
    head: BuilderNode, curr: Optional[BuilderNode] = None
)

Represents a stream of builder nodes.

__iter__ 

__iter__()

Iterator method to iterate over the builder stream.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
87
88
89
def __iter__(self):
    """Iterator method to iterate over the builder stream."""
    return self

__next__ 

__next__()

Next method to get the next item in the builder stream.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
91
92
93
94
95
96
def __next__(self):
    """Next method to get the next item in the builder stream."""
    node = self.read()
    if node is None:
        raise StopIteration
    return node

build_nodes staticmethod 

build_nodes(node: Builder) -> BuilderNode

Build BuilderNode instances from the provided Builder.

Parameters:

Name Type Description Default
node Builder

The root Builder instance.

 required

Returns:

Name Type Description
BuilderNode BuilderNode

The head of the linked list of BuilderNodes.
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@staticmethod
def build_nodes(node: Builder) -> "BuilderNode":
    """
    Build BuilderNode instances from the provided Builder.

    Args:
        node (Builder): The root Builder instance.

    Returns:
        BuilderNode: The head of the linked list of BuilderNodes.
    """
    curr = node
    node = None
    while curr is not None:
        next = curr
        curr = curr.__parent__ if hasattr(curr, "__parent__") else None
        node = BuilderNode(next, node)

    return node

copy 

copy() -> BuilderStream

Create a copy of the builder stream.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
34
35
36
def copy(self) -> "BuilderStream":
    """Create a copy of the builder stream."""
    return BuilderStream(head=self.head, curr=self.curr)

create staticmethod 

create(builder: Builder) -> BuilderStream

Create a BuilderStream instance from a Builder.

Parameters:

Name Type Description Default
builder Builder

The root Builder instance.

 required

Returns:

Name Type Description
BuilderStream BuilderStream

The created BuilderStream instance.
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
@staticmethod
def create(builder: Builder) -> "BuilderStream":
    """
    Create a BuilderStream instance from a Builder.

    Args:
        builder (Builder): The root Builder instance.

    Returns:
        BuilderStream: The created BuilderStream instance.
    """
    head = BuilderStream.build_nodes(builder)
    return BuilderStream(head=head, curr=head)

eat 

eat(
    types: List[Type[Builder]],
    skips: Optional[List[Type[Builder]]] = None,
) -> BuilderNode

Move the stream pointer until a node of specified types is found.

Parameters:

Name Type Description Default
types List[Type[Builder]]

List of types to move the stream pointer to.

 required
skips List[Type[Builder]] | None

List of types to end the stream scan.

 None

Returns:

Name Type Description
BuilderNode BuilderNode

The beginning of the stream which matches a type in types.
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def eat(
    self, types: List[Type[Builder]], skips: Optional[List[Type[Builder]]] = None
) -> BuilderNode:
    """
    Move the stream pointer until a node of specified types is found.

    Args:
        types (List[Type[Builder]]): List of types to move the stream pointer to.
        skips (List[Type[Builder]] | None, optional): List of types to end the stream scan.

    Returns:
        BuilderNode: The beginning of the stream which matches a type in `types`.
    """
    head = self.read_next(types)
    curr = head
    while curr is not None:
        if type(curr.node) not in types:
            if skips and type(curr.node) not in skips:
                break
        curr = curr.next
    self.curr = curr
    return head

read 

read() -> Optional[BuilderNode]

Read the next builder node from the stream.

Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
38
39
40
41
42
43
44
45
def read(self) -> Optional[BuilderNode]:
    """Read the next builder node from the stream."""
    if self.curr is None:
        return None

    node = self.curr
    self.curr = self.curr.next
    return node

read_next 

read_next(
    builder_types: List[type[Builder]],
) -> Optional[BuilderNode]

Read the next builder node of specified types from the stream.

Parameters:

Name Type Description Default
builder_types List[type[Builder]]

List of builder types to read from the stream.

 required

Returns:

Type Description
Optional[BuilderNode]

Optional[BuilderNode]: The next builder node matching one of the specified types, or None if not found.
Source code in .venv/lib/python3.12/site-packages/bloqade/analog/builder/parse/stream.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def read_next(self, builder_types: List[type[Builder]]) -> Optional[BuilderNode]:
    """
    Read the next builder node of specified types from the stream.

    Args:
        builder_types (List[type[Builder]]): List of builder types to read from the stream.

    Returns:
        Optional[BuilderNode]: The next builder node matching one of the specified types, or None if not found.
    """
    node = self.read()
    while node is not None:
        if type(node.node) in builder_types:
            return node
        node = self.read()
    return None