from abc import ABCMeta from typing import TypeVar, Union, List, Any from .base import _LOGGED_MODEL__PROPERTIES, _LOGGED_MODEL__PROPERTY_ATTR_PREFIX, _TimeType, TimeMode, \ _LOGGED_VALUE__PROPERTY_NAME from .data import TimeRangedData from .time_ctl import BaseTime, TimeProxy from .value import LoggedValue _TimeObjectType = TypeVar('_TimeObjectType', bound=BaseTime) class _LoggedModelMeta(ABCMeta): """ Overview: Metaclass of LoggedModel, used to find all LoggedValue properties and register them. Interfaces: ``__init__`` """ def __init__(cls, name: str, bases: tuple, namespace: dict): super().__init__(name, bases, namespace) _properties = [] for k, v in namespace.items(): if isinstance(v, LoggedValue): setattr(v, _LOGGED_VALUE__PROPERTY_NAME, k) _properties.append(k) setattr(cls, _LOGGED_MODEL__PROPERTIES, _properties) class LoggedModel(metaclass=_LoggedModelMeta): """ Overview: A model with timeline (integered time, such as 1st, 2nd, 3rd, can also be modeled as a kind of self-defined discrete time, such as the implement of TickTime). Serveral values have association with each other can be maintained together by using LoggedModel. Example: Define AvgList model like this >>> from ding.utils.autolog import LoggedValue, LoggedModel >>> class AvgList(LoggedModel): >>> value = LoggedValue(float) >>> __property_names = ['value'] >>> >>> def __init__(self, time_: BaseTime, expire: Union[int, float]): >>> LoggedModel.__init__(self, time_, expire) >>> # attention, original value must be set in __init__ function, or it will not >>> # be activated, the timeline of this value will also be unexpectedly affected. >>> self.value = 0.0 >>> self.__register() >>> >>> def __register(self): >>> def __avg_func(prop_name: str) -> float: # function to calculate average value of properties >>> records = self.range_values[prop_name]() >>> (_start_time, _), _ = records[0] >>> (_, _end_time), _ = records[-1] >>> >>> _duration = _end_time - _start_time >>> _sum = sum([_value * (_end_time - _begin_time) for (_begin_time, _end_time), _value in records]) >>> >>> return _sum / _duration >>> >>> for _prop_name in self.__property_names: >>> self.register_attribute_value('avg', _prop_name, partial(__avg_func, prop_name=_prop_name)) Use it like this >>> from ding.utils.autolog import NaturalTime, TimeMode >>> >>> if __name__ == "__main__": >>> _time = NaturalTime() >>> ll = AvgList(_time, expire=10) >>> >>> # just do something here ... >>> >>> print(ll.range_values['value']()) # original range_values function in LoggedModel of last 10 secs >>> print(ll.range_values['value'](TimeMode.ABSOLUTE)) # use absolute time >>> print(ll.avg['value']()) # average value of last 10 secs Interfaces: ``__init__``, ``time``, ``expire``, ``fixed_time``, ``current_time``, ``freeze``, ``unfreeze``, \ ``register_attribute_value``, ``__getattr__``, ``get_property_attribute`` Property: - time (:obj:`BaseTime`): The time. - expire (:obj:`float`): The expire time. """ def __init__(self, time_: _TimeObjectType, expire: _TimeType): """ Overview: Initialize the LoggedModel object using the given arguments. Arguments: - time_ (:obj:`BaseTime`): The time. - expire (:obj:`float`): The expire time. """ self.__time = time_ self.__time_proxy = TimeProxy(self.__time, frozen=False) self.__init_time = self.__time_proxy.time() self.__expire = expire self.__methods = {} self.__prop2attr = {} # used to find registerd attributes list according to property name self.__init_properties() self.__register_default_funcs() @property def __properties(self) -> List[str]: """ Overview: Get all property names. """ return getattr(self, _LOGGED_MODEL__PROPERTIES) def __get_property_ranged_data(self, name: str) -> TimeRangedData: """ Overview: Get ranged data of one property. Arguments: - name (:obj:`str`): The property name. """ return getattr(self, _LOGGED_MODEL__PROPERTY_ATTR_PREFIX + name) def __init_properties(self): """ Overview: Initialize all properties. """ for name in self.__properties: setattr( self, _LOGGED_MODEL__PROPERTY_ATTR_PREFIX + name, TimeRangedData(self.__time_proxy, expire=self.__expire) ) def __get_range_values_func(self, name: str): """ Overview: Get range_values function of one property. Arguments: - name (:obj:`str`): The property name. """ def _func(mode: TimeMode = TimeMode.RELATIVE_LIFECYCLE): _current_time = self.__time_proxy.time() _result = self.__get_property_ranged_data(name).history() if mode == TimeMode.RELATIVE_LIFECYCLE: _result = [(_time - self.__init_time, _data) for _time, _data in _result] elif mode == TimeMode.RELATIVE_CURRENT_TIME: _result = [(_time - _current_time, _data) for _time, _data in _result] _ranges = [] for i in range(0, len(_result) - 1): _this_time, _this_data = _result[i] _next_time, _next_data = _result[i + 1] _ranges.append(((_this_time, _next_time), _this_data)) return _ranges return _func def __register_default_funcs(self): """ Overview: Register default functions. """ for name in self.__properties: self.register_attribute_value('range_values', name, self.__get_range_values_func(name)) @property def time(self) -> _TimeObjectType: """ Overview: Get original time object passed in, can execute method (such as step()) by this property. Returns: BaseTime: time object used by this model """ return self.__time @property def expire(self) -> _TimeType: """ Overview: Get expire time Returns: int or float: time that old value records expired """ return self.__expire def fixed_time(self) -> Union[float, int]: """ Overview: Get fixed time (will be frozen time if time proxy is frozen) This feature can be useful when adding value replay feature (in the future) Returns: int or float: fixed time """ return self.__time_proxy.time() def current_time(self) -> Union[float, int]: """ Overview: Get current time (real time that regardless of time proxy's frozen statement) Returns: int or float: current time """ return self.__time_proxy.current_time() def freeze(self): """ Overview: Freeze time proxy object. This feature can be useful when adding value replay feature (in the future) """ self.__time_proxy.freeze() def unfreeze(self): """ Overview: Unfreeze time proxy object. This feature can be useful when adding value replay feature (in the future) """ self.__time_proxy.unfreeze() def register_attribute_value(self, attribute_name: str, property_name: str, value: Any): """ Overview: Register a new attribute for one of the values. Example can be found in overview of class. Arguments: - attribute_name (:obj:`str`): name of attribute - property_name (:obj:`str`): name of property - value (:obj:`Any`): value of attribute """ self.__methods[attribute_name] = self.__methods.get(attribute_name, {}) self.__methods[attribute_name][property_name] = value if attribute_name == "range_values": # "range_values" is not added to ``self.__prop2attr`` return self.__prop2attr[property_name] = self.__prop2attr.get(property_name, []) self.__prop2attr[property_name].append(attribute_name) def __getattr__(self, attribute_name: str) -> Any: """ Overview: Support all methods registered. Arguments: attribute_name (str): name of attribute Return: A indelible object that can return attribute value. Example: >>> ll = AvgList(NaturalTime(), expire=10) >>> ll.range_value['value'] # get 'range_value' attribute of 'value' property, it should be a function """ if attribute_name in self.__methods.keys(): _attributes = self.__methods[attribute_name] class _Cls: def __getitem__(self, property_name: str): if property_name in _attributes.keys(): return _attributes[property_name] else: raise KeyError( "Attribute {attr_name} for property {prop_name} not found.".format( attr_name=repr(attribute_name), prop_name=repr(property_name), ) ) return _Cls() else: raise KeyError("Attribute {name} not found.".format(name=repr(attribute_name))) def get_property_attribute(self, property_name: str) -> List[str]: """ Overview: Find all registered attributes (except common "range_values" attribute, since "range_values" is not added to ``self.__prop2attr``) of one given property. Arguments: - property_name (:obj:`str`): name of property to query attributes Returns: - attr_list (:obj:`List[str]`): the registered attributes list of the input property """ return self.__prop2attr[property_name]