注册表管理器 (Registry Manager)¶

RegisterMap ¶

RegisterMap()

A global register map for registering and retrieving objects.

一个全局注册表，用于注册和检索对象。

Source code in tfrobot/registry/base.py

def __init__(self) -> None:
    # 全局注册表 用于存储所有的注册信息 KEY为Str类型，VALUE为BaseModel类型或者BaseModel子类或者Callable类型
    self._register_map: dict[str, set[Union[Type[BaseModel], BaseModel, Callable]]] = {}
    # 全局实例容器管理 与全局注册表相反的是，注册表管理所有的「类」，而容器管理的是所有的「实例」
    # 因为我的构建机器人或者组件往往是动态构建，所以在此使用DynamicContainer来管理所有的实例
    self._container_map: dict[str, containers.DynamicContainer] = {}

register ¶

register(key: str, value: Union[Type[BaseModel], BaseModel, Callable]) -> None

Register an object to the register map. If the key does not exist, a new set will be created.

将对象注册到注册表中。如果键不存在，将创建一个新的集合。

Parameters:

Name	Type	Description	Default
`key`	`str`	The key under which the object will be registered. The key should be a string. 用于注册对象的键。键应为字符串。	required
`value`	`Union[Type[BaseModel], BaseModel, Callable]`	The object to be registered. The object should be an instance of BaseModel, a subclass of BaseModel, or a callable. 要注册的对象。对象应为BaseModel的实例，BaseModel的子类，或者是可调用的。	required

Raises:

Type	Description
`TypeError`	If the key is not a string or the value is not an instance of BaseModel, a subclass of BaseModel, or a callable. 如果键不是字符串，或者值不是BaseModel的实例，BaseModel的子类，或者是可调用的。

Example

register_map = RegisterMap() register_map.register("example_key", example_value) # type: ignore

Source code in tfrobot/registry/base.py

def register(self, key: str, value: Union[Type[BaseModel], BaseModel, Callable]) -> None:
    """
    Register an object to the register map. If the key does not exist, a new set will be created.

    将对象注册到注册表中。如果键不存在，将创建一个新的集合。

    Args:
        key (str): The key under which the object will be registered. The key should be a string.

            用于注册对象的键。键应为字符串。
        value (Union[Type[BaseModel], BaseModel, Callable]): The object to be registered. The object should be an
            instance of BaseModel, a subclass of BaseModel, or a callable.

            要注册的对象。对象应为BaseModel的实例，BaseModel的子类，或者是可调用的。

    Raises:
        TypeError: If the key is not a string or the value is not an instance of BaseModel, a subclass of BaseModel,
            or a callable.

            如果键不是字符串，或者值不是BaseModel的实例，BaseModel的子类，或者是可调用的。

    Example:
        >>> register_map = RegisterMap()
        >>> register_map.register("example_key", example_value)  # type: ignore
    """
    if isinstance(value, BaseModel):
        raise ValueError("当前版本的注册中心不支持BaseModel实例的注册，目前仅支持注册「类」与「函数」")
    if (v_name := generate_register_name(value)) in self.get_map(key):
        raise ValueError(f"{v_name} already exists in the register map.")
    # 如果key不存在，则创建一个空的set
    registered_objects = self._register_map.setdefault(key, set())
    registered_objects.add(value)

unregister ¶

unregister(key: str, value: Union[Type[BaseModel], BaseModel, Callable]) -> None

Unregister an object from the register map. If the key or value does not exist, raise a ValueError.

从注册表中注销对象。如果键或值不存在，则引发 ValueError。

The main use case for this method is if a third-party developer needs to use a specific tool name, but this tool name is occupied by the default tool or another tool. On the premise that the third-party developer is clearly aware of the pros and cons, they can use unregister to cancel the registration.

其主要应用场景为，如果第三方开发者需要使用一个特定的工具名，但这个工具名被默认工具或者其它工具占用，在第三方开发者明确知晓其利害的前提下，可以使用unregister进行取消注册。

Parameters:

Name	Type	Description	Default
`key`	`str`	The key under which the object is registered. The key should be a string. 对象注册的键。键应为字符串。	required
`value`	`Union[Type[BaseModel], BaseModel, Callable]`	The object to be unregistered. The object should be an instance of BaseModel, a subclass of BaseModel, or a callable. 要注销的对象。对象应为 BaseModel 的实例，BaseModel 的子类，或者是可调用的。	required

