py_canoe.core.networks.Networks

The Networks object represents the networks of CANoe.

Source code in src\py_canoe\core\networks.py 
16
17
18
def __init__(self, app: 'Application'):
    self.com_object = app.com_object.Networks
    self.diagnostic_devices: dict[str, Diagnostic] = dict()

count property

Return the number of networks in the CANoe application.

control_tester_present(diag_ecu_qualifier_name, value)

Enable or Disable tester present.

Source code in src\py_canoe\core\networks.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def control_tester_present(self, diag_ecu_qualifier_name: str, value: bool) -> bool:
    """Enable or Disable tester present."""
    try:
        diag_device: Diagnostic = self.diagnostic_devices.get(diag_ecu_qualifier_name)
        if diag_device:
            if value:
                diag_device.diag_start_tester_present()
                logger.info(f'{diag_ecu_qualifier_name}: Tester Present started ')
            else:
                diag_device.diag_stop_tester_present()
                logger.info(f'{diag_ecu_qualifier_name}: Tester Present stopped ')
            return True
        else:
            logger.warning(f'No diagnostic device found for: {diag_ecu_qualifier_name}')
            return False
    except Exception as e:
        logger.error(f"Error controlling tester present: {e}")
        return False

fetch_diagnostic_devices()

Get all diagnostic devices from the networks and store them in the diagnostic_devices dictionary.

Source code in src\py_canoe\core\networks.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def fetch_diagnostic_devices(self):
    """Get all diagnostic devices from the networks and store them in the diagnostic_devices dictionary."""
    try:
        for i in range(1, self.count + 1):
            network = self.item(i)
            for j in range(1, network.devices.count + 1):
                device = network.devices.item(j)
                try:
                    diagnostic = getattr(device.com_object, 'Diagnostic', None)
                    if diagnostic:
                        self.diagnostic_devices[device.name] = Diagnostic(diagnostic)
                except Exception:
                    pass
    except Exception as e:
        logger.error(f"Error fetching Diagnostic Devices: {e}")
        return None

get_all_network_names()

Get all network names in the CANoe application.

Source code in src\py_canoe\core\networks.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def get_all_network_names(self) -> list[str]:
    """Get all network names in the CANoe application."""
    try:
        count = self.count
        names = []
        for i in range(1, count + 1):
            network = self.item(i)
            name = network.name
            if name is not None:
                names.append(name)
        logger.info(f'networks: {names}')
        return names
    except Exception as e:
        logger.error(f"Error retrieving network names: {e}")
        return []

item(index)

Get a specific network by index (1-based).

Source code in src\py_canoe\core\networks.py 
25
26
27
def item(self, index: int) -> Network:
    """Get a specific network by index (1-based)."""
    return Network(self.com_object.Item(index))

send_diag_request(diag_ecu_qualifier_name, request, request_in_bytes=True, return_sender_name=False, response_in_bytearray=False, timeout=10.0, poll_s=0.01, **kwargs)

Send diagnostic request to the specified ECU and return the response.

Source code in src\py_canoe\core\networks.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def send_diag_request(self, diag_ecu_qualifier_name: str, request: str, request_in_bytes=True, return_sender_name=False, response_in_bytearray=False, timeout: float = 10.0, poll_s: float = 0.01, **kwargs) -> Union[str, dict]:
    """Send diagnostic request to the specified ECU and return the response."""
    try:
        diag_devices = self.diagnostic_devices
        if not diag_devices or diag_ecu_qualifier_name not in diag_devices:
            self.fetch_diagnostic_devices()
            diag_devices = self.diagnostic_devices
        diag_device = diag_devices.get(diag_ecu_qualifier_name)
        if diag_device:
            if request_in_bytes:
                diag_req_in_bytes = bytearray()
                byte_stream = ''.join(request.split(' '))
                for i in range(0, len(byte_stream), 2):
                    diag_req_in_bytes.append(int(byte_stream[i:i + 2], 16))
                diag_request = diag_device.create_request_from_stream(diag_req_in_bytes)
            else:
                diag_request = diag_device.create_request(request, **kwargs)
            diag_request.send()
            logger.info(f'{diag_ecu_qualifier_name}: Diagnostic Request = {request}')
            start_time = time.time()
            while (diag_request.responses.count == 0 and (time.time() - start_time) < timeout):
                if (time.time() - start_time) >= timeout:
                    logger.warning(f"Diagnostic request timed out after {timeout}s: {request}")
                    return f"ERROR: timeout after {timeout}s"
                wait(poll_s)
            if diag_request.responses.count > 0:
                wait(poll_s)
            diag_responses_dict = {}
            diag_response_including_sender_name = {}
            for i in range(1, diag_request.responses.count + 1):
                diag_response = diag_request.responses.item(i)
                diag_response_positive = diag_response.positive
                response_code = diag_response.response_code
                response_sender = diag_response.sender
                response_stream = diag_response.stream
                response_stream_in_str = " ".join(f"{d:02X}" for d in response_stream).upper()
                diag_responses_dict[response_sender] = {
                    "positive": diag_response_positive,
                    "response_code": response_code,
                    "stream": response_stream,
                    "stream_in_str": response_stream_in_str
                }
                if response_in_bytearray:
                    diag_response_including_sender_name[response_sender] = response_stream
                else:
                    diag_response_including_sender_name[response_sender] = response_stream_in_str
                if diag_response_positive:
                    logger.info(f'{response_sender}: Diagnostic Response = {response_stream_in_str}')
                else:
                    logger.info(f'{response_sender}: Diagnostic Response = {response_stream_in_str}')
            if return_sender_name:
                return diag_response_including_sender_name
            if diag_ecu_qualifier_name in diag_response_including_sender_name:
                return diag_response_including_sender_name[diag_ecu_qualifier_name]
            return next(iter(diag_response_including_sender_name.values()), "")
        else:
            logger.warning(f'No responses received for request: {request}')
            return {"error": "No responses received"}
    except com_error as e:
        logger.error("Error sending diagnostic request: %s", e)
        raise
    except Exception as e:
        logger.error(f"Error sending diagnostic request: {e}")
        return {"error": str(e)}