| """ |
| The atomic unit of a memory is an event. |
| A collection of events form an EventCollection. |
| """ |
|
|
| import json |
| from typing import List, Tuple, Union |
| from .common import Question, ParsedAnswer |
| from bson import ObjectId |
| import datetime |
| import time |
|
|
| |
| def file_location_type(path:str) -> str: |
| """ given a type, it returns a string indicating whether this is `local`, or `url` |
| """ |
| if path.startswith('http://') or path.startswith('https://'): |
| return 'url' |
| return 'local' |
|
|
| def read_event_file(file_location:str) -> dict: |
| """ Event files loaded as json |
| """ |
| with open(file_location, 'r') as f: |
| values = json.load(f) |
| return values |
|
|
| def write_event_file(file_location:str, values: dict): |
| """ Event files stored as json. |
| """ |
| prepare_dir_of(file_location) |
| if file_location_type(file_location) == 'local': |
| with open(file_location, 'w') as f: |
| json.dump(values,f) |
| else: |
| raise NotImplementedError(f'Uncaught file location type {fl_type}') |
| return file_location |
|
|
|
|
| class EventCollection(object): |
| """ Class used for data processing from event files |
| """ |
| def __init__(self): |
| self.id = str(ObjectId()) |
| self.events = [] |
|
|
| def load_from_session(self, event_dir:str, session_token:str) -> "EventCollection": |
| if file_location_type (event_dir) == 'local': |
| assert os.path.exists(os.path.join(event_dir, session_token)), f'{os.path.join(event_dir, session_token)} does not exist' |
| |
| for event_file in os.listdir(os.path.join(event_dir, session_token)): |
| |
| new_event = Event().load_from_event_file(os.path.join(event_dir, session_token, event_file)) |
| |
| |
| |
| self.add_event(new_event) |
| else: |
| raise NotImplementedError(f'Uncaught file location {file_location_type(event_dir)}') |
|
|
| self.time_sorted() |
| return self |
|
|
| def add_event(self, event:"Event") -> "EventCollection": |
| assert isinstance(event, Event), f"invalid type {type(event)} for EventCollection" |
| self.events.append(event) |
| return self |
|
|
| def time_sorted(self) -> "EventCollection": |
| self.events = sorted(self.events) |
| return self |
|
|
| def __str__(self) -> str: |
| printout = "" |
| for ev in sorted(self.events): |
| printout += str(ev)+'\n' |
| return printout |
|
|
| def __len__(self) -> int: |
| return len(self.events) |
|
|
| def filter_to(self, label:Union[str, list[str]]): |
| if isinstance(label, str): |
| label = [label] |
| |
| permissible_classes = [] |
| for l in label: |
| assert l in TYPE2CLASS, f"unknown type {label}" |
| permissible_classes.append(TYPE2CLASS[l]) |
| |
| filtered_events = [ev for ev in self.events if isinstance(ev, tuple(permissible_classes))] |
| return filtered_events |
|
|
|
|
| class Event(object): |
| def __init__(self, session_token:str = None): |
| self.session_token = session_token |
| self.timestamp = str(datetime.datetime.now()) |
| self.type = 'EVENT' |
|
|
| |
| def __lt__(self, obj: "Event") -> bool: |
| if not isinstance(obj, Event): raise ValueError(f'Invalid comparison of type {type(obj)}') |
| return self.timestamp < obj.timestamp |
| def __gt__(self, obj: "Event") -> bool: |
| if not isinstance(obj, Event): raise ValueError(f'Invalid comparison of type {type(obj)}') |
| return self.timestamp > obj.timestamp |
| def __le__(self, obj: "Event") -> bool: |
| if not isinstance(obj, Event): raise ValueError(f'Invalid comparison of type {type(obj)}') |
| return self.timestamp <= obj.timestamp |
| def __ge__(self, obj: "Event") -> bool: |
| if not isinstance(obj, Event): raise ValueError(f'Invalid comparison of type {type(obj)}') |
| return self.timestamp >= obj.timestamp |
| def __eq__(self, obj: "Event") -> bool: |
| if not isinstance(obj, Event): raise ValueError(f'Invalid comparison of type {type(obj)}') |
| return self.timestamp == obj.timestamp |
|
|
| def load_from_event_params(self, **kwargs) -> "Event": |
| for key in kwargs: |
| setattr(self, key, kwargs[key]) |
| return self |
|
|
| def load_from_event_file(self, filepath: str) -> "Event": |
| fields = read_event_file(filepath) |
| return TYPE2CLASS[fields['type']]().load_from_event_params(**fields) |
|
|
| def export(self) -> dict: |
| return vars(self) |
|
|
| def save_to_event_file(self, filepath: str) -> str: |
| attr_dict = self.export() |
| if hasattr(self, 'latency') and self.latency is None: |
| raise ValueError("Forgot to call .tick and .tock?") |
| write_event_file(filepath, attr_dict) |
| return filepath |
|
|
| @property |
| def description(self) -> str: |
| raise NotImplementedError |
|
|
| def __str__(self) -> str: |
| return f'{str(self.timestamp)}: [{self.type}]\n{self.description}' |
|
|
|
|
| class ThinkEvent(Event): |
| def __init__(self, session_token:str = None, qa_sequence:List[Tuple[Question, ParsedAnswer]]=None): |
| super().__init__(session_token) |
| assert isinstance(qa_sequence, list), "qa_sequence should be least, even if len 1." |
| assert all([isinstance(el, tuple) for el in qa_sequence]), "Elements of qa_sequence should be tuple." |
|
|
| self.qa_sequence = qa_sequence |
|
|
| @property |
| def description(self) -> str: |
| st = "" |
| for idx, qa in enumerate(qa_sequence): |
| q, a = qa |
| st += f"#### Question {idx}\n" |
| st += str(q) |
| st += "\n" |
| st += f"#### Answer{idx}\n" |
| st += str(a) |
| st += "\n" |
| return st |
| |
| class ActEvent(Event): |
| def __init__(self, session_token:str = None): |
| super().__init__(session_token) |
|
|
| @property |
| def description(self) -> str: |
| raise NotImplementedError |
|
|
| class ActErrorEvent(Event): |
| """ Event of an error in the execution of an action. |
| """ |
| def __init__(self, session_token:str = None, |
| exception:Exception=None): |
| super().__init__(session_token) |
| self.exception = exception |
|
|
| @property |
| def description(self) -> str: |
| raise NotImplementedError |
|
|
| |
| class ObserveEvent(Event): |
| def __init__(self, session_token:str = None): |
| super().__init__(session_token) |
| |
| @property |
| def description(self) -> str: |
| raise NotImplementedError |
|
|
| class EvaluateEvent(Event): |
| def __init__(self, session_token:str = None, |
| completion_question:Question = None, |
| completion_eval:ParsedAnswer = None): |
| super().__init__(session_token) |
| assert isinstance(completion_question, Question) |
| assert isinstance(completion_eval, ParsedAnswer) |
| self.completion_question = completion_question |
| self.completion_eval = completion_eval |
| |
| @property |
| def description(self) -> str: |
| st = "## Evaluation Question\n" |
| st += "\n" |
| st += str(self.completion_question) |
| st += "\n" |
| st += "## Evaluation Answer\n" |
| st += "\n" |
| st += str(self.completion_eval) |
| st += "\n" |
| return st |
|
|
| class FeedbackEvent(Event): |
| def __init__(self, session_token:str = None, |
| feedback:Union[None, Question]=None): |
| super().__init__(session_token) |
| assert isinstance(feedback, Question) or feedback is None |
| self.feedback = feedback |
|
|
| @property |
| def description(self) -> str: |
| st = "## Reflection:\n" |
| st += str(self.feedback) |
| st += "\n" |
| return st |
|
|
|
|
| class InteractEvent(Event): |
| def __init__(self, session_token:str = None): |
| super().__init__(session_token) |
|
|
| @property |
| def description(self) -> str: |
| raise NotImplementedError |
|
|
|
|
| TYPE2CLASS = { |
| 'EVENT': Event, |
| |
| 'THINK': ThinkEvent, |
| 'ACT': ActEvent, |
| 'ACTERROR': ActErrorEvent, |
| 'OBSERVE': ObserveEvent, |
| 'EVALUATE': EvaluateEvent, |
| 'FEEDBACK': FeedbackEvent, |
| 'INTERACT': InteractEvent, |
| } |
