Factory class to reconstruct BlockInterface objects from dictionaries.
This factory handles the recursive instantiation of complex block trees
and ensures that serialized data types (like strings) are converted
back into internal types (like FaultType Enums).
Source code in src/ecc_analyzer/core/block_factory.py
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66 | class BlockFactory:
"""Factory class to reconstruct BlockInterface objects from dictionaries.
This factory handles the recursive instantiation of complex block trees
and ensures that serialized data types (like strings) are converted
back into internal types (like FaultType Enums).
"""
_REGISTRY: dict[str, Type[BlockInterface]] = {
"SumBlock": SumBlock,
"PipelineBlock": PipelineBlock,
"BasicEvent": BasicEvent,
"CoverageBlock": CoverageBlock,
"SplitBlock": SplitBlock,
"TransformationBlock": TransformationBlock,
}
@staticmethod
def from_dict(data: dict[str, Any]) -> BlockInterface:
"""Creates a block instance from a configuration dictionary.
Args:
data (dict[str, Any]): A dictionary containing the block
configuration. Must include a 'type' key.
Returns:
BlockInterface: An initialized instance of the specified block.
Raises:
ValueError: If the 'type' is unknown or required keys are missing.
"""
params = data.copy()
block_type = params.pop("type", None)
if block_type not in BlockFactory._REGISTRY:
raise ValueError(f"Unknown block type: {block_type}")
cls = BlockFactory._REGISTRY[block_type]
if "sub_blocks" in params:
params["sub_blocks"] = [BlockFactory.from_dict(b) for b in params["sub_blocks"]]
fault_keys = ["fault_type", "target_fault", "source_fault", "fault_to_split"]
for key in fault_keys:
if key in params and isinstance(params[key], str):
params[key] = FaultType[params[key]]
if "distribution_rates" in params:
params["distribution_rates"] = {FaultType[k]: v for k, v in params["distribution_rates"].items()}
return cls(**params)
|
from_dict(data)
staticmethod
Creates a block instance from a configuration dictionary.
Parameters:
| Name |
Type |
Description |
Default |
data
|
dict[str, Any]
|
A dictionary containing the block
configuration. Must include a 'type' key.
|
required
|
Returns:
| Name | Type |
Description |
BlockInterface |
BlockInterface
|
An initialized instance of the specified block.
|
Raises:
| Type |
Description |
ValueError
|
If the 'type' is unknown or required keys are missing.
|
Source code in src/ecc_analyzer/core/block_factory.py
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66 | @staticmethod
def from_dict(data: dict[str, Any]) -> BlockInterface:
"""Creates a block instance from a configuration dictionary.
Args:
data (dict[str, Any]): A dictionary containing the block
configuration. Must include a 'type' key.
Returns:
BlockInterface: An initialized instance of the specified block.
Raises:
ValueError: If the 'type' is unknown or required keys are missing.
"""
params = data.copy()
block_type = params.pop("type", None)
if block_type not in BlockFactory._REGISTRY:
raise ValueError(f"Unknown block type: {block_type}")
cls = BlockFactory._REGISTRY[block_type]
if "sub_blocks" in params:
params["sub_blocks"] = [BlockFactory.from_dict(b) for b in params["sub_blocks"]]
fault_keys = ["fault_type", "target_fault", "source_fault", "fault_to_split"]
for key in fault_keys:
if key in params and isinstance(params[key], str):
params[key] = FaultType[params[key]]
if "distribution_rates" in params:
params["distribution_rates"] = {FaultType[k]: v for k, v in params["distribution_rates"].items()}
return cls(**params)
|