Source code for faust.types.windows
import abc
from datetime import timezone
from typing import List, NamedTuple, Optional
from mode import Seconds
__all__ = ['WindowRange', 'WindowT']
[docs]class WindowRange(NamedTuple):
start: float
end: float
def WindowRange_from_start(start: float, size: float) -> WindowRange:
return WindowRange(start=start, end=start + size - 0.1)
[docs]class WindowT(abc.ABC):
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:
...