ding.utils.autolog.time_ctl

ding.utils.autolog.time_ctl

BaseTime

Overview

Abstract time interface

Interfaces: time

time() abstractmethod

Overview

Get time information

Returns:

Type	Description
Union[int, float]	time(:obj:float, int): time information

NaturalTime

Bases: BaseTime

Overview

Natural time object

Interfaces: __init__, time Example: >>> from ding.utils.autolog.time_ctl import NaturalTime >>> time_ = NaturalTime()

time()

Overview

Get current natural time (float format, unix timestamp)

Returns:

Type	Description
float	time(:obj:float): unix timestamp

Example

from ding.utils.autolog.time_ctl import NaturalTime time_ = NaturalTime() time_.time() 1603896383.8811457

TickTime

Bases: BaseTime

Overview

Tick time object

Interfaces: __init__, step, time Example: >>> from ding.utils.autolog.time_ctl import TickTime >>> time_ = TickTime()

__init__(init=0)

Overview

Constructor of TickTime

Parameters:

Name	Type	Description	Default
- init (		obj:int): initial time, default is 0	required

step(delta=1)

Overview Step the time forward for this TickTime

Parameters:

Name	Type	Description	Default
- delta (		obj:int): steps to step forward, default is 1	required

Returns:

Type	Description
int	time (:obj:int): new time after stepping

Example

from ding.utils.autolog.time_ctl import TickTime time_ = TickTime(0) time_.step() 1 time_.step(2) 3

time()

Overview Get current tick time

Returns:

Name	Type	Description
int	int	current tick time

Example

from ding.utils.autolog.time_ctl import TickTime time_ = TickTime(0) time_.step() time_.time() 1

TimeProxy

Bases: BaseTime

Overview

Proxy of time object, it can freeze time, sometimes useful when reproducing. This object is thread-safe, and also freeze and unfreeze operation is strictly ordered.

Interfaces: __init__, freeze, unfreeze, time, current_time Examples: >>> from ding.utils.autolog.time_ctl import TickTime, TimeProxy >>> tick_time_ = TickTime() >>> time_ = TimeProxy(tick_time_) >>> tick_time_.step() >>> print(tick_time_.time(), time_.time(), time_.current_time()) 1 1 1 >>> time_.freeze() >>> tick_time_.step() >>> print(tick_time_.time(), time_.time(), time_.current_time()) 2 1 2 >>> time_.unfreeze() >>> print(tick_time_.time(), time_.time(), time_.current_time()) 2 2 2

is_frozen property

Overview

Get if this time proxy object is frozen

Returns:

Name	Type	Description
bool	bool	true if it is frozen, otherwise false

__init__(time_, frozen=False, lock_type=LockContextType.THREAD_LOCK)

Overview

Constructor for Time proxy

Parameters:

Name	Type	Description	Default
- time_ (		obj:BaseTime): another time object it based on	required
- frozen (		obj:bool): this object will be frozen immediately if true, otherwise not, default is False	required
- lock_type (		obj:LockContextType): type of the lock, default is THREAD_LOCK	required

freeze()

Overview

Freeze this time proxy

unfreeze()

Overview

Unfreeze this time proxy

time()

Overview

Get time (may be frozen time)

Returns:

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

current_time()

Overview

Get current time (will not be frozen time)

Returns:

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

Full Source Code

../ding/utils/autolog/time_ctl.py

