py_canoe.core.configuration.Configuration

def add_database(self, database_file: str, network: str | int) -> bool:
    """
    Add a database file to the specified channel/network

    Args:
        database_file (str): Path to the database file
        network (str | int): Network name or channel number

    Returns:
        bool: True for success, False for failure
    """
    try:
        if self.app.measurement.running:
            logger.warning("Cannot add database while measurement is running. Please stop the measurement first.")
            return False
        else:
            # Determine whether it is a network name or a channel number
            if isinstance(network, str):
                # Get the channel number based on the network name
                for bus in self.simulation_setup.buses.item():
                    bus: Bus
                    bus_name = getattr(bus, 'name', '')
                    if bus_name == network:
                        try:
                            bus.databases.add(database_file)
                        except AttributeError:
                            return self._add_database_to_channel(database_file, 1)
                        logger.info(f'Database "{database_file}" added successfully to network "{network}".')
                        return True
                logger.warning(f'Network "{network}" not found. Cannot add database.')
                return False
            elif isinstance(network, int):
                for bus in self.simulation_setup.buses.item():
                    bus: Bus
                    channels = bus.channels.item()
                    logger.info(f"Checking bus channels: {[channel.number for channel in channels]}")
                    if not channels:
                        return self._add_database_to_channel(database_file, network)
                    for channel in channels:
                        channel: Channel
                        if channel.number == network:
                            return self._add_database_to_channel(database_file, network)
                return self._add_database_to_channel(database_file, network)
            else:
                logger.warning(f'Invalid network type: {type(network)}. Must be str (network name) or int (channel number).')
                return False
    except Exception as e:
        import traceback
        traceback.print_exc()
        logger.error(f"Error adding database '{database_file}': {e}")
        return False

def add_netWork(self, network_name: str, network_type: BusType = BusType.CAN,
                sw_channel: int = 1) -> Bus:
    """Adds a new network to the configuration.

    Args:
        network_name: Name of the new network.
        network_type: Bus type.  Defaults to ``BusType.CAN``.
        sw_channel: Software channel index (1-based) to assign.
            Defaults to 1.

    Returns:
        The newly added :class:`Bus`, or *None* if a network with
        the same name already exists.
    """
    try:
        bus = self.simulation_setup.buses.add(network_name, network_type)
        if bus is None:
            logger.warning(f"Network {network_name} already exists.")
            return None
        # Assign the software channel if it differs from the default (CAN 1)
        if sw_channel != 1 and bus.channels.count > 0:
            bus.channels.remove(1)
        bus.channels.add(1, sw_channel)
        return bus
    except Exception as e:
        logger.error(f"Failed to add network '{network_name}': {e}")
        return None

def fetch_test_modules(self):
    """Get all test modules from all test environments and store them in self.__test_modules."""
    for te_name, te_inst in self.__test_setup_environments.items():
        for tm_name, tm_inst in te_inst.get_all_test_modules().items():
            # A TestSetupItem object that either can be a TestModule object or a TestSetupFolder object.
            # A TestSetupFolder has items that contain nested TestSetupItems.
            self.__test_modules.append({'name': tm_name, 'object': tm_inst, 'environment': te_name})

def fetch_test_units(self):
    """Get all test units from all test configurations and store them in self.__test_units."""
    for tc_name, tc_inst in self.__test_configurations.items():
        for tu_index in range(1, tc_inst.test_units.count + 1):
            tu_inst = tc_inst.test_units.item(tu_index)
            self.__test_units.append({'name': tu_inst.name, 'object': tu_inst, 'test_configuration': tc_name})
    return self.__test_units

def get_compilation_result(self) -> dict[str, object]:
    """Get CAPL compilation result with error details.

    Compiles all CAPL code in the current configuration and returns
    detailed result including success status and error information.

    Returns:
        dict: Dictionary with keys:
            - "success" (bool): True if compilation succeeded, False otherwise
            - "error" (str | None): Error message if compilation failed, None on success

    Example:
        >>> result = config.get_compilation_result()
        >>> if result["success"]:
        ...     print("Compilation OK")
        ... else:
        ...     print(f"Compilation failed: {result['error']}")
    """
    result = self._compile_and_verify_internal()
    if result["success"]:
        logger.info('CAPL compilation succeeded')
    else:
        logger.warning(f'CAPL compilation failed: {result["error"]}')
    return result

