ding.interaction.base.common

ding.interaction.base.common

ControllableContext

Overview

Basic context-supported class structure

Example: - Common usage

>>> c = MyControllableContext()  # One of the superclasses if ControllableContext
>>> c.start()
>>> try:
>>>     pass  # do anything you like
>>> finally:
>>>     c.close()

- Use with keyword (the same as code above)

>>> c = MyControllableContext()  # One of the superclasses if ControllableContext
>>> with c as cc:   # cc is c, have the same id
>>>     pass  # do anything you like

start() abstractmethod

Overview

Start the context

close() abstractmethod

Overview

Close the context

__enter__()

Overview

Enter the context

Returns: - self (:obj:ControllableContext): Context object itself

__exit__(exc_type, exc_val, exc_tb)

Overview

Exit the context

ControllableService

Bases: ControllableContext

Overview

Controllable service with context support, usually has concurrent feature.

Example: - A common usage

>>> c = MyControllableService()  # One of its superclasses is ControllableService
>>> c.start()
>>> try:
>>>     pass  # do anything you like
>>> finally:
>>>     c.shutdown()  # shutdown the service
>>>     c.join()  # wait until service is down

- Use with keyword (the same as code above)

>>> c = MyControllableService()  # One of its superclasses is ControllableService
>>> with c as cc:   # cc is c, have the same id
>>>     pass  # do anything you like

start() abstractmethod

Overview

Start the service

shutdown() abstractmethod

Overview

Shutdown the service (but service will not down immediately)

join() abstractmethod

Overview

Wait until the service is completely down

close()

Overview

Close the service, wait until the service is down.

random_token(length=None)

Overview

Generate random hex token

Arguments: - length (:obj:Optional[int]): Length of the random token (None means 64) Returns: - token (:obj:str): Generated random token Example: >>> random_token() # '4eAbd5218e3d0da5e7AAFcBF48Ea0Df2dadED1bdDF0B8724FdE1569AA78F24A7' >>> random_token(24) # 'Cd1CdD98caAb8602ac6501aC'

translate_dict_func(d)

Overview

Transform dict with funcs to function generating dict.

Arguments: - d (:obj:Mapping[str, Callable[..., Any]]): Dict with funcs Returns: - func (:obj:Callable[..., Dict[str, Any]]): Function generating dict Example: >>> f1 = lambda x, y: x + y >>> f2 = lambda x, y: x - y >>> f3 = lambda x, y: x * y >>> fx = translate_dict_func({'a': f1, 'b': f2, 'c': f3}) >>> fx(2, 3) # {'a': 5, 'b': -1, 'c': 6} >>> fx(5, 11) # ('a': 16, 'b': -6, 'c': 55}

default_func(return_value=None)

Overview

Transform optional function (maybe None) to function with default value

Argument: - return_value (:obj:): Return value of the default function Returns: - decorator (:obj:Callable[[Callable[..., Any]], Callable[..., Any]]): A decorator function that can decorator optional function to real function (must be not None) Example: >>> f1 = None >>> f2 = lambda x, y: x + y >>> ff1 = default_func()(f1) >>> ft1 = default_func(0)(f1) >>> ff2 = default_func()(f2) >>> ff1(2, 3) # None >>> ft1(2, 3) # 0 >>> ff2(2, 3) # 5

Full Source Code

../ding/interaction/base/common.py