1import time 2from abc import ABCMeta, abstractmethod 3from typing import Union 4 5from ..lock_helper import LockContext, LockContextType 6 7 8class BaseTime(metaclass=ABCMeta): 9 """ 10 Overview: 11 Abstract time interface 12 Interfaces: 13 ``time`` 14 """ 15 16 @abstractmethod 17 def time(self) -> Union[int, float]: 18 """ 19 Overview: 20 Get time information 21 22 Returns: 23 - time(:obj:`float, int`): time information 24 """ 25 raise NotImplementedError 26 27 28class NaturalTime(BaseTime): 29 """ 30 Overview: 31 Natural time object 32 Interfaces: 33 ``__init__``, ``time`` 34 Example: 35 >>> from ding.utils.autolog.time_ctl import NaturalTime 36 >>> time_ = NaturalTime() 37 """ 38 39 def __init__(self): 40 self.__last_time = None 41 42 def time(self) -> float: 43 """ 44 Overview: 45 Get current natural time (float format, unix timestamp) 46 47 Returns: 48 - time(:obj:`float`): unix timestamp 49 50 Example: 51 >>> from ding.utils.autolog.time_ctl import NaturalTime 52 >>> time_ = NaturalTime() 53 >>> time_.time() 54 1603896383.8811457 55 """ 56 _current_time = time.time() 57 if self.__last_time is not None: 58 _current_time = max(_current_time, self.__last_time) 59 60 self.__last_time = _current_time 61 return _current_time 62 63 64class TickTime(BaseTime): 65 """ 66 Overview: 67 Tick time object 68 Interfaces: 69 ``__init__``, ``step``, ``time`` 70 Example: 71 >>> from ding.utils.autolog.time_ctl import TickTime 72 >>> time_ = TickTime() 73 """ 74 75 def __init__(self, init: int = 0): 76 """ 77 Overview: 78 Constructor of TickTime 79 80 Arguments: 81 - init (:obj:`int`): initial time, default is 0 82 """ 83 self.__tick_time = init 84 85 def step(self, delta: int = 1) -> int: 86 """ 87 Overview 88 Step the time forward for this TickTime 89 90 Arguments: 91 - delta (:obj:`int`): steps to step forward, default is 1 92 93 Returns: 94 - time (:obj:`int`): new time after stepping 95 96 Example: 97 >>> from ding.utils.autolog.time_ctl import TickTime 98 >>> time_ = TickTime(0) 99 >>> time_.step() 100 1 101 >>> time_.step(2) 102 3 103 """ 104 if not isinstance(delta, int): 105 raise TypeError("Delta should be positive int, but {actual} found.".format(actual=type(delta).__name__)) 106 elif delta < 1: 107 raise ValueError("Delta should be no less than 1, but {actual} found.".format(actual=repr(delta))) 108 else: 109 self.__tick_time += delta 110 return self.__tick_time 111 112 def time(self) -> int: 113 """ 114 Overview 115 Get current tick time 116 117 Returns: 118 int: current tick time 119 120 Example: 121 >>> from ding.utils.autolog.time_ctl import TickTime 122 >>> time_ = TickTime(0) 123 >>> time_.step() 124 >>> time_.time() 125 1 126 """ 127 return self.__tick_time 128 129 130class TimeProxy(BaseTime): 131 """ 132 Overview: 133 Proxy of time object, it can freeze time, sometimes useful when reproducing. 134 This object is thread-safe, and also freeze and unfreeze operation is strictly ordered. 135 Interfaces: 136 ``__init__``, ``freeze``, ``unfreeze``, ``time``, ``current_time`` 137 Examples: 138 >>> from ding.utils.autolog.time_ctl import TickTime, TimeProxy 139 >>> tick_time_ = TickTime() 140 >>> time_ = TimeProxy(tick_time_) 141 >>> tick_time_.step() 142 >>> print(tick_time_.time(), time_.time(), time_.current_time()) 143 1 1 1 144 >>> time_.freeze() 145 >>> tick_time_.step() 146 >>> print(tick_time_.time(), time_.time(), time_.current_time()) 147 2 1 2 148 >>> time_.unfreeze() 149 >>> print(tick_time_.time(), time_.time(), time_.current_time()) 150 2 2 2 151 """ 152 153 def __init__(self, time_: BaseTime, frozen: bool = False, lock_type: LockContextType = LockContextType.THREAD_LOCK): 154 """ 155 Overview: 156 Constructor for Time proxy 157 158 Arguments: 159 - time_ (:obj:`BaseTime`): another time object it based on 160 - frozen (:obj:`bool`): this object will be frozen immediately if true, otherwise not, default is False 161 - lock_type (:obj:`LockContextType`): type of the lock, default is THREAD_LOCK 162 """ 163 self.__time = time_ 164 self.__current_time = self.__time.time() 165 166 self.__frozen = frozen 167 self.__lock = LockContext(lock_type) 168 self.__frozen_lock = LockContext(lock_type) 169 if self.__frozen: 170 self.__frozen_lock.acquire() 171 172 @property 173 def is_frozen(self) -> bool: 174 """ 175 Overview: 176 Get if this time proxy object is frozen 177 178 Returns: 179 bool: true if it is frozen, otherwise false 180 """ 181 with self.__lock: 182 return self.__frozen 183 184 def freeze(self): 185 """ 186 Overview: 187 Freeze this time proxy 188 """ 189 with self.__lock: 190 self.__frozen_lock.acquire() 191 self.__frozen = True 192 self.__current_time = self.__time.time() 193 194 def unfreeze(self): 195 """ 196 Overview: 197 Unfreeze this time proxy 198 """ 199 with self.__lock: 200 self.__frozen = False 201 self.__frozen_lock.release() 202 203 def time(self) -> Union[int, float]: 204 """ 205 Overview: 206 Get time (may be frozen time) 207 208 Returns: 209 int or float: the time 210 """ 211 with self.__lock: 212 if self.__frozen: 213 return self.__current_time 214 else: 215 return self.__time.time() 216 217 def current_time(self) -> Union[int, float]: 218 """ 219 Overview: 220 Get current time (will not be frozen time) 221 222 Returns: 223 int or float: current time 224 """ 225 return self.__time.time()