Raises:

Type	Description
`ValueError`	If the key does not exist in the register map or the value is not associated with the key. 如果键在注册表中不存在或值与键不关联，则引发 ValueError。

Example

register_map = RegisterMap() register_map.unregister("example_key", example_value) # type: ignore

Source code in tfrobot/registry/base.py

def unregister(self, key: str, value: Union[Type[BaseModel], BaseModel, Callable]) -> None:
    """
    Unregister an object from the register map. If the key or value does not exist, raise a ValueError.

    从注册表中注销对象。如果键或值不存在，则引发 ValueError。

    The main use case for this method is if a third-party developer needs to use a specific tool name, but this tool
        name is occupied by the default tool or another tool. On the premise that the third-party developer is
        clearly aware of the pros and cons, they can use unregister to cancel the registration.

    其主要应用场景为，如果第三方开发者需要使用一个特定的工具名，但这个工具名被默认工具或者其它工具占用，在第三方开发者明确知晓其利害的前提下，可以使用unregister进行取消注册。

    Args:
        key (str): The key under which the object is registered. The key should be a string.

            对象注册的键。键应为字符串。
        value (Union[Type[BaseModel], BaseModel, Callable]): The object to be unregistered. The object should be an
            instance of BaseModel, a subclass of BaseModel, or a callable.

            要注销的对象。对象应为 BaseModel 的实例，BaseModel 的子类，或者是可调用的。

    Raises:
        ValueError: If the key does not exist in the register map or the value is not associated with the key.

            如果键在注册表中不存在或值与键不关联，则引发 ValueError。

    Example:
        >>> register_map = RegisterMap()
        >>> register_map.unregister("example_key", example_value)  # type: ignore
    """
    if key not in self._register_map:
        raise ValueError(f"Key {key} does not exist in the register map.")
    registered_objects = self._register_map[key]
    if value not in registered_objects:
        raise ValueError(f"Value {value} is not registered under key {key}.")
    registered_objects.remove(value)
    if not registered_objects:
        del self._register_map[key]

get ¶

get(key: str) -> set[Union[Type[BaseModel], BaseModel, Callable]]

Get an object from the register map. 从注册表中获取对象。

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the object to be retrieved. The key should be a string. 要检索的对象的键。键应为字符串。	required

Returns:

Type	Description
`set[Union[Type[BaseModel], BaseModel, Callable]]`	set[Union[Type[BaseModel], BaseModel, Callable]]: The object associated with the key. 与键关联的对象。

Raises:

Type	Description
`ValueError`	If the key does not exist in the register map. 如果键在注册表中不存在。

Example

register_map = RegisterMap() register_map.get("example_key")

Source code in tfrobot/registry/base.py

def get(self, key: str) -> set[Union[Type[BaseModel], BaseModel, Callable]]:
    """
    Get an object from the register map.
    从注册表中获取对象。

    Args:
        key (str): The key of the object to be retrieved. The key should be a string.

            要检索的对象的键。键应为字符串。

    Returns:
        set[Union[Type[BaseModel], BaseModel, Callable]]: The object associated with the key.

            与键关联的对象。

    Raises:
        ValueError: If the key does not exist in the register map.

            如果键在注册表中不存在。

    Example:
        >>> register_map = RegisterMap()
        >>> register_map.get("example_key")
    """
    if key not in self._register_map:
        raise ValueError(f"Key {key} does not exist in the register map.")
    return self._register_map[key]

get_map ¶

get_map(key: str) -> dict[str, Union[Type[BaseModel], BaseModel, Callable]]

Return the object list as a dictionary, with the key generated by generate, and support cache mode to ensure that the key value is not recalculated when there are a large number of requests.

将对象列表以字典形式返回，字典的key使用generate生成，同时支持缓存模式，保证大量请求的时候不会重复计算key值。

Parameters:

Name	Type	Description	Default
`key`	`str`	The key for the register list to be retrieved. The key should be a string. 要检索的注册表列表的键。键应为字符串。	required

Returns:

Type	Description
`dict[str, Union[Type[BaseModel], BaseModel, Callable]]`	dict[str, Union[Type[BaseModel], BaseModel, Callable]]: The dictionary of objects associated with the key. 与键关联的对象的字典。

