Source code for faust.types.windows
"""Types related to windowing."""
import abc
from datetime import timezone
from typing import List, Optional, Tuple
from mode import Seconds
__all__ = ["WindowRange", "WindowT"]
WindowRange = Tuple[float, float]
def WindowRange_from_start(start: float, size: float) -> WindowRange:
"""Create new windowrange from start and size."""
end = start + size - 0.1
return (start, end)
[docs]class WindowT(abc.ABC):
"""Type class for windows."""
expires: Optional[float] = None
tz: Optional[timezone] = None
[docs] @abc.abstractmethod
def ranges(self, timestamp: float) -> List[WindowRange]: ...
[docs] @abc.abstractmethod
def stale(self, timestamp: float, latest_timestamp: float) -> bool: ...
[docs] @abc.abstractmethod
def current(self, timestamp: float) -> WindowRange: ...
[docs] @abc.abstractmethod
def earliest(self, timestamp: float) -> WindowRange: ...
[docs] @abc.abstractmethod
def delta(self, timestamp: float, d: Seconds) -> WindowRange: ...