| from typing import Any, Dict, List, Optional, Literal | |
| from collections import defaultdict | |
| from datetime import datetime | |
| from bw_utils import load_json_file, save_json_file, load_jsonl_file, save_jsonl_file | |
| import os | |
| class HistoryManager: | |
| def __init__(self): | |
| self.detailed_history = [] | |
| def add_record(self, record): | |
| """添加一个事件记录 | |
| record = { | |
| "cur_round":cur_round, | |
| "role_code":role_code, | |
| "detail":detail, | |
| "type":act_type, | |
| "initiator":initiator, | |
| "actor":actor | |
| "group":group, | |
| "other_info":other_info, | |
| "record_id":record_id | |
| } | |
| """ | |
| self.detailed_history.append(record) | |
| def modify_record(self, record_id: str, detail: str): | |
| """修改特定记录""" | |
| for record in self.detailed_history: | |
| if record["record_id"] == record_id: | |
| record["detail"] = detail | |
| print(f"Record {record_id} has been modified.") | |
| return record['group'] | |
| def search_record_detail(self, record_id: str): | |
| for record in self.detailed_history[::-1]: | |
| if record["record_id"] == record_id: | |
| return record["detail"] | |
| return None | |
| def get_recent_history(self, recent_k = 5): | |
| return [record["detail"] for record in self.detailed_history[-recent_k:]] | |
| def get_subsequent_history(self,start_idx): | |
| return [record["detail"] for record in self.detailed_history[start_idx:]] | |
| def get_complete_history(self,): | |
| return [record["detail"] for record in self.detailed_history[:]] | |
| def __len__(self): | |
| return len(self.detailed_history) | |
| def __getstate__(self): | |
| states = {key: value for key, value in self.__dict__.items() \ | |
| if isinstance(value, (str, int, list, dict, bool, type(None)))} | |
| return states | |
| def __setstate__(self, states): | |
| self.__dict__.update(states) | |
| def save_to_file(self, root_dir): | |
| filename = os.path.join(root_dir, f"./simulation_history.json") | |
| save_json_file(filename, self.__getstate__() ) | |
| def load_from_file(self, root_dir): | |
| filename = os.path.join(root_dir, f"./simulation_history.json") | |
| states = load_json_file(filename) | |
| self.__setstate__(states) | |