def get_test_module_result(self, test_module_name: str, report_timeout: float = 30.0) -> dict:
    """Get test module execution result including report path, test case
    verdicts, and aggregated statistics.

    Should be called after execute_test_module() to retrieve the results.

    Note: This method does NOT depend on the module's started state. It reads
    the verdict and report information directly from the test module object,
    and waits (up to ``report_timeout`` seconds) for the report-generated
    event if it has not fired yet. All returned data are plain Python objects
    (snapshots), not live COM objects, so they remain valid after the CANoe
    session ends.

    Args:
        test_module_name (str): name of the test module.
        report_timeout (float): maximum time in seconds to wait for the
            report-generated event before giving up. Defaults to 30.0.

    Returns:
        dict: A dictionary with keys:

            - "test_module" (str): name of the test module
            - "verdict" (int): overall test module verdict (0-5)
            - "verdict_name" (str): human-readable verdict name
            - "report" (dict): report information with keys:
                - "success" (bool): whether report generation succeeded
                - "source_full_name" (str): XML report path
                - "generated_full_name" (str): HTML report path
            - "test_cases" (list[dict]): list of test case snapshots, each
              with keys: name, title, enabled, verdict, verdict_name
            - "total" (int): total number of test cases
            - "passed" (int): number of passed test cases
            - "failed" (int): number of failed test cases
            - "other" (int): number of test cases with other verdicts
              (NotAvailable, None, Inconclusive, ErrorInTestSystem)
            - "pass_rate" (float): pass rate as a percentage (0.0-100.0),
              0.0 when there are no test cases

    Example:
        >>> result = cfg.get_test_module_result("MyModule")
        >>> print(f"Module: {result['test_module']}")
        >>> print(f"Verdict: {result['verdict_name']}")
        >>> print(f"Pass rate: {result['pass_rate']:.1f}%")
        >>> print(f"Passed: {result['passed']}/{result['total']}")
        >>> print(f"Report: {result['report']['generated_full_name']}")
        >>> for tc in result['test_cases']:
        ...     print(f"  {tc['name']}: {tc['verdict_name']}")
    """
    try:
        tm_obj = self._find_test_module(test_module_name)
        if tm_obj is None:
            return {}

        # Wait for the report-generated event (bounded by report_timeout) so
        # the report paths are populated. If it already fired, this returns
        # immediately. We do NOT gate on TM_STARTED, because the module may
        # have been stopped (which resets TM_STARTED) by the time results
        # are requested. Use a monotonic clock so the wait is not
        # skewed by GIL/thread scheduling.
        deadline = time.monotonic() + report_timeout
        while not tm_obj.test_module_events.TM_REPORT_GENERATED and time.monotonic() < deadline:
            wait(0.01)
        if not tm_obj.test_module_events.TM_REPORT_GENERATED:
            logger.warning(
                f'Test module "{test_module_name}" report was not generated '
                f'within {report_timeout}s; report paths may be empty.'
            )

        # overall verdict
        verdict = tm_obj.verdict
        verdict_name = tm_obj.VALUE_TABLE_VERDICT.get(verdict, "Unknown")

        # report information from event sink
        report_info = tm_obj.test_module_events.TEST_REPORT_INFORMATION
        report = {
            "success": report_info.get("success", False),
            "source_full_name": report_info.get("source_full_name", ""),
            "generated_full_name": report_info.get("generated_full_name", ""),
        }

        # Convert live COM objects to plain dicts so the data survives
        # after the CANoe session ends and COM objects become invalid.
        raw_cases = tm_obj.get_all_test_cases()
        test_cases = [tc.to_dict() for tc in raw_cases.values()]

        # Aggregate statistics
        total = len(test_cases)
        passed = sum(1 for tc in test_cases if tc["verdict"] == 1)
        failed = sum(1 for tc in test_cases if tc["verdict"] == 2)
        other = total - passed - failed
        pass_rate = (passed / total * 100.0) if total > 0 else 0.0

        return {
            "test_module": test_module_name,
            "verdict": verdict,
            "verdict_name": verdict_name,
            "report": report,
            "test_cases": test_cases,
            "total": total,
            "passed": passed,
            "failed": failed,
            "other": other,
            "pass_rate": pass_rate,
        }

    except Exception as e:
        logger.error(f'failed to get test module result for "{test_module_name}": {e}')
        return {}

def remove_all_networks(self) -> int:
    """Remove ALL networks from the simulation setup.

    CANoe requires at least one bus, so this will leave one bus
    remaining.  Returns the number of networks actually removed.
    """
    count = 0
    names = []
    for bus in self.simulation_setup.buses.item():
        names.append(bus.name)
    # Skip the last one — CANoe requires at least one bus
    for name in names[:-1]:
        if self.remove_netWork(name):
            count += 1
    logger.info(f"Removed {count} network(s).")
    return count

def remove_netWork(self, name: str) -> bool:
    """Remove a network by name.

    Args:
        name: Network name to remove.

    Returns:
        True if found and removed.
    """
    try:
        self.simulation_setup.buses.remove(name=name)
        logger.info(f"Network '{name}' removed.")
        return True
    except Exception as e:
        logger.error(f"Failed to remove network '{name}': {e}")
        return False

def run_compilation(self) -> bool:
    """Run CAPL compilation and return success status.

    Compiles all CAPL code in the current configuration.
    Use get_compilation_result() if you need error details.

    Returns:
        bool: True if compilation succeeded, False otherwise

    Example:
        >>> if config.run_compilation():
        ...     print("Compilation OK")
    """
    result = self._compile_and_verify_internal()
    if result["success"]:
        logger.info('CAPL compilation passed')
    else:
        logger.warning(f'CAPL compilation failed: {result["error"]}')
    return result["success"]