Module `refinery.lib.frame`

Some refinery units produce more than one output when applied to an input. For example, chop will chop the input data into evenly sized blocks and emit each of them as a single output. By default, if no framing syntax is used, multiple outputs are separated by line breaks, which is often desirable when text data is extracted. However, for processing binary data, this is equally often more than useless. To process the list of results generated by any refinery unit, end the command for this unit with the special argument [. This argument has to be the last argument to be recognized as a framing initialization. If this syntax is used, the list of results is emitted in an internal format which allows arbitrarily nested lists of binary chunks to be processed.

Simple Frame Example

$ emit OOOOOOOO | chop 2 [| ccp F | cca . ]
FOO.FOO.FOO.FOO.

Here, the string OOOOOOOO is first chopped into blocks of 2, yielding the frame [OO, OO, OO, OO] which is then forwarded to the next command. If a Unit receives input in framed format, each chunk of the frame is processed individually and emitted as one output chunk. In this case, ccp simply prepends F to every input, producing the frame [FOO, FOO, FOO, FOO]. Finally, cca appends a period to each chunk. When a unit is given the closing bracket as the last argument, this concludes processing of one frame which results in concatenation of all binary chunks in the frame.

Frame Layers

Frames can be nested arbitrarily, and sep can be used to insert a separator (the default is line break) between all chunks in the frame:

$ emit OOOOOOOO | chop 4 [| chop 2 [| ccp F | cca . ]| sep ]
FOO.FOO.
FOO.FOO.

Here, we first produce the two-layered frame tree [[OO,OO], [OO,OO]] by using two chop invocations. We refer to this data as a tree because, well, it is one:

LAYER 1:      [[..],[..]]
                /     \
LAYER 2:    [OO,OO] [OO,OO]

The bottom layer is processed as before, yielding [FOO.FOO., FOO.FOO.]. Next, the unit sep inserts a line break character between the two chunks in this frame.

Adding Line Breaks Easily

Since separating data with line breaks is a common requirement, it is also possible to use one more closing bracket than necessary at the end of a frame to separate all chunks by line breaks:

$ emit OOOOOOOO | chop 4 [| chop 2 [| ccp F | cca . ]]]
FOO.FOO.
FOO.FOO.

Squeezing

Inside a frame, application of a Unit with multiple outputs will substitute the input by the corresponding list of outputs. For example,

$ emit OOOOOOOO | chop 4 [| chop 2 | ccp F ]]

has the exact same output as the following command:

$ emit 00000000 | chop 2 [| ccp F ]]

In the first case, we create the frame [OOOO, OOOO] and then apply chop 2 to each chunk, which results in the frame [OO, OO, OO, OO]. Now, consider the example

$ emit OOCLOOCL | chop 4 [| snip 2::-1 3: ]]
COO
L
COO
L

With what we have learned so far, if we wanted it to spell COOL twice instead,we would have to use the following and slightly awkward syntax:

$ emit OOCLOOCL | chop 4 [| snip 2::-1 3 [| nop ]| sep ]
COOL
COOL

This is because the snip command, by default, will simply insert the list [COO, L] into the complete frame, creating the output sequence [COO, L, COO, L] and all of these chunks will be separated by line breaks. For this reason, the squeeze syntax exists. If the brackets at the end of a refinery command are prefixed by the sequence [], i.e. an opening bracket followed directly by a closing one, then all outputs of the unit are fused into a single output chunk by concatenating them. In our example:

$ emit OOCLOOCL | chop 4 [| snip 2::-1 3 []]]
COOL
COOL

Scoping

It is possible to alter the visibility of Chunk, primarily by using scope. The unit accepts a slice argument which defines the indices of the current frame that remain visible. All subsequent units will only process visible chunks and simply forward the ones that are not visible. Chunks remain invisible when a new frame layer opens:

$ emit BINARY REFINERY [| scope 0 | clower | sep - ]
binary-REFINERY

Here, the scope was limited to the first chunk BINARY which was transformed to lower case, but the second chunk REFINERY was left untouched. A somewhat more complex example:

$ emit aaaaaaaa namtaB [| scope 0 | rex . [| ccp N ]| scope 1 | rev | sep - ]
NaNaNaNaNaNaNaNa-Batman

Note that sep makes all chunks in the frame visible by default, because it is intended to sit at the end of a frame. Otherwise, NaNaNaNaNaNaNaNa and Batman in the above example would not be separated by a dash.

Expand source code Browse git

R"""
Some refinery units produce more than one output when applied to an input. For example,
`refinery.chop` will chop the input data into evenly sized blocks and emit each of them
as a single output. By default, if no framing syntax is used, multiple outputs are
separated by line breaks, which is often desirable when text data is extracted. However,
for processing binary data, this is equally often more than useless. To process the list
of results generated by any refinery unit, end the command for this unit with the
special argument `[`. This argument has to be the last argument to be recognized as a
framing initialization. If this syntax is used, the list of results is emitted in an
internal format which allows arbitrarily nested lists of binary chunks to be processed.

### Simple Frame Example

    $ emit OOOOOOOO | chop 2 [| ccp F | cca . ]
    FOO.FOO.FOO.FOO.

Here, the string `OOOOOOOO` is first chopped into blocks of 2, yielding the **frame**
`[OO, OO, OO, OO]` which is then forwarded to the next command. If a `refinery.units.Unit`
receives input in framed format, each chunk of the frame is processed individually and
emitted as one output chunk. In this case, `refinery.ccp` simply prepends `F` to every
input, producing the frame `[FOO, FOO, FOO, FOO]`. Finally, `refinery.cca` appends a period
to each chunk. When a unit is given the closing bracket as the last argument, this
concludes processing of one frame which results in concatenation of all binary chunks in
the frame.

### Frame Layers

Frames can be nested arbitrarily, and `refinery.sep` can be used to insert a separator
(the default is line break) between all chunks in the frame:

    $ emit OOOOOOOO | chop 4 [| chop 2 [| ccp F | cca . ]| sep ]
    FOO.FOO.
    FOO.FOO.

Here, we first produce the two-layered **frame tree** `[[OO,OO], [OO,OO]]` by using two
`refinery.chop` invocations. We refer to this data as a tree because, well, it is one:

    LAYER 1:      [[..],[..]]
                    /     \
    LAYER 2:    [OO,OO] [OO,OO]

The bottom layer is processed as before, yielding `[FOO.FOO., FOO.FOO.]`. Next, the unit
`refinery.sep` inserts a line break character between the two chunks in this frame.

### Adding Line Breaks Easily

Since separating data with line breaks is a common requirement, it is also possible to use
one more closing bracket than necessary at the end of a frame to separate all chunks by line
breaks:

    $ emit OOOOOOOO | chop 4 [| chop 2 [| ccp F | cca . ]]]
    FOO.FOO.
    FOO.FOO.

### Squeezing

Inside a frame, application of a `refinery.units.Unit` with multiple outputs will substitute the
input by the corresponding list of outputs. For example,

    $ emit OOOOOOOO | chop 4 [| chop 2 | ccp F ]]

has the exact same output as the following command:

    $ emit 00000000 | chop 2 [| ccp F ]]

In the first case, we create the frame `[OOOO, OOOO]` and then apply `chop 2` to each chunk,
which results in the frame `[OO, OO, OO, OO]`. Now, consider the example

    $ emit OOCLOOCL | chop 4 [| snip 2::-1 3: ]]
    COO
    L
    COO
    L

With what we have learned so far, if we wanted it to spell `COOL` twice instead,we would have
to use the following and slightly awkward syntax:

    $ emit OOCLOOCL | chop 4 [| snip 2::-1 3 [| nop ]| sep ]
    COOL
    COOL

This is because the `snip` command, by default, will simply insert the list `[COO, L]` into
the complete frame, creating the output sequence `[COO, L, COO, L]` and all of these chunks
will be separated by line breaks. For this reason, the squeeze syntax exists. If the brackets
at the end of a refinery command are prefixed by the sequence `[]`, i.e. an opening bracket
followed directly by a closing one, then all outputs of the unit are fused into a single
output chunk by concatenating them. In our example:

    $ emit OOCLOOCL | chop 4 [| snip 2::-1 3 []]]
    COOL
    COOL


### Scoping

It is possible to alter the **visibility** of `refinery.lib.frame.Chunk`, primarily by
using `refinery.scope`. The unit accepts a slice argument which defines the indices of
the current frame that remain visible. All subsequent units will only process visible
chunks and simply forward the ones that are not visible. `refinery.lib.frame.Chunk`s
remain invisible when a new frame layer opens:

    $ emit BINARY REFINERY [| scope 0 | clower | sep - ]
    binary-REFINERY

Here, the scope was limited to the first chunk `BINARY` which was transformed to lower
case, but the second chunk `REFINERY` was left untouched. A somewhat more complex example:

    $ emit aaaaaaaa namtaB [| scope 0 | rex . [| ccp N ]| scope 1 | rev | sep - ]
    NaNaNaNaNaNaNaNa-Batman

Note that `refinery.sep` makes all chunks in the frame visible by default, because it is
intended to sit at the end of a frame. Otherwise, `NaNaNaNaNaNaNaNa` and `Batman` in the
above example would not be separated by a dash.
"""
from __future__ import annotations

import io
import itertools
import os
import stat
import sys
import uuid
import zlib

from typing import (
    Any,
    BinaryIO,
    Callable,
    Generator,
    Iterable,
    Iterator,
    Mapping,
    SupportsIndex,
    overload,
)

from refinery.lib.meta import LazyMetaOracle
from refinery.lib.tools import HandOver
from refinery.lib.types import buf

try:
    import msgpack
except ModuleNotFoundError:
    msgpack = None

__all__ = [
    'Chunk',
    'Framed',
    'FrameUnpacker',
    'MAGIC',
    'MSIZE',
    'generate_frame_header'
]


MAGIC = bytes.fromhex('FEED1985C0CAC01AC0DE')
"""
This is the magic signature that is used by refinery to prefix serialized frame data. If a unit
reads data from STDIN that is prefixed with these bytes, it assumes that serialized frame data
follows. Otherwise, the input is treated as a single unframed chunk.
"""
MSIZE = len(MAGIC) + 1
"""
This is the length of the data returned by `refinery.lib.frame.generate_frame_header`.
"""


def generate_frame_header(scope: int):
    """
    This function generates a frame header for a frame tree of depth equal to `scope`. The depth
    is encoded as a single byte following `refinery.lib.frame.MAGIC`. This implies a depth limit
    of 255 for frame trees in refinery, and I dearly hope that noone is insane enough to build a
    refinery pipeline that would be affected.
    """
    if scope > 0xFF:
        raise ValueError('Maximum frame depth exceeded.')
    return B'%s%c' % (MAGIC, scope)


