aegis_sim.recording.checkpointrecorder
Recorder that periodically saves full simulation checkpoints.
1"""Recorder that periodically saves full simulation checkpoints.""" 2 3import logging 4import pathlib 5 6from .recorder import Recorder 7from aegis_sim import variables 8from aegis_sim.parameterization import parametermanager 9from aegis_sim.utilities.funcs import skip 10 11 12class CheckpointRecorder(Recorder): 13 def __init__(self, odir: pathlib.Path): 14 self.odir = odir 15 self.checkpoint_path = odir / "checkpoint" 16 17 def write(self, population, eggs): 18 """Save a checkpoint if CHECKPOINT_RATE says so. 19 20 # OUTPUT SPECIFICATION 21 path: /checkpoint 22 filetype: pickle 23 category: log 24 description: Full simulation checkpoint for resuming. Overwritten each time. 25 trait granularity: N/A 26 time granularity: snapshot 27 frequency parameter: CHECKPOINT_RATE 28 structure: Binary python file (Checkpoint object). 29 """ 30 from aegis_sim.checkpoint import Checkpoint 31 from aegis_sim import submodels 32 from aegis_sim.recording import recordingmanager 33 34 if skip("CHECKPOINT_RATE"): 35 return 36 37 # Flush buffered recorders so on-disk files are consistent with the checkpoint 38 recordingmanager.popsizerecorder.flush_all() 39 recordingmanager.resourcerecorder.flush_all() 40 41 step = variables.steps 42 checkpoint = Checkpoint.capture(population, eggs, variables, submodels, parametermanager) 43 checkpoint.save(self.checkpoint_path) 44 logging.debug(f"Checkpoint recorded at step {step}.")
13class CheckpointRecorder(Recorder): 14 def __init__(self, odir: pathlib.Path): 15 self.odir = odir 16 self.checkpoint_path = odir / "checkpoint" 17 18 def write(self, population, eggs): 19 """Save a checkpoint if CHECKPOINT_RATE says so. 20 21 # OUTPUT SPECIFICATION 22 path: /checkpoint 23 filetype: pickle 24 category: log 25 description: Full simulation checkpoint for resuming. Overwritten each time. 26 trait granularity: N/A 27 time granularity: snapshot 28 frequency parameter: CHECKPOINT_RATE 29 structure: Binary python file (Checkpoint object). 30 """ 31 from aegis_sim.checkpoint import Checkpoint 32 from aegis_sim import submodels 33 from aegis_sim.recording import recordingmanager 34 35 if skip("CHECKPOINT_RATE"): 36 return 37 38 # Flush buffered recorders so on-disk files are consistent with the checkpoint 39 recordingmanager.popsizerecorder.flush_all() 40 recordingmanager.resourcerecorder.flush_all() 41 42 step = variables.steps 43 checkpoint = Checkpoint.capture(population, eggs, variables, submodels, parametermanager) 44 checkpoint.save(self.checkpoint_path) 45 logging.debug(f"Checkpoint recorded at step {step}.")
def
write(self, population, eggs):
18 def write(self, population, eggs): 19 """Save a checkpoint if CHECKPOINT_RATE says so. 20 21 # OUTPUT SPECIFICATION 22 path: /checkpoint 23 filetype: pickle 24 category: log 25 description: Full simulation checkpoint for resuming. Overwritten each time. 26 trait granularity: N/A 27 time granularity: snapshot 28 frequency parameter: CHECKPOINT_RATE 29 structure: Binary python file (Checkpoint object). 30 """ 31 from aegis_sim.checkpoint import Checkpoint 32 from aegis_sim import submodels 33 from aegis_sim.recording import recordingmanager 34 35 if skip("CHECKPOINT_RATE"): 36 return 37 38 # Flush buffered recorders so on-disk files are consistent with the checkpoint 39 recordingmanager.popsizerecorder.flush_all() 40 recordingmanager.resourcerecorder.flush_all() 41 42 step = variables.steps 43 checkpoint = Checkpoint.capture(population, eggs, variables, submodels, parametermanager) 44 checkpoint.save(self.checkpoint_path) 45 logging.debug(f"Checkpoint recorded at step {step}.")
Save a checkpoint if CHECKPOINT_RATE says so.
OUTPUT SPECIFICATION
path: /checkpoint filetype: pickle category: log description: Full simulation checkpoint for resuming. Overwritten each time. trait granularity: N/A time granularity: snapshot frequency parameter: CHECKPOINT_RATE structure: Binary python file (Checkpoint object).