ding.utils.autolog.model

ding.utils.autolog.model

LoggedModel

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.

__properties property

Overview

Get all property names.

time property

Overview

Get original time object passed in, can execute method (such as step()) by this property.

Returns:

Name	Type	Description
BaseTime	_TimeObjectType	time object used by this model

expire property

Overview

Get expire time

Returns:

Type	Description
_TimeType	int or float: time that old value records expired

__init__(time_, expire)

Overview

Initialize the LoggedModel object using the given arguments.

Arguments: - time_ (:obj:BaseTime): The time. - expire (:obj:float): The expire time.

__get_property_ranged_data(name)

Overview

Get ranged data of one property.

Arguments: - name (:obj:str): The property name.

__init_properties()

Overview

Initialize all properties.

__get_range_values_func(name)

Overview

Get range_values function of one property.

Arguments: - name (:obj:str): The property name.

__register_default_funcs()

Overview

Register default functions.

fixed_time()

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:

Type	Description
Union[float, int]	int or float: fixed time

current_time()

Overview

Get current time (real time that regardless of time proxy's frozen statement)

Returns:

Type	Description
Union[float, int]	int or float: current time

freeze()

Overview

Freeze time proxy object. This feature can be useful when adding value replay feature (in the future)

unfreeze()

Overview

Unfreeze time proxy object. This feature can be useful when adding value replay feature (in the future)

register_attribute_value(attribute_name, property_name, value)

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

__getattr__(attribute_name)

Overview

Support all methods registered.

Parameters:

Name	Type	Description	Default
attribute_name	str	name of attribute	required

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

get_property_attribute(property_name)

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

Full Source Code

../ding/utils/autolog/model.py

1from abc import ABCMeta 2from typing import TypeVar, Union, List, Any 3 4from .base import _LOGGED_MODEL__PROPERTIES, _LOGGED_MODEL__PROPERTY_ATTR_PREFIX, _TimeType, TimeMode, \ 5 _LOGGED_VALUE__PROPERTY_NAME 6from .data import TimeRangedData 7from .time_ctl import BaseTime, TimeProxy 8from .value import LoggedValue 9 10_TimeObjectType = TypeVar('_TimeObjectType', bound=BaseTime) 11 12 13class _LoggedModelMeta(ABCMeta): 14 """ 15 Overview: 16 Metaclass of LoggedModel, used to find all LoggedValue properties and register them. 17 Interfaces: 18 ``__init__`` 19 """ 20 21 def __init__(cls, name: str, bases: tuple, namespace: dict): 22 23 super().__init__(name, bases, namespace) 24 25 _properties = [] 26 for k, v in namespace.items(): 27 if isinstance(v, LoggedValue): 28 setattr(v, _LOGGED_VALUE__PROPERTY_NAME, k) 29 _properties.append(k) 30 31 setattr(cls, _LOGGED_MODEL__PROPERTIES, _properties) 32 33 34class LoggedModel(metaclass=_LoggedModelMeta): 35 """ 36 Overview: 37 A model with timeline (integered time, such as 1st, 2nd, 3rd, can also be modeled as a kind 38 of self-defined discrete time, such as the implement of TickTime). Serveral values have association 39 with each other can be maintained together by using LoggedModel. 40 41 Example: 42 Define AvgList model like this 43 44 >>> from ding.utils.autolog import LoggedValue, LoggedModel 45 >>> class AvgList(LoggedModel): 46 >>> value = LoggedValue(float) 47 >>> __property_names = ['value'] 48 >>> 49 >>> def __init__(self, time_: BaseTime, expire: Union[int, float]): 50 >>> LoggedModel.__init__(self, time_, expire) 51 >>> # attention, original value must be set in __init__ function, or it will not 52 >>> # be activated, the timeline of this value will also be unexpectedly affected. 53 >>> self.value = 0.0 54 >>> self.__register() 55 >>> 56 >>> def __register(self): 57 >>> def __avg_func(prop_name: str) -> float: # function to calculate average value of properties 58 >>> records = self.range_values[prop_name]() 59 >>> (_start_time, _), _ = records[0] 60 >>> (_, _end_time), _ = records[-1] 61 >>> 62 >>> _duration = _end_time - _start_time 63 >>> _sum = sum([_value * (_end_time - _begin_time) for (_begin_time, _end_time), _value in records]) 64 >>> 65 >>> return _sum / _duration 66 >>> 67 >>> for _prop_name in self.__property_names: 68 >>> self.register_attribute_value('avg', _prop_name, partial(__avg_func, prop_name=_prop_name)) 69 70 Use it like this 71 72 >>> from ding.utils.autolog import NaturalTime, TimeMode 73 >>> 74 >>> if __name__ == "__main__": 75 >>> _time = NaturalTime() 76 >>> ll = AvgList(_time, expire=10) 77 >>> 78 >>> # just do something here ... 79 >>> 80 >>> print(ll.range_values['value']()) # original range_values function in LoggedModel of last 10 secs 81 >>> print(ll.range_values['value'](TimeMode.ABSOLUTE)) # use absolute time 82 >>> print(ll.avg['value']()) # average value of last 10 secs 83 84 Interfaces: 85 ``__init__``, ``time``, ``expire``, ``fixed_time``, ``current_time``, ``freeze``, ``unfreeze``, \ 86 ``register_attribute_value``, ``__getattr__``, ``get_property_attribute`` 87 88 Property: 89 - time (:obj:`BaseTime`): The time. 90 - expire (:obj:`float`): The expire time. 91 """ 92 93 def __init__(self, time_: _TimeObjectType, expire: _TimeType): 94 """ 95 Overview: 96 Initialize the LoggedModel object using the given arguments. 97 Arguments: 98 - time_ (:obj:`BaseTime`): The time. 99 - expire (:obj:`float`): The expire time. 100 """ 101 102 self.__time = time_ 103 self.__time_proxy = TimeProxy(self.__time, frozen=False) 104 self.__init_time = self.__time_proxy.time() 105 self.__expire = expire 106 107 self.__methods = {} 108 self.__prop2attr = {} # used to find registerd attributes list according to property name 109 110 self.__init_properties() 111 self.__register_default_funcs() 112 113 @property 114 def __properties(self) -> List[str]: 115 """ 116 Overview: 117 Get all property names. 118 """ 119 120 return getattr(self, _LOGGED_MODEL__PROPERTIES) 121 122 def __get_property_ranged_data(self, name: str) -> TimeRangedData: 123 """ 124 Overview: 125 Get ranged data of one property. 126 Arguments: 127 - name (:obj:`str`): The property name. 128 """ 129 130 return getattr(self, _LOGGED_MODEL__PROPERTY_ATTR_PREFIX + name) 131 132 def __init_properties(self): 133 """ 134 Overview: 135 Initialize all properties. 136 """ 137 138 for name in self.__properties: 139 setattr( 140 self, _LOGGED_MODEL__PROPERTY_ATTR_PREFIX + name, 141 TimeRangedData(self.__time_proxy, expire=self.__expire) 142 ) 143 144 def __get_range_values_func(self, name: str): 145 """ 146 Overview: 147 Get range_values function of one property. 148 Arguments: 149 - name (:obj:`str`): The property name. 150 """ 151 152 def _func(mode: TimeMode = TimeMode.RELATIVE_LIFECYCLE): 153 _current_time = self.__time_proxy.time() 154 _result = self.__get_property_ranged_data(name).history() 155 156 if mode == TimeMode.RELATIVE_LIFECYCLE: 157 _result = [(_time - self.__init_time, _data) for _time, _data in _result] 158 elif mode == TimeMode.RELATIVE_CURRENT_TIME: 159 _result = [(_time - _current_time, _data) for _time, _data in _result] 160 161 _ranges = [] 162 for i in range(0, len(_result) - 1): 163 _this_time, _this_data = _result[i] 164 _next_time, _next_data = _result[i + 1] 165 _ranges.append(((_this_time, _next_time), _this_data)) 166 167 return _ranges 168 169 return _func 170 171 def __register_default_funcs(self): 172 """ 173 Overview: 174 Register default functions. 175 """ 176 177 for name in self.__properties: 178 self.register_attribute_value('range_values', name, self.__get_range_values_func(name)) 179 180 @property 181 def time(self) -> _TimeObjectType: 182 """ 183 Overview: 184 Get original time object passed in, can execute method (such as step()) by this property. 185 186 Returns: 187 BaseTime: time object used by this model 188 """ 189 return self.__time 190 191 @property 192 def expire(self) -> _TimeType: 193 """ 194 Overview: 195 Get expire time 196 197 Returns: 198 int or float: time that old value records expired 199 """ 200 return self.__expire 201 202 def fixed_time(self) -> Union[float, int]: 203 """ 204 Overview: 205 Get fixed time (will be frozen time if time proxy is frozen) 206 This feature can be useful when adding value replay feature (in the future) 207 208 Returns: 209 int or float: fixed time 210 """ 211 return self.__time_proxy.time() 212 213 def current_time(self) -> Union[float, int]: 214 """ 215 Overview: 216 Get current time (real time that regardless of time proxy's frozen statement) 217 218 Returns: 219 int or float: current time 220 """ 221 return self.__time_proxy.current_time() 222 223 def freeze(self): 224 """ 225 Overview: 226 Freeze time proxy object. 227 This feature can be useful when adding value replay feature (in the future) 228 """ 229 self.__time_proxy.freeze() 230 231 def unfreeze(self): 232 """ 233 Overview: 234 Unfreeze time proxy object. 235 This feature can be useful when adding value replay feature (in the future) 236 """ 237 self.__time_proxy.unfreeze() 238 239 def register_attribute_value(self, attribute_name: str, property_name: str, value: Any): 240 """ 241 Overview: 242 Register a new attribute for one of the values. Example can be found in overview of class. 243 Arguments: 244 - attribute_name (:obj:`str`): name of attribute 245 - property_name (:obj:`str`): name of property 246 - value (:obj:`Any`): value of attribute 247 """ 248 self.__methods[attribute_name] = self.__methods.get(attribute_name, {}) 249 self.__methods[attribute_name][property_name] = value 250 if attribute_name == "range_values": 251 # "range_values" is not added to ``self.__prop2attr`` 252 return 253 self.__prop2attr[property_name] = self.__prop2attr.get(property_name, []) 254 self.__prop2attr[property_name].append(attribute_name) 255 256 def __getattr__(self, attribute_name: str) -> Any: 257 """ 258 Overview: 259 Support all methods registered. 260 261 Arguments: 262 attribute_name (str): name of attribute 263 264 Return: 265 A indelible object that can return attribute value. 266 267 Example: 268 >>> ll = AvgList(NaturalTime(), expire=10) 269 >>> ll.range_value['value'] # get 'range_value' attribute of 'value' property, it should be a function 270 """ 271 if attribute_name in self.__methods.keys(): 272 _attributes = self.__methods[attribute_name] 273 274 class _Cls: 275 276 def __getitem__(self, property_name: str): 277 if property_name in _attributes.keys(): 278 return _attributes[property_name] 279 else: 280 raise KeyError( 281 "Attribute {attr_name} for property {prop_name} not found.".format( 282 attr_name=repr(attribute_name), 283 prop_name=repr(property_name), 284 ) 285 ) 286 287 return _Cls() 288 else: 289 raise KeyError("Attribute {name} not found.".format(name=repr(attribute_name))) 290 291 def get_property_attribute(self, property_name: str) -> List[str]: 292 """ 293 Overview: 294 Find all registered attributes (except common "range_values" attribute, since "range_values" is not 295 added to ``self.__prop2attr``) of one given property. 296 Arguments: 297 - property_name (:obj:`str`): name of property to query attributes 298 Returns: 299 - attr_list (:obj:`List[str]`): the registered attributes list of the input property 300 """ 301 return self.__prop2attr[property_name]