proxystore.stream.filters

proxystore/stream/filters.py

Set of common stream filters.

NullFilter

Filter which keeps all objects.

__call__

__call__(metadata: dict[str, Any]) -> bool

Apply the filter to event metadata.

Source code in proxystore/stream/filters.py

def __call__(self, metadata: dict[str, Any]) -> bool:
    """Apply the filter to event metadata."""
    return True

SamplingFilter

SamplingFilter(p: float)

Filter that randomly keeps objects.

Parameters:

• p (float) –

  Probability of the filter returning True. I.e., the probability that an object is kept.

Raises:

• ValueError –

  if p is not in the range [0, 1].

Source code in proxystore/stream/filters.py

def __init__(self, p: float) -> None:
    if p < 0 or p > 1:
        raise ValueError(
            f'Filter probability p must be in [0, 1]. Got p={p}.',
        )
    self._p = p

__call__

__call__(metadata: dict[str, Any]) -> bool

Apply the filter to event metadata.

Source code in proxystore/stream/filters.py

def __call__(self, metadata: dict[str, Any]) -> bool:
    """Apply the filter to event metadata."""
    return random.random() < self._p