-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathcheckpoint_manager.py
59 lines (46 loc) · 1.46 KB
/
checkpoint_manager.py
1
2
3
4
5
6
7
8
9
10
11
12
13
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
import pickle
import os
class CheckpointManager:
"""
Encapsulate checkpoint functionality using pickle
"""
def __init__(self, filename: str):
"""
Initialise with given filename
Args:
filename (str): Filename to use for saving/loading checkpoint
"""
self.filename = filename
def save(self, data):
"""
Save data to checkpoint file with pickle
Args:
data: Object to be pickled
"""
# mkdir if doesnt exist
directory = os.path.dirname(self.filename)
if directory and not os.path.exists(directory):
os.makedirs(directory, exist_ok=True)
# Save data
with open(self.filename, "wb") as f:
pickle.dump(data, f)
print(f"Checkpoint saved: {self.filename}")
def load(self):
"""
Load and return data from checkpoint file .pkl
Returns:
Unpickled data object
"""
if not os.path.exists(self.filename):
raise FileNotFoundError(f"Checkpoint file {self.filename} not found.")
with open(self.filename,"rb") as f:
data = pickle.load(f)
print(f"Checkpoint loaded: {self.filename}")
return data
def exists(self):
"""
Check if the checkpoint file exists
Returns:
(bool) True if checkpoint exists, else False.
"""
return os.path.exists(self.filename)