File size: 2,384 Bytes
			
			| e636070 | 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 59 60 61 62 63 64 65 66 67 68 69 70 71 72 | 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)     
 | 
