Shortcuts

Source code for ding.utils.autolog.data

import pickle
from abc import abstractmethod, ABCMeta
from collections import deque
from threading import Lock
from typing import TypeVar, Iterable, List, Tuple, Union

from .time_ctl import BaseTime

_Tp = TypeVar('_Tp')


[docs]class RangedData(metaclass=ABCMeta): """ Overview: A data structure that can store data for a period of time. Interfaces: ``__init__``, ``append``, ``extend``, ``current``, ``history``, ``expire``, ``__bool__``, ``_get_time``. Properties: - expire (:obj:`float`): The expire time. """
[docs] def __init__(self, expire: float, use_pickle: bool = False): """ Overview: Initialize the RangedData object. Arguments: - expire (:obj:`float`): The expire time of the data. - use_pickle (:obj:`bool`): Whether to use pickle to serialize the data. """ self.__expire = expire self.__use_pickle = use_pickle self.__check_expire() self.__data_max_id = 0 self.__data_items = {} self.__data_lock = Lock() self.__last_item = None self.__queue = deque() self.__lock = Lock()
def __check_expire(self): """ Overview: Check the expire time. """ if isinstance(self.__expire, (int, float)): if self.__expire <= 0: raise ValueError( "Expire should be greater than 0, but {actual} found.".format(actual=repr(self.__expire)) ) else: raise TypeError( 'Expire should be int or float, but {actual} found.'.format(actual=type(self.__expire).__name__) ) def __registry_data_item(self, data: _Tp) -> int: """ Overview: Registry the data item. Arguments: - data (:obj:`_Tp`): The data item. """ with self.__data_lock: self.__data_max_id += 1 if self.__use_pickle: self.__data_items[self.__data_max_id] = pickle.dumps(data) else: self.__data_items[self.__data_max_id] = data return self.__data_max_id def __get_data_item(self, data_id: int) -> _Tp: """ Overview: Get the data item. Arguments: - data_id (:obj:`int`): The data id. """ with self.__data_lock: if self.__use_pickle: return pickle.loads(self.__data_items[data_id]) else: return self.__data_items[data_id] def __remove_data_item(self, data_id: int): """ Overview: Remove the data item. Arguments: - data_id (:obj:`int`): The data id. """ with self.__data_lock: del self.__data_items[data_id] def __check_time(self, time_: float): """ Overview: Check the time. Arguments: - time_ (:obj:`float`): The time. """ if self.__queue: _time, _ = self.__queue[-1] if time_ < _time: raise ValueError( "Time {time} invalid for descending from last time {last_time}".format( time=repr(time_), last_time=repr(_time) ) ) def __append_item(self, time_: float, data: _Tp): """ Overview: Append the data item. Arguments: - time_ (:obj:`float`): The time. - data (:obj:`_Tp`): The data item. """ self.__queue.append((time_, self.__registry_data_item(data))) def __flush_history(self): """ Overview: Flush the history data. """ _time = self._get_time() _limit_time = _time - self.__expire while self.__queue: _head_time, _head_id = self.__queue.popleft() if _head_time >= _limit_time: self.__queue.appendleft((_head_time, _head_id)) break else: if self.__last_item: _last_time, _last_id = self.__last_item self.__remove_data_item(_last_id) self.__last_item = (_head_time, _head_id) def __append(self, time_: float, data: _Tp): """ Overview: Append the data. """ self.__check_time(time_) self.__append_item(time_, data) self.__flush_history() def __current(self): """ Overview: Get the current data. """ if self.__queue: _tail_time, _tail_id = self.__queue.pop() self.__queue.append((_tail_time, _tail_id)) return self.__get_data_item(_tail_id) elif self.__last_item: _last_time, _last_id = self.__last_item return self.__get_data_item(_last_id) else: raise ValueError("This range is empty.") def __history_yield(self): """ Overview: Yield the history data. """ _time = self._get_time() _limit_time = _time - self.__expire _latest_time, _latest_id = None, None if self.__last_item: _latest_time, _latest_id = _last_time, _last_id = self.__last_item yield max(_last_time, _limit_time), self.__get_data_item(_last_id) for _item_time, _item_id in self.__queue: _latest_time, _latest_id = _item_time, _item_id yield _item_time, self.__get_data_item(_item_id) if _latest_time is not None and _latest_time < _time: yield _time, self.__get_data_item(_latest_id) def __history(self): """ Overview: Get the history data. """ return list(self.__history_yield())
[docs] def append(self, data: _Tp): """ Overview: Append the data. """ with self.__lock: self.__flush_history() _time = self._get_time() self.__append(_time, data) return self
[docs] def extend(self, iter_: Iterable[_Tp]): """ Overview: Extend the data. """ with self.__lock: self.__flush_history() _time = self._get_time() for item in iter_: self.__append(_time, item) return self
[docs] def current(self) -> _Tp: """ Overview: Get the current data. """ with self.__lock: self.__flush_history() return self.__current()
[docs] def history(self) -> List[Tuple[Union[int, float], _Tp]]: """ Overview: Get the history data. """ with self.__lock: self.__flush_history() return self.__history()
@property def expire(self) -> float: """ Overview: Get the expire time. """ with self.__lock: self.__flush_history() return self.__expire def __bool__(self): """ Overview: Check whether the range is empty. """ with self.__lock: self.__flush_history() return not not (self.__queue or self.__last_item)
[docs] @abstractmethod def _get_time(self) -> float: """ Overview: Get the current time. """ raise NotImplementedError
[docs]class TimeRangedData(RangedData): """ Overview: A data structure that can store data for a period of time. Interfaces: ``__init__``, ``_get_time``, ``append``, ``extend``, ``current``, ``history``, ``expire``, ``__bool__``. Properties: - time (:obj:`BaseTime`): The time. - expire (:obj:`float`): The expire time. """
[docs] def __init__(self, time_: BaseTime, expire: float): """ Overview: Initialize the TimeRangedData object. Arguments: - time_ (:obj:`BaseTime`): The time. - expire (:obj:`float`): The expire time. """ RangedData.__init__(self, expire) self.__time = time_
[docs] def _get_time(self) -> float: """ Overview: Get the current time. """ return self.__time.time()
@property def time(self): """ Overview: Get the time. """ return self.__time