Module refinery.units.meta
A package for units that operate primarily on frames of several of inputs.
Expand source code Browse git
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
A package for units that operate primarily on frames of several of inputs.
"""
from __future__ import annotations
from typing import Iterable, TYPE_CHECKING
from abc import abstractmethod
if TYPE_CHECKING:
from refinery.lib.frame import Chunk
from refinery.units import Arg, Unit
class FrameSlicer(Unit, abstract=True):
def __init__(self, *slice: Arg.Bounds(nargs='*', default=[slice(None, None)]), **keywords):
super().__init__(slice=list(slice), **keywords)
for s in self.args.slice:
if s.step and s.step < 0:
raise ValueError('negative slice steps are not supported here')
class ConditionalUnit(Unit, abstract=True):
"""
Conditional units can be used in two different ways. When a new frame opens after using this
unit, chunks that did not match the condition are moved out of scope for that frame but still
exist and will re-appear after the frame closes. When used inside a frame, however, the unit
works as a filter and will discard any chunks that do not match.
"""
def __init__(
self,
negate: Arg.Switch('-n', help='invert the logic of this filter; drop all matching chunks instead of keeping them') = False,
single: Arg.Switch('-s', help='discard all chunks after filtering a single one that matches the condition') = False,
**kwargs
):
super().__init__(negate=negate, single=single, **kwargs)
@abstractmethod
def match(self, chunk) -> bool:
...
def filter(self, chunks: Iterable[Chunk]):
single: bool = self.args.single
negate: bool = self.args.negate
nested: bool = self.args.nesting > 0 or self.args.squeeze
for chunk in chunks:
skipped = chunk.visible and self.match(chunk) is negate
if skipped:
if not nested:
continue
chunk.set_next_scope(False)
yield chunk
if single and not skipped:
break
Sub-modules
refinery.units.meta.chop
refinery.units.meta.cm
refinery.units.meta.cull
refinery.units.meta.dedup
refinery.units.meta.eat
refinery.units.meta.ef
refinery.units.meta.emit
-
A simple tool to output binary data. Multiple arguments are output in framed format, see
refinery.lib.frame
. refinery.units.meta.group
refinery.units.meta.groupby
refinery.units.meta.iff
refinery.units.meta.iffp
refinery.units.meta.iffs
refinery.units.meta.iffx
refinery.units.meta.max
refinery.units.meta.min
refinery.units.meta.mvg
refinery.units.meta.pad
refinery.units.meta.pick
refinery.units.meta.pop
refinery.units.meta.push
refinery.units.meta.put
refinery.units.meta.queue
-
A simple tool to queue binary data as one or more chunks in the current frame.
refinery.units.meta.reduce
refinery.units.meta.rmv
refinery.units.meta.scope
refinery.units.meta.sep
refinery.units.meta.sorted
refinery.units.meta.swap
refinery.units.meta.transpose
refinery.units.meta.xfcc
Classes
class FrameSlicer (*slice, **keywords)
-
Expand source code Browse git
class FrameSlicer(Unit, abstract=True): def __init__(self, *slice: Arg.Bounds(nargs='*', default=[slice(None, None)]), **keywords): super().__init__(slice=list(slice), **keywords) for s in self.args.slice: if s.step and s.step < 0: raise ValueError('negative slice steps are not supported here')
Ancestors
Subclasses
Class variables
var required_dependencies
var optional_dependencies
Inherited members
class ConditionalUnit (negate=False, single=False, **kwargs)
-
Conditional units can be used in two different ways. When a new frame opens after using this unit, chunks that did not match the condition are moved out of scope for that frame but still exist and will re-appear after the frame closes. When used inside a frame, however, the unit works as a filter and will discard any chunks that do not match.
Expand source code Browse git
class ConditionalUnit(Unit, abstract=True): """ Conditional units can be used in two different ways. When a new frame opens after using this unit, chunks that did not match the condition are moved out of scope for that frame but still exist and will re-appear after the frame closes. When used inside a frame, however, the unit works as a filter and will discard any chunks that do not match. """ def __init__( self, negate: Arg.Switch('-n', help='invert the logic of this filter; drop all matching chunks instead of keeping them') = False, single: Arg.Switch('-s', help='discard all chunks after filtering a single one that matches the condition') = False, **kwargs ): super().__init__(negate=negate, single=single, **kwargs) @abstractmethod def match(self, chunk) -> bool: ... def filter(self, chunks: Iterable[Chunk]): single: bool = self.args.single negate: bool = self.args.negate nested: bool = self.args.nesting > 0 or self.args.squeeze for chunk in chunks: skipped = chunk.visible and self.match(chunk) is negate if skipped: if not nested: continue chunk.set_next_scope(False) yield chunk if single and not skipped: break
Ancestors
Subclasses
Class variables
var required_dependencies
var optional_dependencies
Methods
def match(self, chunk)
-
Expand source code Browse git
@abstractmethod def match(self, chunk) -> bool: ...
Inherited members