1import random 2import string 3from abc import ABCMeta, abstractmethod 4from typing import Optional, Callable, Mapping, Any, Dict 5 6_LENGTH_OF_RANDOM_TOKEN = 64 7 8 9def random_token(length: Optional[int] = None) -> str: 10 """ 11 Overview: 12 Generate random hex token 13 Arguments: 14 - length (:obj:`Optional[int]`): Length of the random token (`None` means `64`) 15 Returns: 16 - token (:obj:`str`): Generated random token 17 Example: 18 >>> random_token() # '4eAbd5218e3d0da5e7AAFcBF48Ea0Df2dadED1bdDF0B8724FdE1569AA78F24A7' 19 >>> random_token(24) # 'Cd1CdD98caAb8602ac6501aC' 20 """ 21 return ''.join([random.choice(string.hexdigits) for _ in range(length or _LENGTH_OF_RANDOM_TOKEN)]) 22 23 24class ControllableContext(metaclass=ABCMeta): 25 """ 26 Overview: 27 Basic context-supported class structure 28 Example: 29 - Common usage 30 31 >>> c = MyControllableContext() # One of the superclasses if ControllableContext 32 >>> c.start() 33 >>> try: 34 >>> pass # do anything you like 35 >>> finally: 36 >>> c.close() 37 38 - Use with keyword (the same as code above) 39 40 >>> c = MyControllableContext() # One of the superclasses if ControllableContext 41 >>> with c as cc: # cc is c, have the same id 42 >>> pass # do anything you like 43 """ 44 45 @abstractmethod 46 def start(self): 47 """ 48 Overview: 49 Start the context 50 """ 51 raise NotImplementedError # pragma: no cover 52 53 @abstractmethod 54 def close(self): 55 """ 56 Overview: 57 Close the context 58 """ 59 raise NotImplementedError # pragma: no cover 60 61 def __enter__(self): 62 """ 63 Overview: 64 Enter the context 65 Returns: 66 - self (:obj:`ControllableContext`): Context object itself 67 """ 68 self.start() 69 return self 70 71 def __exit__(self, exc_type, exc_val, exc_tb): 72 """ 73 Overview: 74 Exit the context 75 """ 76 self.close() 77 78 79class ControllableService(ControllableContext, metaclass=ABCMeta): 80 """ 81 Overview: 82 Controllable service with context support, usually has concurrent feature. 83 Example: 84 - A common usage 85 86 >>> c = MyControllableService() # One of its superclasses is ControllableService 87 >>> c.start() 88 >>> try: 89 >>> pass # do anything you like 90 >>> finally: 91 >>> c.shutdown() # shutdown the service 92 >>> c.join() # wait until service is down 93 94 - Use with keyword (the same as code above) 95 96 >>> c = MyControllableService() # One of its superclasses is ControllableService 97 >>> with c as cc: # cc is c, have the same id 98 >>> pass # do anything you like 99 """ 100 101 @abstractmethod 102 def start(self): 103 """ 104 Overview: 105 Start the service 106 """ 107 raise NotImplementedError # pragma: no cover 108 109 @abstractmethod 110 def shutdown(self): 111 """ 112 Overview: 113 Shutdown the service (but service will not down immediately) 114 """ 115 raise NotImplementedError # pragma: no cover 116 117 @abstractmethod 118 def join(self): 119 """ 120 Overview: 121 Wait until the service is completely down 122 """ 123 raise NotImplementedError # pragma: no cover 124 125 def close(self): 126 """ 127 Overview: 128 Close the service, wait until the service is down. 129 """ 130 self.shutdown() 131 self.join() 132 133 134def translate_dict_func(d: Mapping[str, Callable[..., Any]]) -> Callable[..., Dict[str, Any]]: 135 """ 136 Overview: 137 Transform dict with funcs to function generating dict. 138 Arguments: 139 - d (:obj:`Mapping[str, Callable[..., Any]]`): Dict with funcs 140 Returns: 141 - func (:obj:`Callable[..., Dict[str, Any]]`): Function generating dict 142 Example: 143 >>> f1 = lambda x, y: x + y 144 >>> f2 = lambda x, y: x - y 145 >>> f3 = lambda x, y: x * y 146 >>> fx = translate_dict_func({'a': f1, 'b': f2, 'c': f3}) 147 >>> fx(2, 3) # {'a': 5, 'b': -1, 'c': 6} 148 >>> fx(5, 11) # ('a': 16, 'b': -6, 'c': 55} 149 """ 150 151 def _func(*args, **kwargs) -> Dict[str, Any]: 152 return {k: f(*args, **kwargs) for k, f in d.items()} 153 154 return _func 155 156 157def default_func(return_value=None) -> Callable[[Callable[..., Any]], Callable[..., Any]]: 158 """ 159 Overview: 160 Transform optional function (maybe `None`) to function with default value 161 Argument: 162 - return_value (:obj:): Return value of the default function 163 Returns: 164 - decorator (:obj:`Callable[[Callable[..., Any]], Callable[..., Any]]`): A decorator function \ 165 that can decorator optional function to real function (must be not None) 166 Example: 167 >>> f1 = None 168 >>> f2 = lambda x, y: x + y 169 >>> ff1 = default_func()(f1) 170 >>> ft1 = default_func(0)(f1) 171 >>> ff2 = default_func()(f2) 172 >>> ff1(2, 3) # None 173 >>> ft1(2, 3) # 0 174 >>> ff2(2, 3) # 5 175 """ 176 177 def _decorator(func: Callable[..., Any]) -> Callable[..., Any]: 178 # noinspection PyUnusedLocal 179 def _func(*args, **kwargs): 180 return return_value 181 182 return func or _func 183 184 return _decorator