Source code for evorl.recorders.recorder

 1from abc import ABC, abstractmethod
 2from collections.abc import Mapping, Sequence
 3from typing import Any
 4
 5

[docs]
 6class Recorder(ABC):
 7    """A Recorder Interface."""
 8

[docs]
 9    @abstractmethod
10    def init(self) -> None:
11        """Initialize the recorder."""
12        raise NotImplementedError

13

[docs]
14    @abstractmethod
15    def write(self, data: Mapping[str, Any], step: int | None = None) -> None:
16        """Write data to the recorder."""
17        raise NotImplementedError

18

[docs]
19    @abstractmethod
20    def close(self) -> None:
21        """Finalize the recorder."""
22        raise NotImplementedError


23
24

[docs]
25class ChainRecorder(Recorder):
26    """Container for multiple recorders."""
27
28    def __init__(self, recorders: Sequence[Recorder]):
29        """Initialize the ChainRecorder.
30
31        Args:
32            recorders: A sequence of recorders to use.
33        """
34        self.recorders = recorders
35

[docs]
36    def add_recorder(self, recorder: Recorder) -> None:
37        self.recorders.append(recorder)

38

[docs]
39    def init(self) -> None:
40        for recorder in self.recorders:
41            recorder.init()

42

[docs]
43    def write(self, data: Mapping[str, Any], step: int | None = None) -> None:
44        for recorder in self.recorders:
45            recorder.write(data, step)

46

[docs]
47    def close(self) -> None:
48        for recorder in self.recorders:
49            recorder.close()