from collections import namedtuple from core.global_symbols import NotInit from core.sheerka.services.sheerka_service import BaseService from sdp.sheerkaDataProvider import Event hist = namedtuple("HistoryTest", "text status") # tests purposes only class History: def __init__(self, event: Event, result, extra_info): self.event = event self.result = result self.extra_info = extra_info self._status = NotInit self._is_state_modified = NotInit self._format_instructions = None def __str__(self): if self.is_state_modified is NotInit: state_modified_str = "[?]" else: state_modified_str = "[X]" if self.is_state_modified else "[ ]" event_date = self.event.date.strftime('%d/%m/%Y %H:%M:%S') msg = f"{self.event.get_digest()} {event_date} {state_modified_str} : {self.event.message}" status = self.status if status is not None: msg += f" => {status}" return msg def __repr__(self): return f"History(event={self.event!r}, status={self.status}, result={self.result})" def __eq__(self, other): if id(self) == id(other): return True if isinstance(other, hist): return self.event.message == other.text and self.status == other.status if not isinstance(other, History): return False return self.event == other.event and self.result == other.result def __hash__(self): return hash((self.event, self.result, self.status)) @property def status(self): if self._status is not NotInit: return self._status if self.extra_info and "status" in self.extra_info: self._status = self.extra_info["status"] return self._status self._status = self.result.get_status() if self.result else None return self._status @property def is_state_modified(self): if self._is_state_modified is not NotInit: return self._is_state_modified if self.extra_info and "is_state_modified" in self.extra_info: self._is_state_modified = self.extra_info["is_state_modified"] return self._is_state_modified self._is_state_modified = self.result.is_state_modified() if self.result else None return self._is_state_modified def get_format_instructions(self): return self._format_instructions def set_format_instructions(self, instructions): self._format_instructions = instructions class SheerkaHistoryManager(BaseService): NAME = "History" def __init__(self, sheerka): super().__init__(sheerka) def history(self, depth=10, start=0): """ Load history :param depth: number of items :param start: :return: """ events = list(self.sheerka.om.current_sdp().load_events(depth, start)) for event in events: try: result = self.sheerka.om.current_sdp().load_result(event.get_digest()) except (IOError, KeyError): result = None try: extra_info = self.sheerka.om.current_sdp().load_result_extra_info(event.get_digest()) except (IOError, KeyError): extra_info = None yield History(event, result, extra_info)