|
|
|
|
|
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
|
|
|
|
