Skip to content

algorithm

logger = logging.getLogger(__name__) module-attribute

ALGORITHM_REGISTRY = AlgorithmRegistry() module-attribute

Registry instantiation for algorithms.

Contains the streamsight algorithms by default and allows registration of new algorithms via the register function.

Examples:

from streamsight.pipelines import ALGORITHM_REGISTRY

# Construct an ItemKNN object with parameter K=20
algo = ALGORITHM_REGISTRY.get("ItemKNN")(K=20)

from streamsight.algorithms import ItemKNN

ALGORITHM_REGISTRY.register("HelloWorld", ItemKNN)

# Also construct an ItemKNN object with parameter K=20
algo = ALGORITHM_REGISTRY.get("HelloWorld")(K=20)

AlgorithmRegistry

Bases: Registry[Algorithm]

Registry for easy retrieval of algorithm types by name.

The registry is pre-registered with all streamsight algorithms.

Source code in src/streamsight/registries/algorithm.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
class AlgorithmRegistry(Registry[Algorithm]):
    """Registry for easy retrieval of algorithm types by name.

    The registry is pre-registered with all streamsight algorithms.
    """

    def __init__(self) -> None:
        """Initialize the algorithm registry.

        The registry is initialized with the `streamsight.algorithms` module
        so that all built-in algorithms are available by default.
        """
        module = importlib.import_module("streamsight.algorithms")
        super().__init__(module)

IS_BASE = True class-attribute instance-attribute

name property

Name of the object's class.

:return: Name of the object's class :rtype: str

registered = {} instance-attribute

src = src instance-attribute

get(key)

Retrieve the value for this key.

This value is a Python type, most often a class.

Parameters:

Name Type Description Default
key str

The key to fetch.

 required

Returns:

Type Description
T

The class type associated with the key.

Raises:

Type Description
KeyError

If the key is not found in the registry.
Source code in src/streamsight/registries/base.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def get(self, key: str) -> T:
    """Retrieve the value for this key.

    This value is a Python type, most often a class.

    Args:
        key: The key to fetch.

    Returns:
        The class type associated with the key.

    Raises:
        KeyError: If the key is not found in the registry.
    """
    return self[key]

register(key, cls)

Register a new Python type (most often a class).

After registration, the key can be used to fetch the Python type from the registry.

Parameters:

Name Type Description Default
key str

Key to register the type at. Needs to be unique to the registry.

 required
cls T

Class to register.

 required

Raises:

Type Description
KeyError

If the key is already registered.
Source code in src/streamsight/registries/base.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def register(self, key: str, cls: T) -> None:
    """Register a new Python type (most often a class).

    After registration, the key can be used to fetch the Python type from the registry.

    Args:
        key: Key to register the type at. Needs to be unique to the registry.
        cls: Class to register.

    Raises:
        KeyError: If the key is already registered.
    """
    if key in self.registered:
        raise KeyError(f"key `{key}` already registered")
    self.registered[key] = cls

get_registered_keys(include_base=False)

Get a list of all registered keys.

Returns:

Type Description
list[str]

A list of all registered keys.
Source code in src/streamsight/registries/base.py 
106
107
108
109
110
111
112
113
114
115
def get_registered_keys(self, include_base: bool = False) -> list[str]:
    """Get a list of all registered keys.

    Returns:
        A list of all registered keys.
    """
    if include_base:
        return list(self.registered.keys())
    else:
        return [key for key, cls in self.registered.items() if not getattr(cls, "IS_BASE", True)]

registered_values()

Get a list of all registered types.

Returns:

Type Description
list[T]

A list of all registered types.
Source code in src/streamsight/registries/base.py 
117
118
119
120
121
122
123
def registered_values(self) -> list[T]:
    """Get a list of all registered types.

    Returns:
        A list of all registered types.
    """
    return [self.registered[key] for key in self.get_registered_keys(include_base=False)]

registered_items()

Get a list of all registered key-type pairs.

Returns:

Type Description
list[tuple[str, T]]

A list of all registered key-type pairs.
Source code in src/streamsight/registries/base.py 
125
126
127
128
129
130
131
def registered_items(self) -> list[tuple[str, T]]:
    """Get a list of all registered key-type pairs.

    Returns:
        A list of all registered key-type pairs.
    """
    return [(key, self.registered[key]) for key in self.get_registered_keys(include_base=False)]

AlgorithmEntry

Bases: NamedTuple

Entry for the algorithm registry.

The intended use of this class is to store the name of the algorithm and the parameters that the algorithm should take. Mainly this is used during the building phase of the evaluator pipeline in Builder.

Attributes:

Name Type Description
name str

Name of the algorithm.
params None | dict[str, Any]

Parameters that do not require optimization as key-value pairs, where the key is the hyperparameter name and the value is the value it should take.
Source code in src/streamsight/registries/algorithm.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class AlgorithmEntry(NamedTuple):
    """Entry for the algorithm registry.

    The intended use of this class is to store the name of the algorithm and
    the parameters that the algorithm should take. Mainly this is used during
    the building phase of the evaluator pipeline in `Builder`.

    Attributes:
        name: Name of the algorithm.
        params: Parameters that do not require optimization as key-value
            pairs, where the key is the hyperparameter name and the value is
            the value it should take.
    """

    name: str
    uuid: uuid.UUID | None = None
    params: None | dict[str, Any] = None

name instance-attribute

uuid = None class-attribute instance-attribute

params = None class-attribute instance-attribute