Source code for trinity.buffer.buffer_reader

"""Reader of the buffer."""
from abc import ABC, abstractmethod
from typing import Dict, List, Optional



[docs]
class BufferReader(ABC):
    """Interface of the buffer reader."""


[docs]
    @abstractmethod
    def read(self, batch_size: Optional[int] = None) -> List:
        """Read from buffer."""



[docs]
    @abstractmethod
    async def read_async(self, batch_size: Optional[int] = None) -> List:
        """Read from buffer asynchronously."""


    def __len__(self) -> int:
        """Get the number of samples in buffer."""
        raise NotImplementedError


[docs]
    def state_dict(self) -> Dict:
        return {}



[docs]
    def load_state_dict(self, state_dict: Dict) -> None:
        pass