| | |
| | import copy |
| | import time |
| | from contextlib import contextmanager |
| | from typing import Generator, Optional |
| |
|
| | from mmengine.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 |
| |
|