Raises:

Type	Description
`ValueError`	If the key does not exist in the register map. 如果键在注册表中不存在。

Example

register_map = RegisterMap() register_map.get_map("TOOL")

Source code in tfrobot/registry/base.py

def get_map(self, key: str) -> dict[str, Union[Type[BaseModel], BaseModel, Callable]]:
    """
    Return the object list as a dictionary, with the key generated by generate, and support cache mode to ensure
    that the key value is not recalculated when there are a large number of requests.

    将对象列表以字典形式返回，字典的key使用generate生成，同时支持缓存模式，保证大量请求的时候不会重复计算key值。

    Args:
        key (str): The key for the register list to be retrieved. The key should be a string.

            要检索的注册表列表的键。键应为字符串。

    Returns:
        dict[str, Union[Type[BaseModel], BaseModel, Callable]]: The dictionary of objects associated with the key.

            与键关联的对象的字典。

    Raises:
        ValueError: If the key does not exist in the register map.

            如果键在注册表中不存在。

    Example:
        >>> register_map = RegisterMap()
        >>> register_map.get_map("TOOL")
    """
    return self.get_set_to_map(self._register_map.get(key, set()))

get_container ¶

get_container(key: str) -> DynamicContainer

Get the container from the container map. | 从容器映射中获取容器。

If key not exist, generate a new container and return it. | 如果键不存在，则生成一个新的容器并返回。

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the container to be retrieved. The key should be a string. 要检索的容器的键。键应为字符串。	required

Returns:

Type	Description
`DynamicContainer`	Optional[containers.DynamicContainer]: The container associated with the key. 与键关联的容器。

Source code in tfrobot/registry/base.py

def get_container(self, key: str) -> containers.DynamicContainer:
    """
    Get the container from the container map. | 从容器映射中获取容器。

    If key not exist, generate a new container and return it. | 如果键不存在，则生成一个新的容器并返回。

    Args:
        key (str): The key of the container to be retrieved. The key should be a string.

            要检索的容器的键。键应为字符串。

    Returns:
        Optional[containers.DynamicContainer]: The container associated with the key.

            与键关联的容器。
    """
    return self._container_map.setdefault(key, containers.DynamicContainer())

set_container ¶

set_container(key: str, container: DynamicContainer) -> None

Set the container to the container map. | 将容器设置到容器映射中。

Parameters:

Name	Type	Description	Default
`key`	`str`	The key of the container to be set. The key should be a string. 要设置的容器的键。键应为字符串。	required
`container`	`DynamicContainer`	The container to be set. 要设置的容器。	required

Returns:

Type	Description
`None`	None

Source code in tfrobot/registry/base.py

def set_container(self, key: str, container: containers.DynamicContainer) -> None:
    """
    Set the container to the container map. | 将容器设置到容器映射中。

    Args:
        key (str): The key of the container to be set. The key should be a string.

            要设置的容器的键。键应为字符串。
        container (containers.DynamicContainer): The container to be set.

            要设置的容器。

    Returns:
        None
    """
    self._container_map[key] = container

get_set_to_map ¶

get_set_to_map(obj_set: set[Union[Type[BaseModel], BaseModel, Callable]]) -> dict[str, Union[Type[BaseModel], BaseModel, Callable]]

Return the object list as a dictionary, with the key generated by generate. This function supports cache mode to ensure that the key value is not recalculated when there are a large number of requests.

将对象列表以字典形式返回，字典的key使用generate生成，同时支持缓存模式，保证大量请求的时候不会重复计算key值。

Parameters:

Name	Type	Description	Default
`obj_set`	`set[Union[Type[BaseModel], BaseModel, Callable]]`	The set of objects to be converted into a dictionary. 要转换为字典的对象集合。	required

Returns:

Type	Description
`dict[str, Union[Type[BaseModel], BaseModel, Callable]]`	dict[str, Union[Type[BaseModel], BaseModel, Callable]]: The dictionary of objects. 对象的字典。

Example

register_map = RegisterMap() register_map.get_set_to_map(obj_set)

Source code in tfrobot/registry/base.py

