mbdlyb.functional.diagnoser

Functional diagnostics engine for model-based fault diagnosis.

This module provides diagnostic and system condition management classes:

SystemCondition: Represents system states with associated costs
SystemState: Manages a collection of system conditions
Diagnoser: Main diagnostic reasoning engine for fault localization

The diagnoser supports dynamic diagnostic test selection, cost-aware reasoning, and operating mode management for functional models.

Classes

`Diagnoser`	Main diagnostic reasoning engine for model-based fault localization.
`SystemCondition`	Represents a system condition that can be set and has associated costs.
`SystemState`	Container for system conditions during diagnostic reasoning.

Module Contents

class mbdlyb.functional.diagnoser.Diagnoser(net: mbdlyb.functional.base.Cluster, reasoner_klass: Type[mbdlyb.MBDReasoner], **kwargs)

Main diagnostic reasoning engine for model-based fault localization.

The diagnoser performs inference on functional models to diagnose faults, dynamically selects diagnostic tests based on information gain and cost, and tracks the history and costs of diagnosis.

Variables:

_net (Cluster) – Active functional model cluster.
_orig_net (Cluster) – Original cluster before filtering.
_reasoner (MBDReasoner) – Reasoning engine for probabilistic inference.
_entropy_limit (int) – Maximum number of hardware nodes to consider for entropy computation.
_inference_cache (dict) – Cached inference results.
_system_state (SystemState) – Current system state and conditions.
_operating_modes (dict[OperatingMode, str]) – Active operating mode settings.
_diagnostic_tests (set[DiagnosticTest]) – Available diagnostic tests.
_diagnostic_test_results (set[DiagnosticTestResult]) – Test result nodes.
_performed_diagnostic_tests (list[DiagnosticTest]) – History of performed tests.
_computed_next_diagnostic_tests (Optional[pd.DataFrame]) – Next recommended tests.
_total_diagnostic_cost (float) – Cumulative cost of diagnosis.
_loss_function (Callable) – Custom loss function for test selection.

add_evidence(evidence: mbdlyb.Evidence)

Add evidence and record any diagnostic tests represented by it.

Test-result evidence is grouped by its test and processed as test execution evidence; remaining evidence is added directly to the reasoner.

Parameters:: evidence (Evidence) – Evidence items to add.
Returns:: None

add_operating_mode(operating_modes: dict[mbdlyb.functional.OperatingMode, str])

Select operating modes and update the active network and reasoner.

Parameters:: operating_modes (dict[OperatingMode, str]) – Operating mode selectors mapped to selected states.
Returns:: None

compute_entropy_and_cost_of_diagnostic_tests(evidence: mbdlyb.Evidence | None = None, limit: int = 0) → pandas.DataFrame

Compute entropy, information gain, and cost for each available test.

Parameters:

evidence (Optional[Evidence]) – Hypothetical evidence used for entropy calculations. Defaults to None.
limit (int) – Maximum number of inferred hardware nodes considered; zero uses the configured entropy limit. Defaults to 0.

Returns:

Candidate tests with entropy, information gain, and cost: columns.

Return type:

pd.DataFrame

compute_loss_grid() → tuple[numpy.ndarray, numpy.ndarray, numpy.ndarray]

Compute a grid of entropy, cost, and illustrative loss values.

Returns:

Entropy coordinates, cost: coordinates, and loss matrix.

Return type:

tuple[np.ndarray, np.ndarray, np.ndarray]

compute_next_diagnostic_test(evidence: dict[str, str] = None, limit: int = None) → mbdlyb.functional.base.DiagnosticTest

Return the highest-ranked next diagnostic test, if available.

Parameters:

evidence (dict[str, str]) – Evidence forwarded to candidate scoring. Defaults to None.
limit (int) – Hardware-node limit forwarded to candidate scoring. Defaults to None.

Returns:

Best-ranked test, or None when no candidate is available.

Return type:

DiagnosticTest

compute_next_diagnostic_tests(evidence: mbdlyb.Evidence | None = None, limit: int = 0) → pandas.DataFrame

Compute and rank the next diagnostic tests to perform.

Parameters:

evidence (Optional[Evidence]) – Hypothetical evidence used in scoring. Defaults to None.
limit (int) – Maximum number of inferred hardware nodes considered. Defaults to 0.

Returns:

Ranked candidate test table.

Return type:

pd.DataFrame

drop_evidence(*nodes: mbdlyb.functional.base.FunctionalNode)

Drop evidence for nodes and unperform associated diagnostic tests.

Parameters:: *nodes (FunctionalNode) – Nodes whose evidence should be removed.
Returns:: None

get_direct_observables() → pandas.DataFrame

Return direct observables that do not currently have evidence.

Returns:: Sorted table of unobserved direct observables.
Return type:: pd.DataFrame

get_operating_modes() → pandas.DataFrame

Return operating mode selectors that do not currently have evidence.

Returns:: Sorted table of unobserved operating mode selectors.
Return type:: pd.DataFrame

get_unperformed_diagnostic_tests() → list[mbdlyb.functional.base.DiagnosticTest]

Return diagnostic tests that have not been performed.

Returns:: Available unperformed tests.
Return type:: list[DiagnosticTest]

infer(*nodes: mbdlyb.functional.base.FunctionalNode) → pandas.DataFrame

Infer hardware-state probabilities and cache the result.

Parameters:

*nodes (FunctionalNode) – Optional nodes to infer. With no nodes, the reasoner’s default targets are used.

Returns:

Posterior distributions ordered by rounded health: probability and node name.

Return type:

pd.DataFrame

initialise_system_state(state: dict[str, SystemCondition] | None = None)

Initialize system-state tracking with the supplied conditions.

