File size: 5,048 Bytes
e98bd8c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
# Copyright (c) OpenMMLab. All rights reserved.
import inspect
import threading
import warnings
from collections import OrderedDict
from typing import Type, TypeVar
_lock = threading.RLock()
T = TypeVar('T')
def _accquire_lock() -> None:
"""Acquire the module-level lock for serializing access to shared data.
This should be released with _release_lock().
"""
if _lock:
_lock.acquire()
def _release_lock() -> None:
"""Release the module-level lock acquired by calling _accquire_lock()."""
if _lock:
_lock.release()
class ManagerMeta(type):
"""The metaclass for global accessible class.
The subclasses inheriting from ``ManagerMeta`` will manage their
own ``_instance_dict`` and root instances. The constructors of subclasses
must contain the ``name`` argument.
Examples:
>>> class SubClass1(metaclass=ManagerMeta):
>>> def __init__(self, *args, **kwargs):
>>> pass
AssertionError: <class '__main__.SubClass1'>.__init__ must have the
name argument.
>>> class SubClass2(metaclass=ManagerMeta):
>>> def __init__(self, name):
>>> pass
>>> # valid format.
"""
def __init__(cls, *args):
cls._instance_dict = OrderedDict()
params = inspect.getfullargspec(cls)
params_names = params[0] if params[0] else []
assert 'name' in params_names, f'{cls} must have the `name` argument'
super().__init__(*args)
class ManagerMixin(metaclass=ManagerMeta):
"""``ManagerMixin`` is the base class for classes that have global access
requirements.
The subclasses inheriting from ``ManagerMixin`` can get their
global instances.
Examples:
>>> class GlobalAccessible(ManagerMixin):
>>> def __init__(self, name=''):
>>> super().__init__(name)
>>>
>>> GlobalAccessible.get_instance('name')
>>> instance_1 = GlobalAccessible.get_instance('name')
>>> instance_2 = GlobalAccessible.get_instance('name')
>>> assert id(instance_1) == id(instance_2)
Args:
name (str): Name of the instance. Defaults to ''.
"""
def __init__(self, name: str = ''):
assert isinstance(name, str) and name, \
'name argument must be an non-empty string.'
self._instance_name = name
@classmethod
def get_instance(cls: Type[T], name: str, **kwargs) -> T:
"""Get subclass instance by name if the name exists.
If corresponding name instance has not been created, ``get_instance``
will create an instance, otherwise ``get_instance`` will return the
corresponding instance.
Examples
>>> instance1 = GlobalAccessible.get_instance('name1')
>>> # Create name1 instance.
>>> instance.instance_name
name1
>>> instance2 = GlobalAccessible.get_instance('name1')
>>> # Get name1 instance.
>>> assert id(instance1) == id(instance2)
Args:
name (str): Name of instance. Defaults to ''.
Returns:
object: Corresponding name instance, the latest instance, or root
instance.
"""
_accquire_lock()
assert isinstance(name, str), \
f'type of name should be str, but got {type(cls)}'
instance_dict = cls._instance_dict # type: ignore
# Get the instance by name.
if name not in instance_dict:
instance = cls(name=name, **kwargs) # type: ignore
instance_dict[name] = instance # type: ignore
elif kwargs:
warnings.warn(
f'{cls} instance named of {name} has been created, '
'the method `get_instance` should not accept any other '
'arguments')
# Get latest instantiated instance or root instance.
_release_lock()
return instance_dict[name]
@classmethod
def get_current_instance(cls):
"""Get latest created instance.
Before calling ``get_current_instance``, The subclass must have called
``get_instance(xxx)`` at least once.
Examples
>>> instance = GlobalAccessible.get_current_instance()
AssertionError: At least one of name and current needs to be set
>>> instance = GlobalAccessible.get_instance('name1')
>>> instance.instance_name
name1
>>> instance = GlobalAccessible.get_current_instance()
>>> instance.instance_name
name1
Returns:
object: Latest created instance.
"""
_accquire_lock()
if not cls._instance_dict:
raise RuntimeError(
f'Before calling {cls.__name__}.get_current_instance(), you '
'should call get_instance(name=xxx) at least once.')
name = next(iter(reversed(cls._instance_dict)))
_release_lock()
return cls._instance_dict[name]
|