mbdlyb.functional.diagnoser =========================== .. py:module:: mbdlyb.functional.diagnoser .. autoapi-nested-parse:: 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 ------- .. autoapisummary:: mbdlyb.functional.diagnoser.Diagnoser mbdlyb.functional.diagnoser.SystemCondition mbdlyb.functional.diagnoser.SystemState Module Contents --------------- .. py:class:: 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. :ivar _net: Active functional model cluster. :vartype _net: Cluster :ivar _orig_net: Original cluster before filtering. :vartype _orig_net: Cluster :ivar _reasoner: Reasoning engine for probabilistic inference. :vartype _reasoner: MBDReasoner :ivar _entropy_limit: Maximum number of hardware nodes to consider for entropy computation. :vartype _entropy_limit: int :ivar _inference_cache: Cached inference results. :vartype _inference_cache: dict :ivar _system_state: Current system state and conditions. :vartype _system_state: SystemState :ivar _operating_modes: Active operating mode settings. :vartype _operating_modes: dict[OperatingMode, str] :ivar _diagnostic_tests: Available diagnostic tests. :vartype _diagnostic_tests: set[DiagnosticTest] :ivar _diagnostic_test_results: Test result nodes. :vartype _diagnostic_test_results: set[DiagnosticTestResult] :ivar _performed_diagnostic_tests: History of performed tests. :vartype _performed_diagnostic_tests: list[DiagnosticTest] :ivar _computed_next_diagnostic_tests: Next recommended tests. :vartype _computed_next_diagnostic_tests: Optional[pd.DataFrame] :ivar _total_diagnostic_cost: Cumulative cost of diagnosis. :vartype _total_diagnostic_cost: float :ivar _loss_function: Custom loss function for test selection. :vartype _loss_function: Callable .. py:method:: 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. :param evidence: Evidence items to add. :type evidence: Evidence :returns: None .. py:method:: add_operating_mode(operating_modes: dict[mbdlyb.functional.OperatingMode, str]) Select operating modes and update the active network and reasoner. :param operating_modes: Operating mode selectors mapped to selected states. :type operating_modes: dict[OperatingMode, str] :returns: None .. py:method:: compute_entropy_and_cost_of_diagnostic_tests(evidence: Optional[mbdlyb.Evidence] = None, limit: int = 0) -> pandas.DataFrame Compute entropy, information gain, and cost for each available test. :param evidence: Hypothetical evidence used for entropy calculations. Defaults to None. :type evidence: Optional[Evidence] :param limit: Maximum number of inferred hardware nodes considered; zero uses the configured entropy limit. Defaults to 0. :type limit: int :returns: Candidate tests with entropy, information gain, and cost columns. :rtype: pd.DataFrame .. py:method:: 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. :rtype: tuple[np.ndarray, np.ndarray, np.ndarray] .. py:method:: 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. :param evidence: Evidence forwarded to candidate scoring. Defaults to None. :type evidence: dict[str, str] :param limit: Hardware-node limit forwarded to candidate scoring. Defaults to None. :type limit: int :returns: Best-ranked test, or None when no candidate is available. :rtype: DiagnosticTest .. py:method:: compute_next_diagnostic_tests(evidence: Optional[mbdlyb.Evidence] = None, limit: int = 0) -> pandas.DataFrame Compute and rank the next diagnostic tests to perform. :param evidence: Hypothetical evidence used in scoring. Defaults to None. :type evidence: Optional[Evidence] :param limit: Maximum number of inferred hardware nodes considered. Defaults to 0. :type limit: int :returns: Ranked candidate test table. :rtype: pd.DataFrame .. py:method:: drop_evidence(*nodes: mbdlyb.functional.base.FunctionalNode) Drop evidence for nodes and unperform associated diagnostic tests. :param \*nodes: Nodes whose evidence should be removed. :type \*nodes: FunctionalNode :returns: None .. py:method:: get_direct_observables() -> pandas.DataFrame Return direct observables that do not currently have evidence. :returns: Sorted table of unobserved direct observables. :rtype: pd.DataFrame .. py:method:: get_operating_modes() -> pandas.DataFrame Return operating mode selectors that do not currently have evidence. :returns: Sorted table of unobserved operating mode selectors. :rtype: pd.DataFrame .. py:method:: get_unperformed_diagnostic_tests() -> list[mbdlyb.functional.base.DiagnosticTest] Return diagnostic tests that have not been performed. :returns: Available unperformed tests. :rtype: list[DiagnosticTest] .. py:method:: infer(*nodes: mbdlyb.functional.base.FunctionalNode) -> pandas.DataFrame Infer hardware-state probabilities and cache the result. :param \*nodes: Optional nodes to infer. With no nodes, the reasoner's default targets are used. :type \*nodes: FunctionalNode :returns: Posterior distributions ordered by rounded health probability and node name. :rtype: pd.DataFrame .. py:method:: initialise_system_state(state: Optional[dict[str, SystemCondition]] = None) Initialize system-state tracking with the supplied conditions. :param state: Conditions keyed by name. Defaults to None, which initializes an empty state. :type state: Optional[dict[str, SystemCondition]] :returns: None .. py:method:: perform_diagnostic_test(diagnostic_test: mbdlyb.functional.base.DiagnosticTest, evidence: Optional[mbdlyb.Evidence] = None) Record a performed test, update system state, and optionally add evidence. :param diagnostic_test: Test to record as performed. :type diagnostic_test: DiagnosticTest :param evidence: Evidence produced by the test. Defaults to None. :type evidence: Optional[Evidence] :returns: None .. py:method:: reset(reasoner_klass: Optional[Type[mbdlyb.MBDReasoner]] = None, **kwargs) Reset diagnostic state and initialize the inference targets. :param reasoner_klass: Reasoner class to use. If omitted, the currently active reasoner class is reused. :type reasoner_klass: Optional[Type[MBDReasoner]] :param \*\*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. .. py:method:: set_entropy_limit(limit: int) Set the maximum number of hardware nodes used for entropy calculations. :param limit: Maximum number of inferred hardware nodes; zero disables the limit. :type limit: int :returns: None .. py:method:: set_loss_function(loss_function: Callable[[float, float], float]) Set the function used to rank diagnostic test candidates. :param loss_function: Function used to combine normalized entropy and cost values. :type loss_function: Callable[[float, float], float] :returns: None .. note:: The callable annotation specifies two numeric arguments, while ranking currently invokes the function with a candidate row as one argument. .. py:method:: 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. :rtype: pd.DataFrame .. py:method:: 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. :param diagnostic_test: Test to remove. :type diagnostic_test: DiagnosticTest :returns: None .. py:property:: net :type: mbdlyb.MBDNet Return the active functional network. :returns: Network currently used for diagnosis. :rtype: MBDNet .. py:property:: reasoner :type: mbdlyb.MBDReasoner Return the reasoning engine used for inference. :returns: Active inference engine. :rtype: MBDReasoner .. py:property:: total_diagnostic_cost Return the cumulative cost of performed diagnostic tests. :returns: Total diagnostic cost accumulated so far. :rtype: float .. py:class:: 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. :ivar name: Name of the condition. :vartype name: str :ivar setting: Current state of the condition (true/false). :vartype setting: bool :ivar _cost_to_true: Costs by type to set condition to true. :vartype _cost_to_true: dict[str, float] :ivar _cost_to_false: Costs by type to set condition to false. :vartype _cost_to_false: dict[str, float] .. py:method:: get_cost(required_state: bool) -> float Get the cost to achieve the required state from current setting. :param required_state: The desired state. :type required_state: bool :returns: Cost to achieve the required state (0 if already in that state). :rtype: float .. py:property:: cost_to_false :type: 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. :rtype: float .. py:property:: cost_to_true :type: 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. :rtype: float .. py:attribute:: name :type: str .. py:attribute:: setting :type: bool :value: False .. py:class:: 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. :ivar conditions: Named system conditions. :vartype conditions: dict[str, SystemCondition] .. py:method:: 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. :param diagnostic_test: Test whose cost should be calculated. :type diagnostic_test: DiagnosticTest :returns: Total cost to perform the test. :rtype: float .. py:method:: get_condition(name: str) Get a system condition by name. :param name: Name of the condition. :type name: str :returns: The current method body does not perform the lookup. :rtype: None .. py:method:: insert_condition(system_condition: SystemCondition) Add a system condition. :param system_condition: Condition to add. :type system_condition: SystemCondition :returns: The current method body does not insert the condition. :rtype: None .. py:method:: insert_conditions(system_conditions: list[SystemCondition]) Add multiple system conditions. :param system_conditions: Conditions to add. :type system_conditions: list[SystemCondition] :returns: The current method body does not insert the conditions. :rtype: None .. py:method:: prepare_for_diagnostic_test(diagnostic_test: mbdlyb.functional.base.DiagnosticTest) Update system conditions to satisfy diagnostic test preconditions. :param diagnostic_test: Test whose preconditions should be satisfied. :type diagnostic_test: DiagnosticTest :returns: None .. py:method:: update_condition_setting(name: str, setting: bool) Update the state of a named condition. :param name: Name of the condition to update. :type name: str :param setting: New state value. :type setting: bool :returns: The current method body does not update the condition. :rtype: None .. py:method:: update_conditions_settings(updates: dict[str, bool]) Update states of multiple conditions. :param updates: Mapping of condition names to new states. :type updates: dict[str, bool] :returns: The current method body does not update the conditions. :rtype: None .. py:attribute:: conditions :type: dict[str, SystemCondition]