Parameters:: state (Optional[dict[str, SystemCondition]]) – Conditions keyed by name. Defaults to None, which initializes an empty state.
Returns:: None

perform_diagnostic_test(diagnostic_test: mbdlyb.functional.base.DiagnosticTest, evidence: mbdlyb.Evidence | None = None)

Record a performed test, update system state, and optionally add evidence.

Parameters:

diagnostic_test (DiagnosticTest) – Test to record as performed.
evidence (Optional[Evidence]) – Evidence produced by the test. Defaults to None.

Returns:

None

reset(reasoner_klass: Type[mbdlyb.MBDReasoner] | None = None, **kwargs)

Reset diagnostic state and initialize the inference targets.

Parameters:

reasoner_klass (Optional[Type[MBDReasoner]]) – Reasoner class to use. If omitted, the currently active reasoner class is reused.
**kwargs – Additional keyword arguments passed to the reasoner constructor.

Returns:

None

Raises:

InvalidReasonerError – If no reasoner class is supplied and no active reasoner exists to reuse.

set_entropy_limit(limit: int)

Set the maximum number of hardware nodes used for entropy calculations.

Parameters:: limit (int) – Maximum number of inferred hardware nodes; zero disables the limit.
Returns:: None

set_loss_function(loss_function: Callable[[float, float], float])

Set the function used to rank diagnostic test candidates.

Parameters:: loss_function (Callable[[float, float], float]) – Function used to combine normalized entropy and cost values.
Returns:: None

Note

The callable annotation specifies two numeric arguments, while ranking currently invokes the function with a candidate row as one argument.

sort_next_diagnostic_tests() → pandas.DataFrame

Sort the most recently computed candidate tests by configured loss.

Returns:

Ranked candidates, or an empty DataFrame if none were: computed.

Return type:

pd.DataFrame

unperform_diagnostic_test(diagnostic_test: mbdlyb.functional.base.DiagnosticTest)

Remove a test from the performed-test history and drop its evidence.

If the test is not in the history, this method leaves state unchanged.

Parameters:: diagnostic_test (DiagnosticTest) – Test to remove.
Returns:: None

property net: mbdlyb.MBDNet

Return the active functional network.

Returns:: Network currently used for diagnosis.
Return type:: MBDNet

property reasoner: mbdlyb.MBDReasoner

Return the reasoning engine used for inference.

Returns:: Active inference engine.
Return type:: MBDReasoner

property total_diagnostic_cost

Return the cumulative cost of performed diagnostic tests.

Returns:: Total diagnostic cost accumulated so far.
Return type:: float

class mbdlyb.functional.diagnoser.SystemCondition(name: str, setting: bool = False, cost_to_true: dict[str, float] | float | int = 0, cost_to_false: dict[str, float] | float | int = 0)

Represents a system condition that can be set and has associated costs.

System conditions are states that affect diagnostic test preconditions and influence the overall cost of diagnosis. Each condition can be true or false with potentially different costs for transitioning between states.

Variables:

name (str) – Name of the condition.
setting (bool) – Current state of the condition (true/false).
_cost_to_true (dict[str, float]) – Costs by type to set condition to true.
_cost_to_false (dict[str, float]) – Costs by type to set condition to false.

get_cost(required_state: bool) → float

Get the cost to achieve the required state from current setting.

Parameters:: required_state (bool) – The desired state.
Returns:: Cost to achieve the required state (0 if already in that state).
Return type:: float

property cost_to_false: float

Get cost for setting this condition from true to false.

If multiple types of costs are given, returns their sum.

Returns:: Total cost to set to false.
Return type:: float

property cost_to_true: float

Get cost for setting this condition from false to true.

If multiple types of costs are given, returns their sum.

Returns:: Total cost to set to true.
Return type:: float

name: str

setting: bool = False

class mbdlyb.functional.diagnoser.SystemState

Container for system conditions during diagnostic reasoning.

Manages a set of SystemCondition objects and provides methods for updating their states and preparing for diagnostic tests.

Variables:: conditions (dict[str, SystemCondition]) – Named system conditions.

calculate_cost_of_diagnostic_test(diagnostic_test: mbdlyb.functional.base.DiagnosticTest)

Compute the total cost of a diagnostic test given current system state.

Includes the fixed cost of the test plus any costs needed to satisfy its preconditions from the current system state.

Parameters:: diagnostic_test (DiagnosticTest) – Test whose cost should be calculated.
Returns:: Total cost to perform the test.
Return type:: float

get_condition(name: str)

Get a system condition by name.

Parameters:: name (str) – Name of the condition.
Returns:: The current method body does not perform the lookup.
Return type:: None

insert_condition(system_condition: SystemCondition)

Add a system condition.

Parameters:: system_condition (SystemCondition) – Condition to add.
Returns:: The current method body does not insert the condition.
Return type:: None

insert_conditions(system_conditions: list[SystemCondition])

Add multiple system conditions.

Parameters:: system_conditions (list[SystemCondition]) – Conditions to add.
Returns:: The current method body does not insert the conditions.
Return type:: None

prepare_for_diagnostic_test(diagnostic_test: mbdlyb.functional.base.DiagnosticTest)

Update system conditions to satisfy diagnostic test preconditions.

Parameters:: diagnostic_test (DiagnosticTest) – Test whose preconditions should be satisfied.
Returns:: None

update_condition_setting(name: str, setting: bool)

Update the state of a named condition.

Parameters:

name (str) – Name of the condition to update.
setting (bool) – New state value.

Returns:

The current method body does not update the condition.

Return type:

None

update_conditions_settings(updates: dict[str, bool])

Update states of multiple conditions.

Parameters:: updates (dict[str, bool]) – Mapping of condition names to new states.
Returns:: The current method body does not update the conditions.
Return type:: None

conditions: dict[str, SystemCondition]