Skip to content

recorder

GameRecordFileHandler #

Bases: RotatingFileHandler

Log handler specific to the game recorder, using record filenames with timestamps defined at the creation of the log file.

Source code in cogip/tools/server/recorder.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
class GameRecordFileHandler(RotatingFileHandler):
    """
    Log handler specific to the game recorder, using record filenames
    with timestamps defined at the creation of the log file.
    """

    def __init__(self, root_dir="/var/tmp/cogip", backupCount=20):
        """Class constructor"""
        self._root_dir = Path(root_dir).resolve()
        self._root_dir.mkdir(parents=True, exist_ok=True)
        super().__init__(
            filename=self.newBaseFilename(),
            mode="a",
            maxBytes=0,
            backupCount=backupCount,
            encoding=None,
            delay=False,
        )

    def newBaseFilename(self) -> str:
        """
        Create a log filename based on current time.
        """
        current_time = int(time.time())
        time_tuple = time.localtime(current_time)
        return f"{self._root_dir}/game-{time.strftime('%Y-%m-%d_%H-%M-%S', time_tuple)}.log"

    def doRollover(self):
        """
        Create a new record file and clean-up old files.
        """
        if self.stream:
            self.stream.close()
            self.stream = None

        current_record_file = Path(self.baseFilename)
        if current_record_file.exists() and current_record_file.stat().st_size == 0:
            current_record_file.unlink()

        self.baseFilename = self.newBaseFilename()

        if self.backupCount > 0:
            record_files = sorted(self._root_dir.glob("*.log"))
            if len(record_files) > self.backupCount:
                for _ in range(len(record_files) - self.backupCount):
                    record_files.pop(0).unlink(missing_ok=True)

        if not self.delay:
            self.stream = self._open()

__init__(root_dir='/var/tmp/cogip', backupCount=20) #

Class constructor

Source code in cogip/tools/server/recorder.py 
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(self, root_dir="/var/tmp/cogip", backupCount=20):
    """Class constructor"""
    self._root_dir = Path(root_dir).resolve()
    self._root_dir.mkdir(parents=True, exist_ok=True)
    super().__init__(
        filename=self.newBaseFilename(),
        mode="a",
        maxBytes=0,
        backupCount=backupCount,
        encoding=None,
        delay=False,
    )

doRollover() #

Create a new record file and clean-up old files.

Source code in cogip/tools/server/recorder.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def doRollover(self):
    """
    Create a new record file and clean-up old files.
    """
    if self.stream:
        self.stream.close()
        self.stream = None

    current_record_file = Path(self.baseFilename)
    if current_record_file.exists() and current_record_file.stat().st_size == 0:
        current_record_file.unlink()

    self.baseFilename = self.newBaseFilename()

    if self.backupCount > 0:
        record_files = sorted(self._root_dir.glob("*.log"))
        if len(record_files) > self.backupCount:
            for _ in range(len(record_files) - self.backupCount):
                record_files.pop(0).unlink(missing_ok=True)

    if not self.delay:
        self.stream = self._open()

newBaseFilename() #

Create a log filename based on current time.

Source code in cogip/tools/server/recorder.py 
33
34
35
36
37
38
39
def newBaseFilename(self) -> str:
    """
    Create a log filename based on current time.
    """
    current_time = int(time.time())
    time_tuple = time.localtime(current_time)
    return f"{self._root_dir}/game-{time.strftime('%Y-%m-%d_%H-%M-%S', time_tuple)}.log"

GameRecorder #

Source code in cogip/tools/server/recorder.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class GameRecorder(metaclass=Singleton):
    def __init__(self):
        self._game_recorder = logging.getLogger("GameRecorder")
        self._game_recorder.setLevel(logging.INFO)
        self._game_recorder.propagate = False
        self._loop = asyncio.get_running_loop()
        self._record_dir = os.environ.get("SERVER_RECORD_DIR", "/var/tmp/cogip")
        self._recording = False

        try:
            self._record_handler = GameRecordFileHandler(self._record_dir)
            self._game_recorder.addHandler(self._record_handler)
        except OSError:
            logger.warning(f"Failed to record games in directory '{self._record_dir}'")
            self._record_handler = None

    @property
    def recording(self) -> bool:
        return self._recording

    @recording.setter
    def recording(self, enabled: bool) -> None:
        self._recording = enabled

    def record(self, record: dict[str, Any]) -> None:
        """
        Write a record in the current record file.
        Do it only the the robot the game has started.
        """
        if self._record_handler and self._recording:
            self._game_recorder.info(json.dumps(record))

    async def async_record(self, record: dict[str, Any]) -> None:
        """
        Async version of [`record()`][cogip.tools.server.recorder.GameRecorder.record] .
        """
        await self._loop.run_in_executor(None, self.record, record)

    def do_rollover(self) -> None:
        """
        Switch to a new record file.
        """
        if self._record_handler:
            self._record_handler.doRollover()

    async def async_do_rollover(self) -> None:
        """
        Async version of [`record()`][cogip.tools.server.recorder.GameRecorder.do_rollover] .
        """
        await self._loop.run_in_executor(None, self._record_handler.doRollover)

async_do_rollover() async #

Async version of record() .

Source code in cogip/tools/server/recorder.py 
110
111
112
113
114
async def async_do_rollover(self) -> None:
    """
    Async version of [`record()`][cogip.tools.server.recorder.GameRecorder.do_rollover] .
    """
    await self._loop.run_in_executor(None, self._record_handler.doRollover)

async_record(record) async #

Async version of record() .

Source code in cogip/tools/server/recorder.py 
 97
 98
 99
100
101
async def async_record(self, record: dict[str, Any]) -> None:
    """
    Async version of [`record()`][cogip.tools.server.recorder.GameRecorder.record] .
    """
    await self._loop.run_in_executor(None, self.record, record)

do_rollover() #

Switch to a new record file.

Source code in cogip/tools/server/recorder.py 
103
104
105
106
107
108
def do_rollover(self) -> None:
    """
    Switch to a new record file.
    """
    if self._record_handler:
        self._record_handler.doRollover()

record(record) #

Write a record in the current record file. Do it only the the robot the game has started.

Source code in cogip/tools/server/recorder.py 
89
90
91
92
93
94
95
def record(self, record: dict[str, Any]) -> None:
    """
    Write a record in the current record file.
    Do it only the the robot the game has started.
    """
    if self._record_handler and self._recording:
        self._game_recorder.info(json.dumps(record))