Source code for ding.utils.autolog.time_ctl
import time
from abc import ABCMeta, abstractmethod
from typing import Union
from ..lock_helper import LockContext, LockContextType
[docs]class BaseTime(metaclass=ABCMeta):
"""
Overview:
Abstract time interface
Interfaces:
``time``
"""
[docs] @abstractmethod
def time(self) -> Union[int, float]:
"""
Overview:
Get time information
Returns:
- time(:obj:`float, int`): time information
"""
raise NotImplementedError
[docs]class NaturalTime(BaseTime):
"""
Overview:
Natural time object
Interfaces:
``__init__``, ``time``
Example:
>>> from ding.utils.autolog.time_ctl import NaturalTime
>>> time_ = NaturalTime()
"""
[docs] def time(self) -> float:
"""
Overview:
Get current natural time (float format, unix timestamp)
Returns:
- time(:obj:`float`): unix timestamp
Example:
>>> from ding.utils.autolog.time_ctl import NaturalTime
>>> time_ = NaturalTime()
>>> time_.time()
1603896383.8811457
"""
_current_time = time.time()
if self.__last_time is not None:
_current_time = max(_current_time, self.__last_time)
self.__last_time = _current_time
return _current_time
[docs]class TickTime(BaseTime):
"""
Overview:
Tick time object
Interfaces:
``__init__``, ``step``, ``time``
Example:
>>> from ding.utils.autolog.time_ctl import TickTime
>>> time_ = TickTime()
"""
[docs] def __init__(self, init: int = 0):
"""
Overview:
Constructor of TickTime
Arguments:
- init (:obj:`int`): initial time, default is 0
"""
self.__tick_time = init
[docs] def step(self, delta: int = 1) -> int:
"""
Overview
Step the time forward for this TickTime
Arguments:
- delta (:obj:`int`): steps to step forward, default is 1
Returns:
- 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
"""
if not isinstance(delta, int):
raise TypeError("Delta should be positive int, but {actual} found.".format(actual=type(delta).__name__))
elif delta < 1:
raise ValueError("Delta should be no less than 1, but {actual} found.".format(actual=repr(delta)))
else:
self.__tick_time += delta
return self.__tick_time
[docs] def time(self) -> int:
"""
Overview
Get current tick time
Returns:
int: current tick time
Example:
>>> from ding.utils.autolog.time_ctl import TickTime
>>> time_ = TickTime(0)
>>> time_.step()
>>> time_.time()
1
"""
return self.__tick_time
[docs]class TimeProxy(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
"""
[docs] def __init__(self, time_: BaseTime, frozen: bool = False, lock_type: LockContextType = LockContextType.THREAD_LOCK):
"""
Overview:
Constructor for Time proxy
Arguments:
- time_ (:obj:`BaseTime`): another time object it based on
- frozen (:obj:`bool`): this object will be frozen immediately if true, otherwise not, default is False
- lock_type (:obj:`LockContextType`): type of the lock, default is THREAD_LOCK
"""
self.__time = time_
self.__current_time = self.__time.time()
self.__frozen = frozen
self.__lock = LockContext(lock_type)
self.__frozen_lock = LockContext(lock_type)
if self.__frozen:
self.__frozen_lock.acquire()
@property
def is_frozen(self) -> bool:
"""
Overview:
Get if this time proxy object is frozen
Returns:
bool: true if it is frozen, otherwise false
"""
with self.__lock:
return self.__frozen
[docs] def freeze(self):
"""
Overview:
Freeze this time proxy
"""
with self.__lock:
self.__frozen_lock.acquire()
self.__frozen = True
self.__current_time = self.__time.time()
[docs] def unfreeze(self):
"""
Overview:
Unfreeze this time proxy
"""
with self.__lock:
self.__frozen = False
self.__frozen_lock.release()
[docs] def time(self) -> Union[int, float]:
"""
Overview:
Get time (may be frozen time)
Returns:
int or float: the time
"""
with self.__lock:
if self.__frozen:
return self.__current_time
else:
return self.__time.time()
[docs] def current_time(self) -> Union[int, float]:
"""
Overview:
Get current time (will not be frozen time)
Returns:
int or float: current time
"""
return self.__time.time()