|
|
|
|
|
import copy |
|
|
import time |
|
|
from contextlib import contextmanager |
|
|
from typing import Generator, Optional |
|
|
|
|
|
from ..utils.manager import ManagerMixin, _accquire_lock, _release_lock |
|
|
|
|
|
|
|
|
class DefaultScope(ManagerMixin): |
|
|
"""Scope of current task used to reset the current registry, which can be |
|
|
accessed globally. |
|
|
|
|
|
Consider the case of resetting the current ``Registry`` by |
|
|
``default_scope`` in the internal module which cannot access runner |
|
|
directly, it is difficult to get the ``default_scope`` defined in |
|
|
``Runner``. However, if ``Runner`` created ``DefaultScope`` instance |
|
|
by given ``default_scope``, the internal module can get |
|
|
``default_scope`` by ``DefaultScope.get_current_instance`` everywhere. |
|
|
|
|
|
Args: |
|
|
name (str): Name of default scope for global access. |
|
|
scope_name (str): Scope of current task. |
|
|
|
|
|
Examples: |
|
|
>>> from mmengine.model import MODELS |
|
|
>>> # Define default scope in runner. |
|
|
>>> DefaultScope.get_instance('task', scope_name='mmdet') |
|
|
>>> # Get default scope globally. |
|
|
>>> scope_name = DefaultScope.get_instance('task').scope_name |
|
|
""" |
|
|
|
|
|
def __init__(self, name: str, scope_name: str): |
|
|
super().__init__(name) |
|
|
assert isinstance( |
|
|
scope_name, |
|
|
str), (f'scope_name should be a string, but got {scope_name}') |
|
|
self._scope_name = scope_name |
|
|
|
|
|
@property |
|
|
def scope_name(self) -> str: |
|
|
""" |
|
|
Returns: |
|
|
str: Get current scope. |
|
|
""" |
|
|
return self._scope_name |
|
|
|
|
|
@classmethod |
|
|
def get_current_instance(cls) -> Optional['DefaultScope']: |
|
|
"""Get latest created default scope. |
|
|
|
|
|
Since default_scope is an optional argument for ``Registry.build``. |
|
|
``get_current_instance`` should return ``None`` if there is no |
|
|
``DefaultScope`` created. |
|
|
|
|
|
Examples: |
|
|
>>> default_scope = DefaultScope.get_current_instance() |
|
|
>>> # There is no `DefaultScope` created yet, |
|
|
>>> # `get_current_instance` return `None`. |
|
|
>>> default_scope = DefaultScope.get_instance( |
|
|
>>> 'instance_name', scope_name='mmengine') |
|
|
>>> default_scope.scope_name |
|
|
mmengine |
|
|
>>> default_scope = DefaultScope.get_current_instance() |
|
|
>>> default_scope.scope_name |
|
|
mmengine |
|
|
|
|
|
Returns: |
|
|
Optional[DefaultScope]: Return None If there has not been |
|
|
``DefaultScope`` instance created yet, otherwise return the |
|
|
latest created DefaultScope instance. |
|
|
""" |
|
|
_accquire_lock() |
|
|
if cls._instance_dict: |
|
|
instance = super().get_current_instance() |
|
|
else: |
|
|
instance = None |
|
|
_release_lock() |
|
|
return instance |
|
|
|
|
|
@classmethod |
|
|
@contextmanager |
|
|
def overwrite_default_scope(cls, scope_name: Optional[str]) -> Generator: |
|
|
"""Overwrite the current default scope with `scope_name`""" |
|
|
if scope_name is None: |
|
|
yield |
|
|
else: |
|
|
tmp = copy.deepcopy(cls._instance_dict) |
|
|
|
|
|
time.sleep(1e-6) |
|
|
cls.get_instance(f'overwrite-{time.time()}', scope_name=scope_name) |
|
|
try: |
|
|
yield |
|
|
finally: |
|
|
cls._instance_dict = tmp |
|
|
|