@cached(cache=LRUCache(maxsize=128), key=make_hashable_key)
def get_set_to_map(
    self, obj_set: set[Union[Type[BaseModel], BaseModel, Callable]]
) -> dict[str, Union[Type[BaseModel], BaseModel, Callable]]:
    """
    Return the object list as a dictionary, with the key generated by generate. This function supports cache mode
    to ensure that the key value is not recalculated when there are a large number of requests.

    将对象列表以字典形式返回，字典的key使用generate生成，同时支持缓存模式，保证大量请求的时候不会重复计算key值。

    Args:
        obj_set (set[Union[Type[BaseModel], BaseModel, Callable]]): The set of objects to be converted into a
            dictionary.

            要转换为字典的对象集合。

    Returns:
        dict[str, Union[Type[BaseModel], BaseModel, Callable]]: The dictionary of objects.

            对象的字典。

    Example:
        >>> register_map = RegisterMap()
        >>> register_map.get_set_to_map(obj_set)
    """
    return {generate_register_name(obj): obj for obj in obj_set}

keys ¶

keys() -> KeysView[str]

Get the keys of the register map.

获取注册表的键。

Returns:

Type	Description
`KeysView[str]`	KeysView[str]: A view object that displays the keys in the register map. 显示注册表中键的视图对象。

Example

register_map = RegisterMap() keys = register_map.keys() print(keys)

Source code in tfrobot/registry/base.py

def keys(self) -> KeysView[str]:
    """
    Get the keys of the register map.

    获取注册表的键。

    Returns:
        KeysView[str]: A view object that displays the keys in the register map.

            显示注册表中键的视图对象。

    Example:
        >>> register_map = RegisterMap()
        >>> keys = register_map.keys()
        >>> print(keys)
    """
    return self._register_map.keys()

items ¶

items() -> Iterable[tuple[str, set[Union[Type[BaseModel], BaseModel, Callable]]]]

Get the items of the register map.

获取注册表的项目。

Returns:

Type	Description
`Iterable[tuple[str, set[Union[Type[BaseModel], BaseModel, Callable]]]]`	Iterable[tuple[str, set[Union[Type[BaseModel], BaseModel, Callable]]]]: An iterable over the (key, value) pairs in the register map. 注册表中的（键，值）对的迭代器。

Example

register_map = RegisterMap() items = register_map.items() for item in items: print(item)

Source code in tfrobot/registry/base.py

def items(self) -> Iterable[tuple[str, set[Union[Type[BaseModel], BaseModel, Callable]]]]:
    """
    Get the items of the register map.

    获取注册表的项目。

    Returns:
        Iterable[tuple[str, set[Union[Type[BaseModel], BaseModel, Callable]]]]: An iterable over the (key, value)
            pairs in the register map.

            注册表中的（键，值）对的迭代器。

    Example:
        >>> register_map = RegisterMap()
        >>> items = register_map.items()
        >>> for item in items:
        >>>     print(item)
    """
    return self._register_map.items()

find_by_name ¶

find_by_name(name: str) -> Union[Type[BaseModel], BaseModel, Callable]

Find an object by name.

通过名称查找对象。

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the object to find. 要查找的对象的名称	required

Returns:

Type	Description
`Union[Type[BaseModel], BaseModel, Callable]`	Union[Type[BaseModel], BaseModel, Callable]: The object associated with the name. 与名称关联的对象。

Raises:

Type	Description
`ValueError`	If the name does not exist in the register map. 如果名称在注册表中不存在。

Example

register_map = RegisterMap() obj = register_map.find_by_name("example_name")

Source code in tfrobot/registry/base.py

def find_by_name(self, name: str) -> Union[Type[BaseModel], BaseModel, Callable]:
    """
    Find an object by name.

    通过名称查找对象。

    Args:
        name (str): The name of the object to find.

            要查找的对象的名称

    Returns:
        Union[Type[BaseModel], BaseModel, Callable]: The object associated with the name.

            与名称关联的对象。

    Raises:
        ValueError: If the name does not exist in the register map.

            如果名称在注册表中不存在。

    Example:
        >>> register_map = RegisterMap()
        >>> obj = register_map.find_by_name("example_name")
    """
    for key in self.keys():
        for k, v in self.get_map(key).items():
            if k == name:
                return v
    raise ValueError(f"Can not find object by name {name}")