class Chunk(bytearray):
    """
    Represents the individual chunks in a frame. The `refinery.units.Unit.filter` method receives
    an iterable of `refinery.lib.frame.Chunk`s.
    """
    temp: Any
    """
    Units can use this field to transport temporary data from `refinery.units.Unit.filter` to
    `refinery.units.Unit.process` or `refinery.units.Unit.reverse` for the same chunk. The frame
    handler clears it once the unit has processed the chunk, so that it may refer back to the chunk.
    """
    uuid: uuid.UUID
    """
    Each chunk object carries a unique identifier. The `refinery.units.DelayedArgumentProxy` uses
    this property to check whether `refinery.units.Unit` command-line arguments were previously
    evaluated against this chunk. Otherwise `refinery.lib.argformats.DelayedArgument`s that alter
    the input data could produce unexpected results when the argument proxy is mapped against the
    same chunk twice.
    """

    __slots__ = (
        '_meta',
        '_view',
        '_path',
        '_fill_scope',
        '_fill_batch',
        'temp',
        'uuid',
        '__weakref__',
    )

    def __init__(
        self,
        data: buf | int | None = None,
        path: list[int] | None = None,
        view: list[bool] | None = None,
        meta: Mapping[str, Any] | None = None,
        seed: dict[str, list] | None = None,
        fill_scope: bool | None = None,
        fill_batch: int | None = None,
        ignore_chunk_properties: bool = False,
    ):
        if data is None:
            bytearray.__init__(self)
        else:
            bytearray.__init__(self, data)

        self.uuid = uuid.uuid4()
        self.temp = None

        if path is None:
            path = []
        if view is None:
            view = [False] * len(path)
        elif len(view) != len(path):
            raise ValueError('view must have the same length as path')

        if not ignore_chunk_properties and isinstance(data, Chunk):
            path = path or list(data.path)
            view = view or list(data.view)
            meta = meta or data.meta
            fill_scope = fill_scope or data._fill_scope
            fill_batch = fill_batch or data._fill_batch

        self._view: list[bool] = view
        self._path: list[int] = path
        self._fill_scope: bool | None = fill_scope
        self._fill_batch: bool | None = fill_batch

        self._meta = m = LazyMetaOracle(self, scope=self.scope, seed=seed, owned=True)
        if meta is not None:
            m.update(meta)

    @classmethod
    def Wrap(cls, data):
        if isinstance(data, cls):
            return data
        return cls(data)

    def set_next_scope(self, visible: bool) -> None:
        self._fill_scope = visible

    def set_next_batch(self, batch: int) -> None:
        """
        This function allows units to emit trees of depth one rather than lists. When a unit emits
        a chunk at index `a`, sets the next batch to `b`, and when a double frame opens after this
        unit's invocation, then said chunk will have `a/b` added to its path. By default, `b` would
        always be `0`. For example, the `refinery.rex` unit uses this feature. As a result:

            $ emit #1yellow-#3red-#2orange | rex #(.)([a-z]+) {1} {2} [[| pop x:e ]| rep v:x ]]
            yellow
            red
            red
            red
            orange
            orange

        The double frame after `refinery.rex` looks like this:

            [[1,yellow],[2,red],[3,orange]]

        By default, the frame would simply look like this:

            [[1,yellow,2,red,3,orange]]

        This feature is useful for `refinery.units.Unit`s that produce multiple outputs for each of
        a number of intermediate results - in the case of `refinery.rex`, that intermediate result
        is a regular expression match, and `refinery.rex` allows to produce different outputs for
        each of those.
        """
        self._fill_batch = batch

    @property
    def scope(self) -> int:
        """
        This value is the length of `refinery.lib.frame.Chunk.path` and therefore corresponds to
        the depth of the frame tree. It is called "scope" because it is equally the scope at which
        new metadata variables for this chunk will be created.
        """
        return len(self._path)

    @property
    def view(self) -> list[bool]:
        """
        This tuple of boolean values indicates the visibility of this chunk at each layer of
        the frame tree. The `refinery.scope` unit can be used to change visibility of chunks
        within a frame.
        """
        return self._view

    @property
    def path(self) -> list[int]:
        """
        The vertices in each frame tree layer are sequentially numbered by their order of
        appearance in the stream. The `refinery.lib.frame.Chunk.path` contains the numbers of
        the vertices (in each layer) which define the path from the root of the frame tree
        to the leaf vertex representing this `refinery.lib.frame.Chunk`
        """
        return self._path

    @property
    def meta(self) -> LazyMetaOracle:
        """
        Every chunk can contain a dictionary of arbitrary metadata. Further details about this data
        are available in the module-level documetnation of `refinery.lib.meta`.
        """
        return self._meta

    @property
    def visible(self):
        """
        This property defines whether the chunk is currently visible. It defaults to true if the
        chunk is not part of a frame and is otherwise the same as the last element of the tuple
        `refinery.lib.frame.Chunk.view`. Setting this property will correspondingly alter the last
        entry of `refinery.lib.frame.Chunk.view`.
        Setting this property on an unframed `refinery.lib.frame.Chunk` raises an `AttributeError`.
        """
        return not self._view or self._view[~0]

    @property
    def scopable(self):
        """
        This property defines whether the chunk can be made visible in the current frame.
        """
        return len(self._view) <= 1 or self._view[~1]

    @visible.setter
    def visible(self, value: bool):
        view = self._view
        if not view:
            if not value:
                raise AttributeError('cannot make chunk invisible outside frame')
        else:
            view[~0] = value

    def _relocated(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, serialized metadata, and fill scope of this chunk after it has been
        moved by the given nesting and placed at the given position.
        """
        scope = self.scope + nest
        fs = self._fill_scope
        fb = self._fill_batch
        if nest > 0:
            view = list(self._view)
            path = list(self._path)
            if fs is not None:
                view.extend(itertools.repeat(self.visible, nest - 1))
                view.append(fs)
                fs = None
            else:
                view.extend(itertools.repeat(self.visible, nest))
            if fb is not None and nest > 1:
                path.append(position)
                path.append(fb)
                path.extend(itertools.repeat(0, nest - 2))
            else:
                path.append(position)
                path.extend(itertools.repeat(0, nest - 1))
        elif nest < 0:
            view = self._view[:nest]
            path = self._path[:nest]
        else:
            view = list(self._view)
            path = list(self._path)
        assert len(path) == scope
        assert len(view) == scope
        return path, view, self._meta.serialize(scope), fs

    def serialize(self, nest: int = 0, position: int = 0) -> tuple[bytes | bytearray, ...]:
        """
        Returns the serialized form of this chunk, moved by the given nesting and placed at the
        given position, as a sequence of buffers to be written in order. Their concatenation is a
        msgpack array of the path, view, metadata, fill scope, and data. For a large chunk, the
        first buffer ends with the header of the data, and the chunk itself follows uncopied.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        if (size := len(self)) < _INLINE:
            packed = msgpack.packb((path, view, meta, fs, self))
            assert isinstance(packed, bytes)
            return packed,
        packed = msgpack.packb((path, view, meta, fs))
        assert isinstance(packed, bytes)
        head = bytearray(packed)
        head[0] += 1
        head.extend(_bin_header(size))
        return head, self

    def pack(self, nest: int = 0, position: int = 0) -> bytes:
        """
        Returns the serialized form of this chunk as a single byte string; see
        `refinery.lib.frame.Chunk.serialize`.
        """
        return B''.join(self.serialize(nest, position))

    def _moved(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, metadata, and fill scope that a unit in the same process receives
        for this chunk when it is moved by the given nesting and placed at the given position. The
        metadata is serialized and read back as in a stream.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        meta = msgpack.unpackb(msgpack.packb(meta), use_list=True, strict_map_key=False)
        return path, view, meta, fs

    @staticmethod
    def _received(data: buf, moved: tuple[list[int], list[bool], dict, bool | None]) -> Chunk:
        """
        Returns a chunk with a copy of the given data and the path, view, metadata, and fill scope
        returned by `Chunk._moved`.
        """
        path, view, meta, fs = moved
        return Chunk(data, path, view, None, meta, fs, None, ignore_chunk_properties=True)

    def transfer(self, nest: int = 0, position: int = 0) -> Chunk:
        """
        Returns the chunk that a unit in the same process receives when this chunk is passed on,
        moved by the given nesting and placed at the given position. The data is copied, and the
        metadata is serialized and read back as in a stream: The receiver gets values of its own,
        and a value that cannot be serialized raises the same error as in `Chunk.serialize`.
        """
        return Chunk._received(self, self._moved(nest, position))

    def __repr__(self) -> str:
        layer = '/'.join(str(p) if s else F'!{p}' for p, s in zip(self._path, self._view))
        layer = layer and '/' + layer
        return F'<chunk{layer}:{bytes(self)!r}>'

    def intersect(self, other: Chunk):
        """
        Removes all meta variables from this chunk whose value differs from those of the `other`
        inut chunk.
        """
        other_meta = other._meta
        meta = self._meta
        for key, value in list(meta.items()):
            if other_meta.get(key) != value:
                meta.discard(key)
        return self

    def __str__(self):
        try:
            return self.decode('UTF8')
        except UnicodeDecodeError:
            return self.hex()

    def __hash__(self):
        return hash(zlib.adler32(self))

    @overload
    def __getitem__(self, k: str) -> int | str | float | bytes | None:
        ...

    @overload
    def __getitem__(self, key: SupportsIndex, /) -> int:
        ...

    @overload
    def __getitem__(self, key: slice, /) -> bytearray:
        ...

    def __getitem__(self, k):
        if not isinstance(k, str):
            return bytearray.__getitem__(self, k)
        return self._meta.__getitem__(k)

    def __setitem__(self, bounds, value):
        if isinstance(bounds, str):
            self._meta[bounds] = value
        else:
            bytearray.__setitem__(self, bounds, value)

    def meta_var_int(self, key: str) -> int:
        value = self._meta.__getitem__(key)
        if not isinstance(value, int):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not int.')
        return value

    def meta_var_str(self, key: str) -> str:
        value = self._meta.__getitem__(key)
        if not isinstance(value, str):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not str.')
        return value

    def truncate(self, scope: int = 0):
        """
        Truncate the `refinery.lib.frame.Chunk.path` and `refinery.lib.frame.Chunk.view` lists
        to the given length, setting the `refinery.lib.frame.Chunk.scope` to the given value.
        """
        del self._path[scope:]
        del self._view[scope:]
        return self

    def copy(self, meta=True, data=True) -> Chunk:
        """
        Produce a copy of this chunk. The metadata is copied if the `meta` argument is `True`,
        otherwise the copy has no metadata. The body of the chunk is copied only if the `data`
        argument is `True`.
        """
        data = data and self or None
        copy = Chunk(
            data,
            path=list(self._path),
            view=list(self._view),
            fill_scope=self._fill_scope,
            fill_batch=self._fill_batch,
            ignore_chunk_properties=True,
        )
        if meta:
            copy.meta.update(self.meta)
        if copy.meta.scope != copy.scope:
            raise RuntimeError
        return copy

    def __copy__(self):
        return self.copy()

    def __deepcopy__(self, memo):
        raise NotImplementedError


_INLINE = 0x4000
_READ = 0x10000
_UNSIZED_READ = 0x100000
_BIN_WIDTH = {0xC4: 1, 0xC5: 2, 0xC6: 4}


def _bin_header(size: int) -> bytes:
    if size < 0x100:
        return bytes((0xC4, size))
    if size < 0x10000:
        return B'\xC5' + size.to_bytes(2, 'big')
    if size < 0x100000000:
        return B'\xC6' + size.to_bytes(4, 'big')
    raise ValueError(F'A chunk of size {size} is too large to be serialized.')


def _read_exactly(stream: BinaryIO, size: int) -> bytearray:
    """
    Reads the given number of bytes from the stream, or fewer if it ends.
    """
    data = bytearray()
    while (rest := size - len(data)) > 0 and (more := stream.read(rest)):
        data.extend(more)
    return data


def _readinto(stream: BinaryIO, view: memoryview) -> int:
    """
    Reads from the stream into the view until the view is full or the stream ends, and returns the
    number of bytes read.
    """
    total = 0
    size = len(view)
    readinto = getattr(stream, 'readinto', None)
    while total < size:
        if readinto is not None:
            count = readinto(view[total:])
        elif count := len(data := stream.read(size - total)):
            view[total:total + count] = data
            del data
        if not count:
            break
        total += count
    return total


def _truncated():
    return ValueError('The frame stream ended in the middle of a chunk.')


def _references(item: object) -> int:
    """
    Returns the reference count of the given object. An object that only a local variable of the
    calling function refers to has the same count as a new object that is assigned to a local
    variable of that function.
    """
    return sys.getrefcount(item)


def _remaining_size(stream: BinaryIO) -> int | None:
    """
    Returns the number of bytes that remain in the stream if it can be determined reliably. This is
    only the case for regular files and for streams in memory: Pipes claim to be seekable on some
    platforms, and a stream that decodes a file, like a decompressing one, has to decode all of it
    to seek to its end.
    """
    try:
        if not stream.seekable():
            return None
        if not hasattr(stream, 'getbuffer'):
            if not isinstance(getattr(stream, 'raw', stream), io.FileIO):
                return None
            if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode):
                return None
        position = stream.tell()
        end = stream.seek(0, io.SEEK_END)
        stream.seek(position)
    except Exception:
        return None
    return max(end - position, 0)


def _read_unframed(stream: BinaryIO, head: buf) -> Chunk:
    """
    Reads the rest of an unframed input stream into a chunk that begins with the given head. When
    the stream can tell how much remains, the data is read directly into a chunk of that size;
    otherwise, the chunk grows with every piece that is read. A stream that can read into a buffer
    reads every piece into the same one.
    """
    if (size := _remaining_size(stream)) is not None:
        start = len(head)
        chunk = Chunk(start + size)
        with memoryview(chunk) as view:
            view[:start] = head
            done = start + _readinto(stream, view[start:])
        del chunk[done:]
        piece_size = _READ
    else:
        chunk = Chunk(head)
        piece_size = _UNSIZED_READ
    if (readinto := getattr(stream, 'readinto', None)) is None:
        read = getattr(stream, 'read1', stream.read)
        while piece := read(piece_size):
            chunk.extend(piece)
            del piece
        return chunk
    with memoryview(bytearray(piece_size)) as buffer:
        while count := readinto(buffer):
            chunk.extend(buffer[:count])
    return chunk


def _read_chunks(stream: BinaryIO) -> Iterator[Chunk]:
    """
    Reads the serialized chunks of a framed stream after its header. The first four items of each
    msgpack array are unpacked, then the data is read directly into the chunk.
    """
    assert msgpack
    unpacker = msgpack.Unpacker(use_list=True, max_buffer_size=0xFFFFFFFF, strict_map_key=False)
    read = getattr(stream, 'read1', stream.read)
    fed = 0
    box: list[Chunk] = []

    def fill() -> bool:
        nonlocal fed
        if not (data := read(_READ) or stream.read(_READ)):
            return False
        unpacker.feed(data)
        fed += len(data)
        return True

    def take(count: int) -> bytes:
        while fed - unpacker.tell() < count:
            if not fill():
                raise _truncated()
        return unpacker.read_bytes(count)

    while True:
        items: list | None = None
        while True:
            try:
                if items is None:
                    if unpacker.read_array_header() != 5:
                        raise ValueError('The frame stream contains an item that is not a serialized chunk.')
                    items = []
                while len(items) < 4:
                    items.append(unpacker.unpack())
            except msgpack.OutOfData:
                if not fill():
                    if items is None and fed == unpacker.tell():
                        return
                    raise _truncated()
            else:
                break
        path, view, meta, fs = items
        del items
        code, = take(1)
        if (width := _BIN_WIDTH.get(code)) is None:
            raise ValueError('The frame stream contains a chunk whose data is not a byte string.')
        size = int.from_bytes(take(width), 'big')
        if (buffered := fed - unpacker.tell()) >= size:
            yield Chunk(unpacker.read_bytes(size), path=path, view=view, seed=meta, fill_scope=fs)
            continue
        chunk = Chunk(size, path=path, view=view, seed=meta, fill_scope=fs)
        with memoryview(chunk) as target:
            target[:buffered] = unpacker.read_bytes(buffered)
            if _readinto(stream, target[buffered:]) < size - buffered:
                raise _truncated()
        box.append(chunk)
        del chunk
        yield box.pop()


def _read_stream(stream: BinaryIO | None) -> Iterator[buf]:
    """
    Reads a byte stream and yields its frame header and chunks if it is framed. Otherwise, it
    yields its entire content as one chunk.
    """
    if stream is None or not (head := _read_exactly(stream, len(MAGIC))):
        yield Chunk()
    elif head != MAGIC:
        yield _read_unframed(stream, head)
    elif not (depth := stream.read(1)):
        raise ValueError('The frame stream ended after the frame magic.')
    else:
        yield generate_frame_header(depth[0])
        yield from _read_chunks(stream)


class Placement:
    """
    A chunk of framed output, together with the change of nesting and the position at which it
    enters the frame tree of the next unit. The chunk is moved only once its receiver is known:
    it is serialized for a stream, and transferred to a unit in the same process.
    """
    __slots__ = 'chunk', 'nesting', 'position'

    def __init__(self, chunk: Chunk, nesting: int, position: int):
        self.chunk = chunk
        self.nesting = nesting
        self.position = position

    def serialize(self) -> tuple[bytes | bytearray, ...]:
        return self.chunk.serialize(self.nesting, self.position)

    def transfer(self) -> Chunk:
        return self.chunk.transfer(self.nesting, self.position)


class ChunkFeed:
    """
    The input of a `refinery.lib.frame.FrameUnpacker` when it comes from the same process: the
    output of a unit or a list of chunks. A framed feed begins with a frame header, as generated by
    `refinery.lib.frame.generate_frame_header`, and continues with the chunks that the receiving
    unit reads. An unframed feed consists of the pieces of one input: a chunk followed by more
    chunks or byte strings.
    """
    __slots__ = 'items',

    def __init__(self, items: Iterable[buf]):
        self.items = iter(items)


class FrameUnpacker(Iterable[Chunk]):
    """
    Provides a unified interface to read both framed and raw input data from a stream. After
    loading a framed input stream, the object provides an iterator over the first **frame** in
    the bottom **layer** of the frame tree. Consider this doubly layered frame tree:

        [[FOO, BAR], [BOO, BAZ]]

    The `refinery.lib.frame.FrameUnpacker` object will first be an iterator over the first frame
    `[FOO, BAR]`. After consuming this iterator, the `refinery.lib.frame.FrameUnpacker.nextframe`
    method can be called to load the next frame, at which point the object will become an
    iterator over `[BOO, BAZ]`.
    """
    next_chunk: Chunk | None
    depth: int
    trunk: tuple[int, ...]
    check: tuple[int, ...]
    items: Iterator[buf] | None
    finished: bool
    framed: bool

    def __init__(self, stream: BinaryIO | ChunkFeed | None):
        self.finished = False
        self.trunk = ()
        self.check = ()
        self.items = None
        self.depth = 0
        self.next_chunk = None
        self.framed = False
        items = stream.items if isinstance(stream, ChunkFeed) else _read_stream(stream)
        first = next(items, None)
        if first is None:
            self.next_chunk = Chunk()
        elif isinstance(first, Chunk):
            probe = object()
            if _references(first) > _references(probe) and (second := next(items, None)) is not None:
                items = itertools.chain((second,), items)
                del second
            if _references(first) > _references(probe):
                chunk = Chunk(first, ignore_chunk_properties=True)
            else:
                _adopt(chunk := first)
            del first
            for item in items:
                chunk.extend(item)
                del item
            self.next_chunk = chunk
        elif len(first) != MSIZE or first[:len(MAGIC)] != MAGIC:
            raise ValueError('The input began with an object that is neither a chunk nor a frame header.')
        else:
            self.depth = first[~0]
            self.framed = True
            self.items = items
            self._advance()

    def _advance(self) -> bool:
        self.next_chunk = None
        if not self.finished and (items := self.items) is not None:
            if (chunk := next(items, None)) is not None:
                if not isinstance(chunk, Chunk):
                    raise ValueError('A frame contained an object that is not a chunk.')
                self.next_chunk = chunk
                self.check = tuple(chunk.path)
                if chunk.scope != self.depth:
                    raise RuntimeError(F'Frame of depth {self.depth} contained chunk of scope {chunk.scope}.')
                return True
        self.finished = True
        return False

    def nextframe(self) -> bool:
        """
        Once the iterator is consumed, calling this function will return `True` if
        and only if another frame with input data has been loaded, in which case
        the object will provide an iterator over the freshly loaded frame. If this
        function returns `False`, all input data has been consumed.
        """
        if self.finished:
            return False
        self.trunk = self.check
        return True

    def abort(self):
        """
        Abort unpacking chunks from the frame.
        """
        if self.depth > 1:
            while not self.finished and self.trunk == self.check:
                self._advance()
        else:
            self.items = None
            self.next_chunk = None
            self.finished = True

    @property
    def eol(self) -> bool:
        """
        Specifies whether the current frame was fully consumed.
        """
        return self.trunk != self.peek

    @property
    def scopable(self) -> bool:
        """
        Whether the chunks of the next frame can be made visible; see `Chunk.scopable`.
        """
        chunk = self.next_chunk
        assert chunk is not None
        return chunk.scopable

    @property
    def peek(self) -> tuple[int, ...]:
        """
        Contains the identifier of the next frame.
        """
        return self.check

    def __iter__(self) -> Generator[Chunk]:
        if self.finished:
            return
        if not self.framed:
            yield self.next_chunk
            self.next_chunk = None
            self.finished = True
            return
        while not self.finished and self.trunk == self.check:
            yield self.next_chunk
            self._advance()


def _adopt(chunk: Chunk):
    """
    Makes the given chunk, which nothing else refers to, the unframed input of a unit.
    """
    chunk.uuid = uuid.uuid4()
    chunk.temp = None
    chunk._path = []
    chunk._view = []
    chunk._fill_scope = None
    chunk._fill_batch = None
    chunk._meta = LazyMetaOracle(chunk, scope=0, owned=True)


def _indexed(index: int, chunk: Chunk) -> Chunk:
    chunk.meta.index = index
    return chunk


class Framed:
    """
    A proxy interface to ingest and output framed data. It is given an `action` to be
    performed for each elementary chunk of data, a `stream` of input data, and an integer
    argument `nested` which specifies the relative amount of nesting to be performed
    by the interface. This parameter should either be `1` if the interface should output
    the results at an additional layer, `0` if the nesting depth of the data should
    remain unchanged, and a negative amount if frame layers are to be collapsed. After
    initialization, the `refinery.lib.frame.Framed` object is an iterator over the output of the
    operation with all framing already taken care of: Framed output is a frame header followed by
    one `refinery.lib.frame.Placement` for each chunk, and unframed output consists of chunks.
    """
    def __init__(
        self,
        action : Callable[[Chunk], Iterable[Chunk]],
        stream : BinaryIO | ChunkFeed | None,
        filter : Callable[[Iterable[Chunk]], Iterable[Chunk]],
        finish : Callable[[], Iterable[Chunk]],
        nesting: int = 0,
        squeeze: bool = False,
        filter_all: bool = False,
    ):
        self.unpack = FrameUnpacker(stream)
        self.action = action
        self.filter = filter
        self.finish = finish
        self.filter_all = filter_all
        self.nesting = nesting
        self.squeeze = squeeze

    def _apply_filter(self) -> Iterable[Chunk]:
        chunks = map(_indexed, itertools.count(), self.unpack)
        if self.unpack.scopable or self.filter_all:
            chunks = self.filter(chunks)
        yield from chunks

        if not self.unpack.eol:  # filter did not consume the iterable
            self.unpack.abort()

        if self.unpack.finished and self.finish:
            yield from self.finish()

    @property
    def unframed(self) -> bool:
        """
        This property is true if the output data is not framed.
        """
        return self.nesting + self.unpack.depth < 1

    @property
    def framebreak(self) -> bool:
        """
        This property will be true if the data generated by this framing interface should
        be separated by linebreaks. This happens when one of the following is true:
        - The requested nesting was smaller than required to close all existing frames.
        - The input data was not framed and the nesting did not increase in this unit
        """
        if not self.unpack.framed:
            return self.nesting < 1
        return self.nesting + self.unpack.depth < 0

    def _results(self, chunk: Chunk) -> Iterator[Chunk]:
        if chunk.visible:
            return self._generate_chunks(chunk)
        return HandOver(chunk)

    def _all_results(self) -> Iterator[Chunk]:
        while self.unpack.nextframe():
            yield from itertools.chain.from_iterable(map(self._results, self._apply_filter()))

    @staticmethod
    def _merged(pieces: Iterable[Chunk]) -> Iterator[Chunk]:
        """
        Concatenates the given chunks into at most one chunk, which keeps the metadata that all of
        them share. A single chunk is passed on as it is. Otherwise, the result is a new chunk,
        because the unit that produced the pieces may still use them, and the same chunk may occur
        more than once. The metadata of each piece is taken when the piece arrives, because a unit
        may label the same chunk again before passing it on once more.
        """
        it = iter(pieces)
        for first in it:
            break
        else:
            return HandOver(None)
        merged = first.copy(data=False)
        shared = True
        for piece in it:
            if first is not None:
                merged.extend(first)
                first = None
            if shared:
                shared = bool(merged.intersect(piece).meta)
            merged.extend(piece)
            del piece
        return HandOver(merged if first is None else first)

    def _generate_chunks(self, parent: Chunk):
        """
        Generates the outputs of the unit for the given input, placed where the input is and with
        its metadata. A squeezed output is merged once the unit has produced all of its pieces, so
        it inherits the metadata that the input had before the unit ran: The unit may have labelled
        its input in the meantime.
        """
        path = list(parent.path)
        view = list(parent.view)
        origin = parent.copy(data=False) if self.squeeze else parent
        meta = origin.meta
        scope = parent.scope

        def inherit(chunk: Chunk):
            if chunk is parent:
                return chunk
            if path:
                chunk._path[:] = path
            if view and not chunk._view:
                chunk._view[:] = view
            chunk._meta.inherit(meta)
            return chunk.truncate(scope)

        yield from map(inherit, self._generate_bytes(parent))

    def _generate_bytes(self, data: Chunk):
        try:
            outputs = self.action(data)
            if self.squeeze:
                outputs = self._merged(outputs)
            yield from outputs
        finally:
            data.temp = None

    def __iter__(self):
        nesting = self.nesting
        scope = max(self.unpack.depth + nesting, 0)
        if self.unpack.finished:
            if scope:
                yield generate_frame_header(scope)
            return
        repeat = itertools.repeat
        if nesting > 0:
            assert scope
            yield generate_frame_header(scope)
            nestings = repeat(nesting)
            while self.unpack.nextframe():
                for position, results in enumerate(map(self._results, self._apply_filter())):
                    yield from map(Placement, results, nestings, repeat(position))
        elif not self.unpack.framed:
            yield from itertools.chain.from_iterable(map(self._generate_bytes, self._apply_filter()))
        elif nesting == 0:
            assert scope
            yield generate_frame_header(scope)
            yield from map(Placement, self._all_results(), repeat(0), repeat(0))
        elif not scope:
            yield from map(Chunk.truncate, self._all_results())
        else:
            yield generate_frame_header(scope)
            check = scope + 1
            groups = itertools.groupby(self._all_results(), lambda result: result.path[:check])
            merged = itertools.chain.from_iterable(self._merged(group) for _, group in groups)
            yield from map(Placement, merged, repeat(nesting), repeat(0))

Global variables

var MAGIC: This is the magic signature that is used by refinery to prefix serialized frame data. If a unit reads data from STDIN that is prefixed with these bytes, it assumes that serialized frame data follows. Otherwise, the input is treated as a single unframed chunk.
var MSIZE: This is the length of the data returned by generate_frame_header().

Functions

def generate_frame_header(scope)

This function generates a frame header for a frame tree of depth equal to scope. The depth is encoded as a single byte following MAGIC. This implies a depth limit of 255 for frame trees in refinery, and I dearly hope that noone is insane enough to build a refinery pipeline that would be affected.

Expand source code Browse git

def generate_frame_header(scope: int):
    """
    This function generates a frame header for a frame tree of depth equal to `scope`. The depth
    is encoded as a single byte following `refinery.lib.frame.MAGIC`. This implies a depth limit
    of 255 for frame trees in refinery, and I dearly hope that noone is insane enough to build a
    refinery pipeline that would be affected.
    """
    if scope > 0xFF:
        raise ValueError('Maximum frame depth exceeded.')
    return B'%s%c' % (MAGIC, scope)

Classes

class Chunk (data=None, path=None, view=None, meta=None, seed=None, fill_scope=None, fill_batch=None, ignore_chunk_properties=False)

Represents the individual chunks in a frame. The Unit.filter() method receives an iterable of Chunks.

Expand source code Browse git

class Chunk(bytearray):
    """
    Represents the individual chunks in a frame. The `refinery.units.Unit.filter` method receives
    an iterable of `refinery.lib.frame.Chunk`s.
    """
    temp: Any
    """
    Units can use this field to transport temporary data from `refinery.units.Unit.filter` to
    `refinery.units.Unit.process` or `refinery.units.Unit.reverse` for the same chunk. The frame
    handler clears it once the unit has processed the chunk, so that it may refer back to the chunk.
    """
    uuid: uuid.UUID
    """
    Each chunk object carries a unique identifier. The `refinery.units.DelayedArgumentProxy` uses
    this property to check whether `refinery.units.Unit` command-line arguments were previously
    evaluated against this chunk. Otherwise `refinery.lib.argformats.DelayedArgument`s that alter
    the input data could produce unexpected results when the argument proxy is mapped against the
    same chunk twice.
    """

    __slots__ = (
        '_meta',
        '_view',
        '_path',
        '_fill_scope',
        '_fill_batch',
        'temp',
        'uuid',
        '__weakref__',
    )

    def __init__(
        self,
        data: buf | int | None = None,
        path: list[int] | None = None,
        view: list[bool] | None = None,
        meta: Mapping[str, Any] | None = None,
        seed: dict[str, list] | None = None,
        fill_scope: bool | None = None,
        fill_batch: int | None = None,
        ignore_chunk_properties: bool = False,
    ):
        if data is None:
            bytearray.__init__(self)
        else:
            bytearray.__init__(self, data)

        self.uuid = uuid.uuid4()
        self.temp = None

        if path is None:
            path = []
        if view is None:
            view = [False] * len(path)
        elif len(view) != len(path):
            raise ValueError('view must have the same length as path')

        if not ignore_chunk_properties and isinstance(data, Chunk):
            path = path or list(data.path)
            view = view or list(data.view)
            meta = meta or data.meta
            fill_scope = fill_scope or data._fill_scope
            fill_batch = fill_batch or data._fill_batch

        self._view: list[bool] = view
        self._path: list[int] = path
        self._fill_scope: bool | None = fill_scope
        self._fill_batch: bool | None = fill_batch

        self._meta = m = LazyMetaOracle(self, scope=self.scope, seed=seed, owned=True)
        if meta is not None:
            m.update(meta)

    @classmethod
    def Wrap(cls, data):
        if isinstance(data, cls):
            return data
        return cls(data)

    def set_next_scope(self, visible: bool) -> None:
        self._fill_scope = visible

    def set_next_batch(self, batch: int) -> None:
        """
        This function allows units to emit trees of depth one rather than lists. When a unit emits
        a chunk at index `a`, sets the next batch to `b`, and when a double frame opens after this
        unit's invocation, then said chunk will have `a/b` added to its path. By default, `b` would
        always be `0`. For example, the `refinery.rex` unit uses this feature. As a result:

            $ emit #1yellow-#3red-#2orange | rex #(.)([a-z]+) {1} {2} [[| pop x:e ]| rep v:x ]]
            yellow
            red
            red
            red
            orange
            orange

        The double frame after `refinery.rex` looks like this:

            [[1,yellow],[2,red],[3,orange]]

        By default, the frame would simply look like this:

            [[1,yellow,2,red,3,orange]]

        This feature is useful for `refinery.units.Unit`s that produce multiple outputs for each of
        a number of intermediate results - in the case of `refinery.rex`, that intermediate result
        is a regular expression match, and `refinery.rex` allows to produce different outputs for
        each of those.
        """
        self._fill_batch = batch

    @property
    def scope(self) -> int:
        """
        This value is the length of `refinery.lib.frame.Chunk.path` and therefore corresponds to
        the depth of the frame tree. It is called "scope" because it is equally the scope at which
        new metadata variables for this chunk will be created.
        """
        return len(self._path)

    @property
    def view(self) -> list[bool]:
        """
        This tuple of boolean values indicates the visibility of this chunk at each layer of
        the frame tree. The `refinery.scope` unit can be used to change visibility of chunks
        within a frame.
        """
        return self._view

    @property
    def path(self) -> list[int]:
        """
        The vertices in each frame tree layer are sequentially numbered by their order of
        appearance in the stream. The `refinery.lib.frame.Chunk.path` contains the numbers of
        the vertices (in each layer) which define the path from the root of the frame tree
        to the leaf vertex representing this `refinery.lib.frame.Chunk`
        """
        return self._path

    @property
    def meta(self) -> LazyMetaOracle:
        """
        Every chunk can contain a dictionary of arbitrary metadata. Further details about this data
        are available in the module-level documetnation of `refinery.lib.meta`.
        """
        return self._meta

    @property
    def visible(self):
        """
        This property defines whether the chunk is currently visible. It defaults to true if the
        chunk is not part of a frame and is otherwise the same as the last element of the tuple
        `refinery.lib.frame.Chunk.view`. Setting this property will correspondingly alter the last
        entry of `refinery.lib.frame.Chunk.view`.
        Setting this property on an unframed `refinery.lib.frame.Chunk` raises an `AttributeError`.
        """
        return not self._view or self._view[~0]

    @property
    def scopable(self):
        """
        This property defines whether the chunk can be made visible in the current frame.
        """
        return len(self._view) <= 1 or self._view[~1]

    @visible.setter
    def visible(self, value: bool):
        view = self._view
        if not view:
            if not value:
                raise AttributeError('cannot make chunk invisible outside frame')
        else:
            view[~0] = value

    def _relocated(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, serialized metadata, and fill scope of this chunk after it has been
        moved by the given nesting and placed at the given position.
        """
        scope = self.scope + nest
        fs = self._fill_scope
        fb = self._fill_batch
        if nest > 0:
            view = list(self._view)
            path = list(self._path)
            if fs is not None:
                view.extend(itertools.repeat(self.visible, nest - 1))
                view.append(fs)
                fs = None
            else:
                view.extend(itertools.repeat(self.visible, nest))
            if fb is not None and nest > 1:
                path.append(position)
                path.append(fb)
                path.extend(itertools.repeat(0, nest - 2))
            else:
                path.append(position)
                path.extend(itertools.repeat(0, nest - 1))
        elif nest < 0:
            view = self._view[:nest]
            path = self._path[:nest]
        else:
            view = list(self._view)
            path = list(self._path)
        assert len(path) == scope
        assert len(view) == scope
        return path, view, self._meta.serialize(scope), fs

    def serialize(self, nest: int = 0, position: int = 0) -> tuple[bytes | bytearray, ...]:
        """
        Returns the serialized form of this chunk, moved by the given nesting and placed at the
        given position, as a sequence of buffers to be written in order. Their concatenation is a
        msgpack array of the path, view, metadata, fill scope, and data. For a large chunk, the
        first buffer ends with the header of the data, and the chunk itself follows uncopied.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        if (size := len(self)) < _INLINE:
            packed = msgpack.packb((path, view, meta, fs, self))
            assert isinstance(packed, bytes)
            return packed,
        packed = msgpack.packb((path, view, meta, fs))
        assert isinstance(packed, bytes)
        head = bytearray(packed)
        head[0] += 1
        head.extend(_bin_header(size))
        return head, self

    def pack(self, nest: int = 0, position: int = 0) -> bytes:
        """
        Returns the serialized form of this chunk as a single byte string; see
        `refinery.lib.frame.Chunk.serialize`.
        """
        return B''.join(self.serialize(nest, position))

    def _moved(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, metadata, and fill scope that a unit in the same process receives
        for this chunk when it is moved by the given nesting and placed at the given position. The
        metadata is serialized and read back as in a stream.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        meta = msgpack.unpackb(msgpack.packb(meta), use_list=True, strict_map_key=False)
        return path, view, meta, fs

    @staticmethod
    def _received(data: buf, moved: tuple[list[int], list[bool], dict, bool | None]) -> Chunk:
        """
        Returns a chunk with a copy of the given data and the path, view, metadata, and fill scope
        returned by `Chunk._moved`.
        """
        path, view, meta, fs = moved
        return Chunk(data, path, view, None, meta, fs, None, ignore_chunk_properties=True)

    def transfer(self, nest: int = 0, position: int = 0) -> Chunk:
        """
        Returns the chunk that a unit in the same process receives when this chunk is passed on,
        moved by the given nesting and placed at the given position. The data is copied, and the
        metadata is serialized and read back as in a stream: The receiver gets values of its own,
        and a value that cannot be serialized raises the same error as in `Chunk.serialize`.
        """
        return Chunk._received(self, self._moved(nest, position))

    def __repr__(self) -> str:
        layer = '/'.join(str(p) if s else F'!{p}' for p, s in zip(self._path, self._view))
        layer = layer and '/' + layer
        return F'<chunk{layer}:{bytes(self)!r}>'

    def intersect(self, other: Chunk):
        """
        Removes all meta variables from this chunk whose value differs from those of the `other`
        inut chunk.
        """
        other_meta = other._meta
        meta = self._meta
        for key, value in list(meta.items()):
            if other_meta.get(key) != value:
                meta.discard(key)
        return self

    def __str__(self):
        try:
            return self.decode('UTF8')
        except UnicodeDecodeError:
            return self.hex()

    def __hash__(self):
        return hash(zlib.adler32(self))

    @overload
    def __getitem__(self, k: str) -> int | str | float | bytes | None:
        ...

    @overload
    def __getitem__(self, key: SupportsIndex, /) -> int:
        ...

    @overload
    def __getitem__(self, key: slice, /) -> bytearray:
        ...

    def __getitem__(self, k):
        if not isinstance(k, str):
            return bytearray.__getitem__(self, k)
        return self._meta.__getitem__(k)

    def __setitem__(self, bounds, value):
        if isinstance(bounds, str):
            self._meta[bounds] = value
        else:
            bytearray.__setitem__(self, bounds, value)

    def meta_var_int(self, key: str) -> int:
        value = self._meta.__getitem__(key)
        if not isinstance(value, int):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not int.')
        return value

    def meta_var_str(self, key: str) -> str:
        value = self._meta.__getitem__(key)
        if not isinstance(value, str):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not str.')
        return value

    def truncate(self, scope: int = 0):
        """
        Truncate the `refinery.lib.frame.Chunk.path` and `refinery.lib.frame.Chunk.view` lists
        to the given length, setting the `refinery.lib.frame.Chunk.scope` to the given value.
        """
        del self._path[scope:]
        del self._view[scope:]
        return self

    def copy(self, meta=True, data=True) -> Chunk:
        """
        Produce a copy of this chunk. The metadata is copied if the `meta` argument is `True`,
        otherwise the copy has no metadata. The body of the chunk is copied only if the `data`
        argument is `True`.
        """
        data = data and self or None
        copy = Chunk(
            data,
            path=list(self._path),
            view=list(self._view),
            fill_scope=self._fill_scope,
            fill_batch=self._fill_batch,
            ignore_chunk_properties=True,
        )
        if meta:
            copy.meta.update(self.meta)
        if copy.meta.scope != copy.scope:
            raise RuntimeError
        return copy

    def __copy__(self):
        return self.copy()

    def __deepcopy__(self, memo):
        raise NotImplementedError

Ancestors

builtins.bytearray

Static methods

def Wrap(data)

Instance variables

var scope

This value is the length of Chunk.path and therefore corresponds to the depth of the frame tree. It is called "scope" because it is equally the scope at which new metadata variables for this chunk will be created.

Expand source code Browse git

@property
def scope(self) -> int:
    """
    This value is the length of `refinery.lib.frame.Chunk.path` and therefore corresponds to
    the depth of the frame tree. It is called "scope" because it is equally the scope at which
    new metadata variables for this chunk will be created.
    """
    return len(self._path)

var view

This tuple of boolean values indicates the visibility of this chunk at each layer of the frame tree. The scope unit can be used to change visibility of chunks within a frame.

Expand source code Browse git

@property
def view(self) -> list[bool]:
    """
    This tuple of boolean values indicates the visibility of this chunk at each layer of
    the frame tree. The `refinery.scope` unit can be used to change visibility of chunks
    within a frame.
    """
    return self._view

var path

The vertices in each frame tree layer are sequentially numbered by their order of appearance in the stream. The Chunk.path contains the numbers of the vertices (in each layer) which define the path from the root of the frame tree to the leaf vertex representing this Chunk

Expand source code Browse git

@property
def path(self) -> list[int]:
    """
    The vertices in each frame tree layer are sequentially numbered by their order of
    appearance in the stream. The `refinery.lib.frame.Chunk.path` contains the numbers of
    the vertices (in each layer) which define the path from the root of the frame tree
    to the leaf vertex representing this `refinery.lib.frame.Chunk`
    """
    return self._path

var meta

Every chunk can contain a dictionary of arbitrary metadata. Further details about this data are available in the module-level documetnation of refinery.lib.meta.

Expand source code Browse git

@property
def meta(self) -> LazyMetaOracle:
    """
    Every chunk can contain a dictionary of arbitrary metadata. Further details about this data
    are available in the module-level documetnation of `refinery.lib.meta`.
    """
    return self._meta

var visible

This property defines whether the chunk is currently visible. It defaults to true if the chunk is not part of a frame and is otherwise the same as the last element of the tuple Chunk.view. Setting this property will correspondingly alter the last entry of Chunk.view. Setting this property on an unframed Chunk raises an AttributeError.

Expand source code Browse git

@property
def visible(self):
    """
    This property defines whether the chunk is currently visible. It defaults to true if the
    chunk is not part of a frame and is otherwise the same as the last element of the tuple
    `refinery.lib.frame.Chunk.view`. Setting this property will correspondingly alter the last
    entry of `refinery.lib.frame.Chunk.view`.
    Setting this property on an unframed `refinery.lib.frame.Chunk` raises an `AttributeError`.
    """
    return not self._view or self._view[~0]

var scopable

This property defines whether the chunk can be made visible in the current frame.

Expand source code Browse git

@property
def scopable(self):
    """
    This property defines whether the chunk can be made visible in the current frame.
    """
    return len(self._view) <= 1 or self._view[~1]

var temp

Units can use this field to transport temporary data from Unit.filter() to Unit.process() or UnitBase.reverse() for the same chunk. The frame handler clears it once the unit has processed the chunk, so that it may refer back to the chunk.

Expand source code Browse git

class Chunk(bytearray):
    """
    Represents the individual chunks in a frame. The `refinery.units.Unit.filter` method receives
    an iterable of `refinery.lib.frame.Chunk`s.
    """
    temp: Any
    """
    Units can use this field to transport temporary data from `refinery.units.Unit.filter` to
    `refinery.units.Unit.process` or `refinery.units.Unit.reverse` for the same chunk. The frame
    handler clears it once the unit has processed the chunk, so that it may refer back to the chunk.
    """
    uuid: uuid.UUID
    """
    Each chunk object carries a unique identifier. The `refinery.units.DelayedArgumentProxy` uses
    this property to check whether `refinery.units.Unit` command-line arguments were previously
    evaluated against this chunk. Otherwise `refinery.lib.argformats.DelayedArgument`s that alter
    the input data could produce unexpected results when the argument proxy is mapped against the
    same chunk twice.
    """

    __slots__ = (
        '_meta',
        '_view',
        '_path',
        '_fill_scope',
        '_fill_batch',
        'temp',
        'uuid',
        '__weakref__',
    )

    def __init__(
        self,
        data: buf | int | None = None,
        path: list[int] | None = None,
        view: list[bool] | None = None,
        meta: Mapping[str, Any] | None = None,
        seed: dict[str, list] | None = None,
        fill_scope: bool | None = None,
        fill_batch: int | None = None,
        ignore_chunk_properties: bool = False,
    ):
        if data is None:
            bytearray.__init__(self)
        else:
            bytearray.__init__(self, data)

        self.uuid = uuid.uuid4()
        self.temp = None

        if path is None:
            path = []
        if view is None:
            view = [False] * len(path)
        elif len(view) != len(path):
            raise ValueError('view must have the same length as path')

        if not ignore_chunk_properties and isinstance(data, Chunk):
            path = path or list(data.path)
            view = view or list(data.view)
            meta = meta or data.meta
            fill_scope = fill_scope or data._fill_scope
            fill_batch = fill_batch or data._fill_batch

        self._view: list[bool] = view
        self._path: list[int] = path
        self._fill_scope: bool | None = fill_scope
        self._fill_batch: bool | None = fill_batch

        self._meta = m = LazyMetaOracle(self, scope=self.scope, seed=seed, owned=True)
        if meta is not None:
            m.update(meta)

    @classmethod
    def Wrap(cls, data):
        if isinstance(data, cls):
            return data
        return cls(data)

    def set_next_scope(self, visible: bool) -> None:
        self._fill_scope = visible

    def set_next_batch(self, batch: int) -> None:
        """
        This function allows units to emit trees of depth one rather than lists. When a unit emits
        a chunk at index `a`, sets the next batch to `b`, and when a double frame opens after this
        unit's invocation, then said chunk will have `a/b` added to its path. By default, `b` would
        always be `0`. For example, the `refinery.rex` unit uses this feature. As a result:

            $ emit #1yellow-#3red-#2orange | rex #(.)([a-z]+) {1} {2} [[| pop x:e ]| rep v:x ]]
            yellow
            red
            red
            red
            orange
            orange

        The double frame after `refinery.rex` looks like this:

            [[1,yellow],[2,red],[3,orange]]

        By default, the frame would simply look like this:

            [[1,yellow,2,red,3,orange]]

        This feature is useful for `refinery.units.Unit`s that produce multiple outputs for each of
        a number of intermediate results - in the case of `refinery.rex`, that intermediate result
        is a regular expression match, and `refinery.rex` allows to produce different outputs for
        each of those.
        """
        self._fill_batch = batch

    @property
    def scope(self) -> int:
        """
        This value is the length of `refinery.lib.frame.Chunk.path` and therefore corresponds to
        the depth of the frame tree. It is called "scope" because it is equally the scope at which
        new metadata variables for this chunk will be created.
        """
        return len(self._path)

    @property
    def view(self) -> list[bool]:
        """
        This tuple of boolean values indicates the visibility of this chunk at each layer of
        the frame tree. The `refinery.scope` unit can be used to change visibility of chunks
        within a frame.
        """
        return self._view

    @property
    def path(self) -> list[int]:
        """
        The vertices in each frame tree layer are sequentially numbered by their order of
        appearance in the stream. The `refinery.lib.frame.Chunk.path` contains the numbers of
        the vertices (in each layer) which define the path from the root of the frame tree
        to the leaf vertex representing this `refinery.lib.frame.Chunk`
        """
        return self._path

    @property
    def meta(self) -> LazyMetaOracle:
        """
        Every chunk can contain a dictionary of arbitrary metadata. Further details about this data
        are available in the module-level documetnation of `refinery.lib.meta`.
        """
        return self._meta

    @property
    def visible(self):
        """
        This property defines whether the chunk is currently visible. It defaults to true if the
        chunk is not part of a frame and is otherwise the same as the last element of the tuple
        `refinery.lib.frame.Chunk.view`. Setting this property will correspondingly alter the last
        entry of `refinery.lib.frame.Chunk.view`.
        Setting this property on an unframed `refinery.lib.frame.Chunk` raises an `AttributeError`.
        """
        return not self._view or self._view[~0]

    @property
    def scopable(self):
        """
        This property defines whether the chunk can be made visible in the current frame.
        """
        return len(self._view) <= 1 or self._view[~1]

    @visible.setter
    def visible(self, value: bool):
        view = self._view
        if not view:
            if not value:
                raise AttributeError('cannot make chunk invisible outside frame')
        else:
            view[~0] = value

    def _relocated(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, serialized metadata, and fill scope of this chunk after it has been
        moved by the given nesting and placed at the given position.
        """
        scope = self.scope + nest
        fs = self._fill_scope
        fb = self._fill_batch
        if nest > 0:
            view = list(self._view)
            path = list(self._path)
            if fs is not None:
                view.extend(itertools.repeat(self.visible, nest - 1))
                view.append(fs)
                fs = None
            else:
                view.extend(itertools.repeat(self.visible, nest))
            if fb is not None and nest > 1:
                path.append(position)
                path.append(fb)
                path.extend(itertools.repeat(0, nest - 2))
            else:
                path.append(position)
                path.extend(itertools.repeat(0, nest - 1))
        elif nest < 0:
            view = self._view[:nest]
            path = self._path[:nest]
        else:
            view = list(self._view)
            path = list(self._path)
        assert len(path) == scope
        assert len(view) == scope
        return path, view, self._meta.serialize(scope), fs

    def serialize(self, nest: int = 0, position: int = 0) -> tuple[bytes | bytearray, ...]:
        """
        Returns the serialized form of this chunk, moved by the given nesting and placed at the
        given position, as a sequence of buffers to be written in order. Their concatenation is a
        msgpack array of the path, view, metadata, fill scope, and data. For a large chunk, the
        first buffer ends with the header of the data, and the chunk itself follows uncopied.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        if (size := len(self)) < _INLINE:
            packed = msgpack.packb((path, view, meta, fs, self))
            assert isinstance(packed, bytes)
            return packed,
        packed = msgpack.packb((path, view, meta, fs))
        assert isinstance(packed, bytes)
        head = bytearray(packed)
        head[0] += 1
        head.extend(_bin_header(size))
        return head, self

    def pack(self, nest: int = 0, position: int = 0) -> bytes:
        """
        Returns the serialized form of this chunk as a single byte string; see
        `refinery.lib.frame.Chunk.serialize`.
        """
        return B''.join(self.serialize(nest, position))

    def _moved(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, metadata, and fill scope that a unit in the same process receives
        for this chunk when it is moved by the given nesting and placed at the given position. The
        metadata is serialized and read back as in a stream.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        meta = msgpack.unpackb(msgpack.packb(meta), use_list=True, strict_map_key=False)
        return path, view, meta, fs

    @staticmethod
    def _received(data: buf, moved: tuple[list[int], list[bool], dict, bool | None]) -> Chunk:
        """
        Returns a chunk with a copy of the given data and the path, view, metadata, and fill scope
        returned by `Chunk._moved`.
        """
        path, view, meta, fs = moved
        return Chunk(data, path, view, None, meta, fs, None, ignore_chunk_properties=True)

    def transfer(self, nest: int = 0, position: int = 0) -> Chunk:
        """
        Returns the chunk that a unit in the same process receives when this chunk is passed on,
        moved by the given nesting and placed at the given position. The data is copied, and the
        metadata is serialized and read back as in a stream: The receiver gets values of its own,
        and a value that cannot be serialized raises the same error as in `Chunk.serialize`.
        """
        return Chunk._received(self, self._moved(nest, position))

    def __repr__(self) -> str:
        layer = '/'.join(str(p) if s else F'!{p}' for p, s in zip(self._path, self._view))
        layer = layer and '/' + layer
        return F'<chunk{layer}:{bytes(self)!r}>'

    def intersect(self, other: Chunk):
        """
        Removes all meta variables from this chunk whose value differs from those of the `other`
        inut chunk.
        """
        other_meta = other._meta
        meta = self._meta
        for key, value in list(meta.items()):
            if other_meta.get(key) != value:
                meta.discard(key)
        return self

    def __str__(self):
        try:
            return self.decode('UTF8')
        except UnicodeDecodeError:
            return self.hex()

    def __hash__(self):
        return hash(zlib.adler32(self))

    @overload
    def __getitem__(self, k: str) -> int | str | float | bytes | None:
        ...

    @overload
    def __getitem__(self, key: SupportsIndex, /) -> int:
        ...

    @overload
    def __getitem__(self, key: slice, /) -> bytearray:
        ...

    def __getitem__(self, k):
        if not isinstance(k, str):
            return bytearray.__getitem__(self, k)
        return self._meta.__getitem__(k)

    def __setitem__(self, bounds, value):
        if isinstance(bounds, str):
            self._meta[bounds] = value
        else:
            bytearray.__setitem__(self, bounds, value)

    def meta_var_int(self, key: str) -> int:
        value = self._meta.__getitem__(key)
        if not isinstance(value, int):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not int.')
        return value

    def meta_var_str(self, key: str) -> str:
        value = self._meta.__getitem__(key)
        if not isinstance(value, str):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not str.')
        return value

    def truncate(self, scope: int = 0):
        """
        Truncate the `refinery.lib.frame.Chunk.path` and `refinery.lib.frame.Chunk.view` lists
        to the given length, setting the `refinery.lib.frame.Chunk.scope` to the given value.
        """
        del self._path[scope:]
        del self._view[scope:]
        return self

    def copy(self, meta=True, data=True) -> Chunk:
        """
        Produce a copy of this chunk. The metadata is copied if the `meta` argument is `True`,
        otherwise the copy has no metadata. The body of the chunk is copied only if the `data`
        argument is `True`.
        """
        data = data and self or None
        copy = Chunk(
            data,
            path=list(self._path),
            view=list(self._view),
            fill_scope=self._fill_scope,
            fill_batch=self._fill_batch,
            ignore_chunk_properties=True,
        )
        if meta:
            copy.meta.update(self.meta)
        if copy.meta.scope != copy.scope:
            raise RuntimeError
        return copy

    def __copy__(self):
        return self.copy()

    def __deepcopy__(self, memo):
        raise NotImplementedError

var uuid

Each chunk object carries a unique identifier. The DelayedArgumentProxy uses this property to check whether Unit command-line arguments were previously evaluated against this chunk. Otherwise DelayedArguments that alter the input data could produce unexpected results when the argument proxy is mapped against the same chunk twice.

Expand source code Browse git

class Chunk(bytearray):
    """
    Represents the individual chunks in a frame. The `refinery.units.Unit.filter` method receives
    an iterable of `refinery.lib.frame.Chunk`s.
    """
    temp: Any
    """
    Units can use this field to transport temporary data from `refinery.units.Unit.filter` to
    `refinery.units.Unit.process` or `refinery.units.Unit.reverse` for the same chunk. The frame
    handler clears it once the unit has processed the chunk, so that it may refer back to the chunk.
    """
    uuid: uuid.UUID
    """
    Each chunk object carries a unique identifier. The `refinery.units.DelayedArgumentProxy` uses
    this property to check whether `refinery.units.Unit` command-line arguments were previously
    evaluated against this chunk. Otherwise `refinery.lib.argformats.DelayedArgument`s that alter
    the input data could produce unexpected results when the argument proxy is mapped against the
    same chunk twice.
    """

    __slots__ = (
        '_meta',
        '_view',
        '_path',
        '_fill_scope',
        '_fill_batch',
        'temp',
        'uuid',
        '__weakref__',
    )

    def __init__(
        self,
        data: buf | int | None = None,
        path: list[int] | None = None,
        view: list[bool] | None = None,
        meta: Mapping[str, Any] | None = None,
        seed: dict[str, list] | None = None,
        fill_scope: bool | None = None,
        fill_batch: int | None = None,
        ignore_chunk_properties: bool = False,
    ):
        if data is None:
            bytearray.__init__(self)
        else:
            bytearray.__init__(self, data)

        self.uuid = uuid.uuid4()
        self.temp = None

        if path is None:
            path = []
        if view is None:
            view = [False] * len(path)
        elif len(view) != len(path):
            raise ValueError('view must have the same length as path')

        if not ignore_chunk_properties and isinstance(data, Chunk):
            path = path or list(data.path)
            view = view or list(data.view)
            meta = meta or data.meta
            fill_scope = fill_scope or data._fill_scope
            fill_batch = fill_batch or data._fill_batch

        self._view: list[bool] = view
        self._path: list[int] = path
        self._fill_scope: bool | None = fill_scope
        self._fill_batch: bool | None = fill_batch

        self._meta = m = LazyMetaOracle(self, scope=self.scope, seed=seed, owned=True)
        if meta is not None:
            m.update(meta)

    @classmethod
    def Wrap(cls, data):
        if isinstance(data, cls):
            return data
        return cls(data)

    def set_next_scope(self, visible: bool) -> None:
        self._fill_scope = visible

    def set_next_batch(self, batch: int) -> None:
        """
        This function allows units to emit trees of depth one rather than lists. When a unit emits
        a chunk at index `a`, sets the next batch to `b`, and when a double frame opens after this
        unit's invocation, then said chunk will have `a/b` added to its path. By default, `b` would
        always be `0`. For example, the `refinery.rex` unit uses this feature. As a result:

            $ emit #1yellow-#3red-#2orange | rex #(.)([a-z]+) {1} {2} [[| pop x:e ]| rep v:x ]]
            yellow
            red
            red
            red
            orange
            orange

        The double frame after `refinery.rex` looks like this:

            [[1,yellow],[2,red],[3,orange]]

        By default, the frame would simply look like this:

            [[1,yellow,2,red,3,orange]]

        This feature is useful for `refinery.units.Unit`s that produce multiple outputs for each of
        a number of intermediate results - in the case of `refinery.rex`, that intermediate result
        is a regular expression match, and `refinery.rex` allows to produce different outputs for
        each of those.
        """
        self._fill_batch = batch

    @property
    def scope(self) -> int:
        """
        This value is the length of `refinery.lib.frame.Chunk.path` and therefore corresponds to
        the depth of the frame tree. It is called "scope" because it is equally the scope at which
        new metadata variables for this chunk will be created.
        """
        return len(self._path)

    @property
    def view(self) -> list[bool]:
        """
        This tuple of boolean values indicates the visibility of this chunk at each layer of
        the frame tree. The `refinery.scope` unit can be used to change visibility of chunks
        within a frame.
        """
        return self._view

    @property
    def path(self) -> list[int]:
        """
        The vertices in each frame tree layer are sequentially numbered by their order of
        appearance in the stream. The `refinery.lib.frame.Chunk.path` contains the numbers of
        the vertices (in each layer) which define the path from the root of the frame tree
        to the leaf vertex representing this `refinery.lib.frame.Chunk`
        """
        return self._path

    @property
    def meta(self) -> LazyMetaOracle:
        """
        Every chunk can contain a dictionary of arbitrary metadata. Further details about this data
        are available in the module-level documetnation of `refinery.lib.meta`.
        """
        return self._meta

    @property
    def visible(self):
        """
        This property defines whether the chunk is currently visible. It defaults to true if the
        chunk is not part of a frame and is otherwise the same as the last element of the tuple
        `refinery.lib.frame.Chunk.view`. Setting this property will correspondingly alter the last
        entry of `refinery.lib.frame.Chunk.view`.
        Setting this property on an unframed `refinery.lib.frame.Chunk` raises an `AttributeError`.
        """
        return not self._view or self._view[~0]

    @property
    def scopable(self):
        """
        This property defines whether the chunk can be made visible in the current frame.
        """
        return len(self._view) <= 1 or self._view[~1]

    @visible.setter
    def visible(self, value: bool):
        view = self._view
        if not view:
            if not value:
                raise AttributeError('cannot make chunk invisible outside frame')
        else:
            view[~0] = value

    def _relocated(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, serialized metadata, and fill scope of this chunk after it has been
        moved by the given nesting and placed at the given position.
        """
        scope = self.scope + nest
        fs = self._fill_scope
        fb = self._fill_batch
        if nest > 0:
            view = list(self._view)
            path = list(self._path)
            if fs is not None:
                view.extend(itertools.repeat(self.visible, nest - 1))
                view.append(fs)
                fs = None
            else:
                view.extend(itertools.repeat(self.visible, nest))
            if fb is not None and nest > 1:
                path.append(position)
                path.append(fb)
                path.extend(itertools.repeat(0, nest - 2))
            else:
                path.append(position)
                path.extend(itertools.repeat(0, nest - 1))
        elif nest < 0:
            view = self._view[:nest]
            path = self._path[:nest]
        else:
            view = list(self._view)
            path = list(self._path)
        assert len(path) == scope
        assert len(view) == scope
        return path, view, self._meta.serialize(scope), fs

    def serialize(self, nest: int = 0, position: int = 0) -> tuple[bytes | bytearray, ...]:
        """
        Returns the serialized form of this chunk, moved by the given nesting and placed at the
        given position, as a sequence of buffers to be written in order. Their concatenation is a
        msgpack array of the path, view, metadata, fill scope, and data. For a large chunk, the
        first buffer ends with the header of the data, and the chunk itself follows uncopied.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        if (size := len(self)) < _INLINE:
            packed = msgpack.packb((path, view, meta, fs, self))
            assert isinstance(packed, bytes)
            return packed,
        packed = msgpack.packb((path, view, meta, fs))
        assert isinstance(packed, bytes)
        head = bytearray(packed)
        head[0] += 1
        head.extend(_bin_header(size))
        return head, self

    def pack(self, nest: int = 0, position: int = 0) -> bytes:
        """
        Returns the serialized form of this chunk as a single byte string; see
        `refinery.lib.frame.Chunk.serialize`.
        """
        return B''.join(self.serialize(nest, position))

    def _moved(self, nest: int, position: int) -> tuple[list[int], list[bool], dict, bool | None]:
        """
        Returns the path, view, metadata, and fill scope that a unit in the same process receives
        for this chunk when it is moved by the given nesting and placed at the given position. The
        metadata is serialized and read back as in a stream.
        """
        assert msgpack
        path, view, meta, fs = self._relocated(nest, position)
        meta = msgpack.unpackb(msgpack.packb(meta), use_list=True, strict_map_key=False)
        return path, view, meta, fs

    @staticmethod
    def _received(data: buf, moved: tuple[list[int], list[bool], dict, bool | None]) -> Chunk:
        """
        Returns a chunk with a copy of the given data and the path, view, metadata, and fill scope
        returned by `Chunk._moved`.
        """
        path, view, meta, fs = moved
        return Chunk(data, path, view, None, meta, fs, None, ignore_chunk_properties=True)

    def transfer(self, nest: int = 0, position: int = 0) -> Chunk:
        """
        Returns the chunk that a unit in the same process receives when this chunk is passed on,
        moved by the given nesting and placed at the given position. The data is copied, and the
        metadata is serialized and read back as in a stream: The receiver gets values of its own,
        and a value that cannot be serialized raises the same error as in `Chunk.serialize`.
        """
        return Chunk._received(self, self._moved(nest, position))

    def __repr__(self) -> str:
        layer = '/'.join(str(p) if s else F'!{p}' for p, s in zip(self._path, self._view))
        layer = layer and '/' + layer
        return F'<chunk{layer}:{bytes(self)!r}>'

    def intersect(self, other: Chunk):
        """
        Removes all meta variables from this chunk whose value differs from those of the `other`
        inut chunk.
        """
        other_meta = other._meta
        meta = self._meta
        for key, value in list(meta.items()):
            if other_meta.get(key) != value:
                meta.discard(key)
        return self

    def __str__(self):
        try:
            return self.decode('UTF8')
        except UnicodeDecodeError:
            return self.hex()

    def __hash__(self):
        return hash(zlib.adler32(self))

    @overload
    def __getitem__(self, k: str) -> int | str | float | bytes | None:
        ...

    @overload
    def __getitem__(self, key: SupportsIndex, /) -> int:
        ...

    @overload
    def __getitem__(self, key: slice, /) -> bytearray:
        ...

    def __getitem__(self, k):
        if not isinstance(k, str):
            return bytearray.__getitem__(self, k)
        return self._meta.__getitem__(k)

    def __setitem__(self, bounds, value):
        if isinstance(bounds, str):
            self._meta[bounds] = value
        else:
            bytearray.__setitem__(self, bounds, value)

    def meta_var_int(self, key: str) -> int:
        value = self._meta.__getitem__(key)
        if not isinstance(value, int):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not int.')
        return value

    def meta_var_str(self, key: str) -> str:
        value = self._meta.__getitem__(key)
        if not isinstance(value, str):
            raise ValueError(F'The variable {key} was of type {type(value).__name__}, not str.')
        return value

    def truncate(self, scope: int = 0):
        """
        Truncate the `refinery.lib.frame.Chunk.path` and `refinery.lib.frame.Chunk.view` lists
        to the given length, setting the `refinery.lib.frame.Chunk.scope` to the given value.
        """
        del self._path[scope:]
        del self._view[scope:]
        return self

    def copy(self, meta=True, data=True) -> Chunk:
        """
        Produce a copy of this chunk. The metadata is copied if the `meta` argument is `True`,
        otherwise the copy has no metadata. The body of the chunk is copied only if the `data`
        argument is `True`.
        """
        data = data and self or None
        copy = Chunk(
            data,
            path=list(self._path),
            view=list(self._view),
            fill_scope=self._fill_scope,
            fill_batch=self._fill_batch,
            ignore_chunk_properties=True,
        )
        if meta:
            copy.meta.update(self.meta)
        if copy.meta.scope != copy.scope:
            raise RuntimeError
        return copy

    def __copy__(self):
        return self.copy()

    def __deepcopy__(self, memo):
        raise NotImplementedError

Methods

def set_next_scope(self, visible)

Expand source code Browse git

def set_next_scope(self, visible: bool) -> None:
    self._fill_scope = visible

def set_next_batch(self, batch)

This function allows units to emit trees of depth one rather than lists. When a unit emits a chunk at index a, sets the next batch to b, and when a double frame opens after this unit's invocation, then said chunk will have a/b added to its path. By default, b would always be 0. For example, the rex unit uses this feature. As a result:

$ emit #1yellow-#3red-#2orange | rex #(.)([a-z]+) {1} {2} [[| pop x:e ]| rep v:x ]]
yellow
red
red
red
orange
orange

The double frame after rex looks like this:

[[1,yellow],[2,red],[3,orange]]

By default, the frame would simply look like this:

[[1,yellow,2,red,3,orange]]

This feature is useful for Units that produce multiple outputs for each of a number of intermediate results - in the case of rex, that intermediate result is a regular expression match, and rex allows to produce different outputs for each of those.

Expand source code Browse git

def set_next_batch(self, batch: int) -> None:
    """
    This function allows units to emit trees of depth one rather than lists. When a unit emits
    a chunk at index `a`, sets the next batch to `b`, and when a double frame opens after this
    unit's invocation, then said chunk will have `a/b` added to its path. By default, `b` would
    always be `0`. For example, the `refinery.rex` unit uses this feature. As a result:

        $ emit #1yellow-#3red-#2orange | rex #(.)([a-z]+) {1} {2} [[| pop x:e ]| rep v:x ]]
        yellow
        red
        red
        red
        orange
        orange

    The double frame after `refinery.rex` looks like this:

        [[1,yellow],[2,red],[3,orange]]

    By default, the frame would simply look like this:

        [[1,yellow,2,red,3,orange]]

    This feature is useful for `refinery.units.Unit`s that produce multiple outputs for each of
    a number of intermediate results - in the case of `refinery.rex`, that intermediate result
    is a regular expression match, and `refinery.rex` allows to produce different outputs for
    each of those.
    """
    self._fill_batch = batch

def serialize(self, nest=0, position=0)

Returns the serialized form of this chunk, moved by the given nesting and placed at the given position, as a sequence of buffers to be written in order. Their concatenation is a msgpack array of the path, view, metadata, fill scope, and data. For a large chunk, the first buffer ends with the header of the data, and the chunk itself follows uncopied.

Expand source code Browse git

def serialize(self, nest: int = 0, position: int = 0) -> tuple[bytes | bytearray, ...]:
    """
    Returns the serialized form of this chunk, moved by the given nesting and placed at the
    given position, as a sequence of buffers to be written in order. Their concatenation is a
    msgpack array of the path, view, metadata, fill scope, and data. For a large chunk, the
    first buffer ends with the header of the data, and the chunk itself follows uncopied.
    """
    assert msgpack
    path, view, meta, fs = self._relocated(nest, position)
    if (size := len(self)) < _INLINE:
        packed = msgpack.packb((path, view, meta, fs, self))
        assert isinstance(packed, bytes)
        return packed,
    packed = msgpack.packb((path, view, meta, fs))
    assert isinstance(packed, bytes)
    head = bytearray(packed)
    head[0] += 1
    head.extend(_bin_header(size))
    return head, self

def pack(self, nest=0, position=0)

Returns the serialized form of this chunk as a single byte string; see Chunk.serialize().

Expand source code Browse git

def pack(self, nest: int = 0, position: int = 0) -> bytes:
    """
    Returns the serialized form of this chunk as a single byte string; see
    `refinery.lib.frame.Chunk.serialize`.
    """
    return B''.join(self.serialize(nest, position))

def transfer(self, nest=0, position=0)

Returns the chunk that a unit in the same process receives when this chunk is passed on, moved by the given nesting and placed at the given position. The data is copied, and the metadata is serialized and read back as in a stream: The receiver gets values of its own, and a value that cannot be serialized raises the same error as in Chunk.serialize().

Expand source code Browse git

def transfer(self, nest: int = 0, position: int = 0) -> Chunk:
    """
    Returns the chunk that a unit in the same process receives when this chunk is passed on,
    moved by the given nesting and placed at the given position. The data is copied, and the
    metadata is serialized and read back as in a stream: The receiver gets values of its own,
    and a value that cannot be serialized raises the same error as in `Chunk.serialize`.
    """
    return Chunk._received(self, self._moved(nest, position))

def intersect(self, other)

Removes all meta variables from this chunk whose value differs from those of the other inut chunk.

Expand source code Browse git

def intersect(self, other: Chunk):
    """
    Removes all meta variables from this chunk whose value differs from those of the `other`
    inut chunk.
    """
    other_meta = other._meta
    meta = self._meta
    for key, value in list(meta.items()):
        if other_meta.get(key) != value:
            meta.discard(key)
    return self

def meta_var_int(self, key)

Expand source code Browse git

def meta_var_int(self, key: str) -> int:
    value = self._meta.__getitem__(key)
    if not isinstance(value, int):
        raise ValueError(F'The variable {key} was of type {type(value).__name__}, not int.')
    return value

def meta_var_str(self, key)

Expand source code Browse git

def meta_var_str(self, key: str) -> str:
    value = self._meta.__getitem__(key)
    if not isinstance(value, str):
        raise ValueError(F'The variable {key} was of type {type(value).__name__}, not str.')
    return value

def truncate(self, scope=0)

Truncate the Chunk.path and Chunk.view lists to the given length, setting the Chunk.scope to the given value.

Expand source code Browse git

def truncate(self, scope: int = 0):
    """
    Truncate the `refinery.lib.frame.Chunk.path` and `refinery.lib.frame.Chunk.view` lists
    to the given length, setting the `refinery.lib.frame.Chunk.scope` to the given value.
    """
    del self._path[scope:]
    del self._view[scope:]
    return self

def copy(self, meta=True, data=True)

Produce a copy of this chunk. The metadata is copied if the meta argument is True, otherwise the copy has no metadata. The body of the chunk is copied only if the data argument is True.

Expand source code Browse git

def copy(self, meta=True, data=True) -> Chunk:
    """
    Produce a copy of this chunk. The metadata is copied if the `meta` argument is `True`,
    otherwise the copy has no metadata. The body of the chunk is copied only if the `data`
    argument is `True`.
    """
    data = data and self or None
    copy = Chunk(
        data,
        path=list(self._path),
        view=list(self._view),
        fill_scope=self._fill_scope,
        fill_batch=self._fill_batch,
        ignore_chunk_properties=True,
    )
    if meta:
        copy.meta.update(self.meta)
    if copy.meta.scope != copy.scope:
        raise RuntimeError
    return copy

class Framed (action, stream, filter, finish, nesting=0, squeeze=False, filter_all=False)

A proxy interface to ingest and output framed data. It is given an action to be performed for each elementary chunk of data, a stream of input data, and an integer argument nested which specifies the relative amount of nesting to be performed by the interface. This parameter should either be 1 if the interface should output the results at an additional layer, 0 if the nesting depth of the data should remain unchanged, and a negative amount if frame layers are to be collapsed. After initialization, the Framed object is an iterator over the output of the operation with all framing already taken care of: Framed output is a frame header followed by one refinery.lib.frame.Placement for each chunk, and unframed output consists of chunks.

Expand source code Browse git

class Framed:
    """
    A proxy interface to ingest and output framed data. It is given an `action` to be
    performed for each elementary chunk of data, a `stream` of input data, and an integer
    argument `nested` which specifies the relative amount of nesting to be performed
    by the interface. This parameter should either be `1` if the interface should output
    the results at an additional layer, `0` if the nesting depth of the data should
    remain unchanged, and a negative amount if frame layers are to be collapsed. After
    initialization, the `refinery.lib.frame.Framed` object is an iterator over the output of the
    operation with all framing already taken care of: Framed output is a frame header followed by
    one `refinery.lib.frame.Placement` for each chunk, and unframed output consists of chunks.
    """
    def __init__(
        self,
        action : Callable[[Chunk], Iterable[Chunk]],
        stream : BinaryIO | ChunkFeed | None,
        filter : Callable[[Iterable[Chunk]], Iterable[Chunk]],
        finish : Callable[[], Iterable[Chunk]],
        nesting: int = 0,
        squeeze: bool = False,
        filter_all: bool = False,
    ):
        self.unpack = FrameUnpacker(stream)
        self.action = action
        self.filter = filter
        self.finish = finish
        self.filter_all = filter_all
        self.nesting = nesting
        self.squeeze = squeeze

    def _apply_filter(self) -> Iterable[Chunk]:
        chunks = map(_indexed, itertools.count(), self.unpack)
        if self.unpack.scopable or self.filter_all:
            chunks = self.filter(chunks)
        yield from chunks

        if not self.unpack.eol:  # filter did not consume the iterable
            self.unpack.abort()

        if self.unpack.finished and self.finish:
            yield from self.finish()

    @property
    def unframed(self) -> bool:
        """
        This property is true if the output data is not framed.
        """
        return self.nesting + self.unpack.depth < 1

    @property
    def framebreak(self) -> bool:
        """
        This property will be true if the data generated by this framing interface should
        be separated by linebreaks. This happens when one of the following is true:
        - The requested nesting was smaller than required to close all existing frames.
        - The input data was not framed and the nesting did not increase in this unit
        """
        if not self.unpack.framed:
            return self.nesting < 1
        return self.nesting + self.unpack.depth < 0

    def _results(self, chunk: Chunk) -> Iterator[Chunk]:
        if chunk.visible:
            return self._generate_chunks(chunk)
        return HandOver(chunk)

    def _all_results(self) -> Iterator[Chunk]:
        while self.unpack.nextframe():
            yield from itertools.chain.from_iterable(map(self._results, self._apply_filter()))

    @staticmethod
    def _merged(pieces: Iterable[Chunk]) -> Iterator[Chunk]:
        """
        Concatenates the given chunks into at most one chunk, which keeps the metadata that all of
        them share. A single chunk is passed on as it is. Otherwise, the result is a new chunk,
        because the unit that produced the pieces may still use them, and the same chunk may occur
        more than once. The metadata of each piece is taken when the piece arrives, because a unit
        may label the same chunk again before passing it on once more.
        """
        it = iter(pieces)
        for first in it:
            break
        else:
            return HandOver(None)
        merged = first.copy(data=False)
        shared = True
        for piece in it:
            if first is not None:
                merged.extend(first)
                first = None
            if shared:
                shared = bool(merged.intersect(piece).meta)
            merged.extend(piece)
            del piece
        return HandOver(merged if first is None else first)

    def _generate_chunks(self, parent: Chunk):
        """
        Generates the outputs of the unit for the given input, placed where the input is and with
        its metadata. A squeezed output is merged once the unit has produced all of its pieces, so
        it inherits the metadata that the input had before the unit ran: The unit may have labelled
        its input in the meantime.
        """
        path = list(parent.path)
        view = list(parent.view)
        origin = parent.copy(data=False) if self.squeeze else parent
        meta = origin.meta
        scope = parent.scope

        def inherit(chunk: Chunk):
            if chunk is parent:
                return chunk
            if path:
                chunk._path[:] = path
            if view and not chunk._view:
                chunk._view[:] = view
            chunk._meta.inherit(meta)
            return chunk.truncate(scope)

        yield from map(inherit, self._generate_bytes(parent))

    def _generate_bytes(self, data: Chunk):
        try:
            outputs = self.action(data)
            if self.squeeze:
                outputs = self._merged(outputs)
            yield from outputs
        finally:
            data.temp = None

    def __iter__(self):
        nesting = self.nesting
        scope = max(self.unpack.depth + nesting, 0)
        if self.unpack.finished:
            if scope:
                yield generate_frame_header(scope)
            return
        repeat = itertools.repeat
        if nesting > 0:
            assert scope
            yield generate_frame_header(scope)
            nestings = repeat(nesting)
            while self.unpack.nextframe():
                for position, results in enumerate(map(self._results, self._apply_filter())):
                    yield from map(Placement, results, nestings, repeat(position))
        elif not self.unpack.framed:
            yield from itertools.chain.from_iterable(map(self._generate_bytes, self._apply_filter()))
        elif nesting == 0:
            assert scope
            yield generate_frame_header(scope)
            yield from map(Placement, self._all_results(), repeat(0), repeat(0))
        elif not scope:
            yield from map(Chunk.truncate, self._all_results())
        else:
            yield generate_frame_header(scope)
            check = scope + 1
            groups = itertools.groupby(self._all_results(), lambda result: result.path[:check])
            merged = itertools.chain.from_iterable(self._merged(group) for _, group in groups)
            yield from map(Placement, merged, repeat(nesting), repeat(0))

Instance variables

var unframed

This property is true if the output data is not framed.

Expand source code Browse git

@property
def unframed(self) -> bool:
    """
    This property is true if the output data is not framed.
    """
    return self.nesting + self.unpack.depth < 1

var framebreak

This property will be true if the data generated by this framing interface should be separated by linebreaks. This happens when one of the following is true: - The requested nesting was smaller than required to close all existing frames. - The input data was not framed and the nesting did not increase in this unit

Expand source code Browse git

@property
def framebreak(self) -> bool:
    """
    This property will be true if the data generated by this framing interface should
    be separated by linebreaks. This happens when one of the following is true:
    - The requested nesting was smaller than required to close all existing frames.
    - The input data was not framed and the nesting did not increase in this unit
    """
    if not self.unpack.framed:
        return self.nesting < 1
    return self.nesting + self.unpack.depth < 0

class FrameUnpacker (stream)

Provides a unified interface to read both framed and raw input data from a stream. After loading a framed input stream, the object provides an iterator over the first frame in the bottom layer of the frame tree. Consider this doubly layered frame tree:

[[FOO, BAR], [BOO, BAZ]]

The FrameUnpacker object will first be an iterator over the first frame [FOO, BAR]. After consuming this iterator, the FrameUnpacker.nextframe() method can be called to load the next frame, at which point the object will become an iterator over [BOO, BAZ].

Expand source code Browse git

class FrameUnpacker(Iterable[Chunk]):
    """
    Provides a unified interface to read both framed and raw input data from a stream. After
    loading a framed input stream, the object provides an iterator over the first **frame** in
    the bottom **layer** of the frame tree. Consider this doubly layered frame tree:

        [[FOO, BAR], [BOO, BAZ]]

    The `refinery.lib.frame.FrameUnpacker` object will first be an iterator over the first frame
    `[FOO, BAR]`. After consuming this iterator, the `refinery.lib.frame.FrameUnpacker.nextframe`
    method can be called to load the next frame, at which point the object will become an
    iterator over `[BOO, BAZ]`.
    """
    next_chunk: Chunk | None
    depth: int
    trunk: tuple[int, ...]
    check: tuple[int, ...]
    items: Iterator[buf] | None
    finished: bool
    framed: bool

    def __init__(self, stream: BinaryIO | ChunkFeed | None):
        self.finished = False
        self.trunk = ()
        self.check = ()
        self.items = None
        self.depth = 0
        self.next_chunk = None
        self.framed = False
        items = stream.items if isinstance(stream, ChunkFeed) else _read_stream(stream)
        first = next(items, None)
        if first is None:
            self.next_chunk = Chunk()
        elif isinstance(first, Chunk):
            probe = object()
            if _references(first) > _references(probe) and (second := next(items, None)) is not None:
                items = itertools.chain((second,), items)
                del second
            if _references(first) > _references(probe):
                chunk = Chunk(first, ignore_chunk_properties=True)
            else:
                _adopt(chunk := first)
            del first
            for item in items:
                chunk.extend(item)
                del item
            self.next_chunk = chunk
        elif len(first) != MSIZE or first[:len(MAGIC)] != MAGIC:
            raise ValueError('The input began with an object that is neither a chunk nor a frame header.')
        else:
            self.depth = first[~0]
            self.framed = True
            self.items = items
            self._advance()

    def _advance(self) -> bool:
        self.next_chunk = None
        if not self.finished and (items := self.items) is not None:
            if (chunk := next(items, None)) is not None:
                if not isinstance(chunk, Chunk):
                    raise ValueError('A frame contained an object that is not a chunk.')
                self.next_chunk = chunk
                self.check = tuple(chunk.path)
                if chunk.scope != self.depth:
                    raise RuntimeError(F'Frame of depth {self.depth} contained chunk of scope {chunk.scope}.')
                return True
        self.finished = True
        return False

    def nextframe(self) -> bool:
        """
        Once the iterator is consumed, calling this function will return `True` if
        and only if another frame with input data has been loaded, in which case
        the object will provide an iterator over the freshly loaded frame. If this
        function returns `False`, all input data has been consumed.
        """
        if self.finished:
            return False
        self.trunk = self.check
        return True

    def abort(self):
        """
        Abort unpacking chunks from the frame.
        """
        if self.depth > 1:
            while not self.finished and self.trunk == self.check:
                self._advance()
        else:
            self.items = None
            self.next_chunk = None
            self.finished = True

    @property
    def eol(self) -> bool:
        """
        Specifies whether the current frame was fully consumed.
        """
        return self.trunk != self.peek

    @property
    def scopable(self) -> bool:
        """
        Whether the chunks of the next frame can be made visible; see `Chunk.scopable`.
        """
        chunk = self.next_chunk
        assert chunk is not None
        return chunk.scopable

    @property
    def peek(self) -> tuple[int, ...]:
        """
        Contains the identifier of the next frame.
        """
        return self.check

    def __iter__(self) -> Generator[Chunk]:
        if self.finished:
            return
        if not self.framed:
            yield self.next_chunk
            self.next_chunk = None
            self.finished = True
            return
        while not self.finished and self.trunk == self.check:
            yield self.next_chunk
            self._advance()

Ancestors

collections.abc.Iterable
typing.Generic

Class variables

var next_chunk: The type of the None singleton.
var depth: The type of the None singleton.
var trunk: The type of the None singleton.
var check: The type of the None singleton.
var items: The type of the None singleton.
var finished: The type of the None singleton.
var framed: The type of the None singleton.

Instance variables

var eol

Specifies whether the current frame was fully consumed.

Expand source code Browse git

@property
def eol(self) -> bool:
    """
    Specifies whether the current frame was fully consumed.
    """
    return self.trunk != self.peek

var scopable

Whether the chunks of the next frame can be made visible; see Chunk.scopable.

Expand source code Browse git

@property
def scopable(self) -> bool:
    """
    Whether the chunks of the next frame can be made visible; see `Chunk.scopable`.
    """
    chunk = self.next_chunk
    assert chunk is not None
    return chunk.scopable

var peek

Contains the identifier of the next frame.

Expand source code Browse git

@property
def peek(self) -> tuple[int, ...]:
    """
    Contains the identifier of the next frame.
    """
    return self.check

Methods

def nextframe(self)

Once the iterator is consumed, calling this function will return True if and only if another frame with input data has been loaded, in which case the object will provide an iterator over the freshly loaded frame. If this function returns False, all input data has been consumed.

Expand source code Browse git

def nextframe(self) -> bool:
    """
    Once the iterator is consumed, calling this function will return `True` if
    and only if another frame with input data has been loaded, in which case
    the object will provide an iterator over the freshly loaded frame. If this
    function returns `False`, all input data has been consumed.
    """
    if self.finished:
        return False
    self.trunk = self.check
    return True

def abort(self)

Abort unpacking chunks from the frame.

Expand source code Browse git

def abort(self):
    """
    Abort unpacking chunks from the frame.
    """
    if self.depth > 1:
        while not self.finished and self.trunk == self.check:
            self._advance()
    else:
        self.items = None
        self.next_chunk = None
        self.finished = True