Module streamgen.parameter
⚙️ parameters are variables that change over time according to a schedule.
View Source
"""⚙️ parameters are variables that change over time according to a schedule."""
from __future__ import annotations
from collections.abc import Iterable
from copy import deepcopy
from itertools import cycle
from typing import Generic, Self, TypeAlias, TypedDict, TypeVar
from beartype import beartype
from streamgen.enums import (
ParameterOutOfRangeStrategy,
ParameterOutOfRangeStrategyLit,
)
from streamgen.exceptions import ParameterOutOfRangeError
T = TypeVar("T")
@beartype()
class Parameter(Generic[T]):
"""⚙️ parameters are variables that change over time according to a schedule.
Args:
name (str): variable name of the parameter. Defaults to "param".
value (Generic[T] | None): the value of the parameter.
if None and schedule is defined, use the first value of the schedule.
Defaults to None.
schedule (Iterable[T] | None): a schedule for the parameter.
Defaults to None.
parameter_out_of_range_strategy
(ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit):
strategy which defines what happens when calling `update` and there is
no valid next value available. Defaults to `"hold"`, where the last
valid value is held (not updated).
"""
def __init__( # noqa: D107
self,
name: str = "param",
value: T | None = None,
schedule: Iterable[T] | None = None,
strategy: ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit = "hold",
) -> None:
assert "." not in name, "`.` in parameter names are reserved for scopes." # noqa: S101
self.name = name
self.value = value
self.schedule = iter(schedule) if schedule is not None else None
self.strategy = strategy
if strategy == ParameterOutOfRangeStrategy.CYCLE:
self.schedule = cycle(self.schedule)
if self.value is None and self.schedule:
self.value = self.update()
self._initial_schedule = deepcopy(self.schedule)
self._initial_value = self.value
def update(self) -> T:
"""🆙 updates the value according to the schedule and strategy.
Returns:
T: updated value
Raises:
ParameterOutOfRangeError: when an `update` leads to an invalid value.
"""
if self.schedule is None:
return self.value
try:
self.value = next(self.schedule)
except StopIteration as err:
if self.strategy == ParameterOutOfRangeStrategy.RAISE_EXCEPTION:
raise ParameterOutOfRangeError from err
return self.value
def __getitem__(self, idx: int) -> T:
"""🫱 gets the value after a certain number of update steps.
This function resets the current schedule to its original schedule during construction.
Returns:
T: value after `idx` updates
Raises:
ParameterOutOfRangeError: when an `update` leads to an invalid value.
"""
self.schedule = deepcopy(self._initial_schedule)
self.value = self._initial_value
for _ in range(idx):
self.update()
return self.value
def __repr__(self) -> str:
"""🏷️ Returns the string representation `str(self)`.
Returns:
str: string representation of self
"""
return f"{self.name}={self.value}"
def __or__(self, param: Self) -> ParameterStore:
"""➕ combines two Parameters to a `ParameterStore` using `|`.
Args:
param (Parameter): another parameter
Returns:
ParameterStore: combined parameter store
""" # noqa: RUF002
return ParameterStore([self, param])
class ParameterDict(Generic[T], TypedDict, total=False):
"""📖 typed dictionary of `streamgen.parameter.Parameter`."""
name: str | None
value: T | None
schedule: Iterable[T] | None
strategy: ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit | None
ScopedParameterDict: TypeAlias = dict[str, ParameterDict | dict[str, ParameterDict]]
"""🔭📖 representation of multiple `streamgen.parameter.Parameter` as a dictionary.
The dictionary can be nested one level to create parameter scopes.
All top-level parameters are considered as having `scope=None`.
Examples:
>>> params = {
"var1": {
"value": 1,
"schedule": [2,3],
"strategy": "cycle",
},
"var2": {
"name": "var2", # can be present, but is not needed
"schedule": [0.1, 0.2, 0.3],
},
"scope1": {
"var1": { # var1 can be used again since its inside a scope
"value": 1,
"schedule": [2,3],
"strategy": "cycle",
},
},
}
"""
# 🔄️ bottom-level import required to avoid circular dependency
from streamgen.parameter.store import ParameterStore # noqa: E402
Sub-modules
Variables
T
Classes
Parameter
class Parameter(
name: str = 'param',
value: Optional[~T] = None,
schedule: collections.abc.Iterable[~T] | None = None,
strategy: Union[streamgen.enums.ParameterOutOfRangeStrategy, Literal['hold', 'cycle', 'raise exception']] = 'hold'
)
⚙️ parameters are variables that change over time according to a schedule.
Attributes
| Name | Type | Description | Default |
|---|---|---|---|
| name | str | variable name of the parameter. Defaults to "param". | "param" |
| value | Generic[T] | None | the value of the parameter. if None and schedule is defined, use the first value of the schedule. Defaults to None. |
| schedule | Iterable[T] | None | a schedule for the parameter. Defaults to None. |
| parameter_out_of_range_strategy | ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit | strategy which defines what happens when calling update and there isno valid next value available. Defaults to "hold", where the lastvalid value is held (not updated). |
View Source
@beartype()
class Parameter(Generic[T]):
"""⚙️ parameters are variables that change over time according to a schedule.
Args:
name (str): variable name of the parameter. Defaults to "param".
value (Generic[T] | None): the value of the parameter.
if None and schedule is defined, use the first value of the schedule.
Defaults to None.
schedule (Iterable[T] | None): a schedule for the parameter.
Defaults to None.
parameter_out_of_range_strategy
(ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit):
strategy which defines what happens when calling `update` and there is
no valid next value available. Defaults to `"hold"`, where the last
valid value is held (not updated).
"""
def __init__( # noqa: D107
self,
name: str = "param",
value: T | None = None,
schedule: Iterable[T] | None = None,
strategy: ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit = "hold",
) -> None:
assert "." not in name, "`.` in parameter names are reserved for scopes." # noqa: S101
self.name = name
self.value = value
self.schedule = iter(schedule) if schedule is not None else None
self.strategy = strategy
if strategy == ParameterOutOfRangeStrategy.CYCLE:
self.schedule = cycle(self.schedule)
if self.value is None and self.schedule:
self.value = self.update()
self._initial_schedule = deepcopy(self.schedule)
self._initial_value = self.value
def update(self) -> T:
"""🆙 updates the value according to the schedule and strategy.
Returns:
T: updated value
Raises:
ParameterOutOfRangeError: when an `update` leads to an invalid value.
"""
if self.schedule is None:
return self.value
try:
self.value = next(self.schedule)
except StopIteration as err:
if self.strategy == ParameterOutOfRangeStrategy.RAISE_EXCEPTION:
raise ParameterOutOfRangeError from err
return self.value
def __getitem__(self, idx: int) -> T:
"""🫱 gets the value after a certain number of update steps.
This function resets the current schedule to its original schedule during construction.
Returns:
T: value after `idx` updates
Raises:
ParameterOutOfRangeError: when an `update` leads to an invalid value.
"""
self.schedule = deepcopy(self._initial_schedule)
self.value = self._initial_value
for _ in range(idx):
self.update()
return self.value
def __repr__(self) -> str:
"""🏷️ Returns the string representation `str(self)`.
Returns:
str: string representation of self
"""
return f"{self.name}={self.value}"
def __or__(self, param: Self) -> ParameterStore:
"""➕ combines two Parameters to a `ParameterStore` using `|`.
Args:
param (Parameter): another parameter
Returns:
ParameterStore: combined parameter store
""" # noqa: RUF002
return ParameterStore([self, param])
Ancestors (in MRO)
- typing.Generic
Methods
update
def update(
self
) -> ~T
🆙 updates the value according to the schedule and strategy.
Returns:
| Type | Description |
|---|---|
| T | updated value |
Raises:
| Type | Description |
|---|---|
| ParameterOutOfRangeError | when an update leads to an invalid value. |
View Source
def update(self) -> T:
"""🆙 updates the value according to the schedule and strategy.
Returns:
T: updated value
Raises:
ParameterOutOfRangeError: when an `update` leads to an invalid value.
"""
if self.schedule is None:
return self.value
try:
self.value = next(self.schedule)
except StopIteration as err:
if self.strategy == ParameterOutOfRangeStrategy.RAISE_EXCEPTION:
raise ParameterOutOfRangeError from err
return self.value
ParameterDict
class ParameterDict(
/,
*args,
**kwargs
)
📖 typed dictionary of streamgen.parameter.Parameter.
View Source
class ParameterDict(Generic[T], TypedDict, total=False):
"""📖 typed dictionary of `streamgen.parameter.Parameter`."""
name: str | None
value: T | None
schedule: Iterable[T] | None
strategy: ParameterOutOfRangeStrategy | ParameterOutOfRangeStrategyLit | None
Ancestors (in MRO)
- typing.Generic
- builtins.dict
Methods
clear
def clear(
...
)
D.clear() -> None. Remove all items from D.
copy
def copy(
...
)
D.copy() -> a shallow copy of D
fromkeys
def fromkeys(
iterable,
value=None,
/
)
Create a new dictionary with keys from iterable and values set to value.
get
def get(
self,
key,
default=None,
/
)
Return the value for key if key is in the dictionary, else default.
items
def items(
...
)
D.items() -> a set-like object providing a view on D's items
keys
def keys(
...
)
D.keys() -> a set-like object providing a view on D's keys
pop
def pop(
...
)
D.pop(k[,d]) -> v, remove specified key and return the corresponding value.
If the key is not found, return the default if given; otherwise, raise a KeyError.
popitem
def popitem(
self,
/
)
Remove and return a (key, value) pair as a 2-tuple.
Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.
setdefault
def setdefault(
self,
key,
default=None,
/
)
Insert key with a value of default if key is not in the dictionary.
Return the value for key if key is in the dictionary, else default.
update
def update(
...
)
D.update([E, ]**F) -> None. Update D from dict/iterable E and F.
If E is present and has a .keys() method, then does: for k in E: D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]
values
def values(
...
)
D.values() -> an object providing a view on D's values