Skip to content

Services (Offline)

Each UDS service is implemented as a class in pyudskit.services. Every class can:

  • build_request(...)
  • parse_response(...)
  • validate(...)

Session Management

  • DiagnosticSessionControl
  • ECUReset
  • SecurityAccess
  • CommunicationControl
  • Authentication
  • TesterPresent
  • AccessTimingParameter
  • SecuredDataTransmission
  • ControlDTCSetting
  • ResponseOnEvent
  • LinkControl

Data Transmission

  • ReadDataByIdentifier
  • ReadMemoryByAddress
  • ReadScalingDataByIdentifier
  • ReadDataByPeriodicIdentifier
  • DynamicallyDefineDataIdentifier
  • WriteDataByIdentifier
  • WriteMemoryByAddress

DTC Management

  • ClearDiagnosticInformation
  • ReadDTCInformation

I/O Control

  • InputOutputControlByIdentifier

Routine Control

  • RoutineControl

Upload / Download

  • RequestDownload
  • RequestUpload
  • TransferData
  • RequestTransferExit
  • RequestFileTransfer

API Docs

pyudskit.services

pyudskit.services.session_management.diagnostic_session_control.DiagnosticSessionControl

Bases: UDSService

Source code in pyudskit/services/session_management/diagnostic_session_control.py 
 7
 8
 9
10
11
12
13
14
15
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
class DiagnosticSessionControl(UDSService):
    SID = 0x10
    NAME = "DiagnosticSessionControl"
    GROUP = "Session Management"
    POSITIVE_RESPONSE_SID = 0x50

    SESSIONS = {
        "default": 0x01,
        "programming": 0x02,
        "extended": 0x03,
    }

    def build_request(self, session: str | int = "extended") -> ServiceResult:
        ok, errors = self.validate(session=session)
        if not ok:
            return self._result(False, b"", {}, "Invalid session", errors)
        sub = session if isinstance(session, int) else self.SESSIONS[session]
        data = bytes([self.SID, sub])
        return self._result(True, data, {"session": session}, "DiagnosticSessionControl request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        fields: dict = {}
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if not self.is_positive_response(data):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        if len(data) < 2:
            return self._result(False, data, {}, "Response too short", ["response too short"])
        session_byte = data[1]
        session_name = {
            0x01: "defaultDiagnosticSession",
            0x02: "programmingSession",
            0x03: "extendedDiagnosticSession",
        }.get(session_byte, f"0x{session_byte:02X}")
        fields["session"] = session_name
        if len(data) >= 4:
            fields["p2ServerMax_ms"] = int.from_bytes(data[2:4], "big")
        if len(data) >= 6:
            fields["p2StarServerMax_ms"] = int.from_bytes(data[4:6], "big")
        return self._result(True, data, fields, "DiagnosticSessionControl response")

    def validate(self, session: str | int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if isinstance(session, int):
            if not (0 <= session <= 0xFF):
                errors.append("session subFunction must be 0..255")
        else:
            if session not in self.SESSIONS:
                errors.append("session must be one of: default, programming, extended")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.ecu_reset.ECUReset

Bases: UDSService

Source code in pyudskit/services/session_management/ecu_reset.py 
 7
 8
 9
10
11
12
13
14
15
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
class ECUReset(UDSService):
    SID = 0x11
    NAME = "ECUReset"
    GROUP = "Session Management"

    RESET_TYPES = {
        "hard": 0x01,
        "keyOffOn": 0x02,
        "soft": 0x03,
    }

    def build_request(self, reset_type: str = "hard") -> ServiceResult:
        ok, errors = self.validate(reset_type=reset_type)
        if not ok:
            return self._result(False, b"", {}, "Invalid reset type", errors)
        sub = self.RESET_TYPES[reset_type]
        data = bytes([self.SID, sub])
        return self._result(True, data, {"reset_type": reset_type}, "ECUReset request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if len(data) < 2 or data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"reset_type": data[1]}, "ECUReset response")

    def validate(self, reset_type: str, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if reset_type not in self.RESET_TYPES:
            errors.append("reset_type must be hard, keyOffOn, or soft")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.security_access.SecurityAccess

Bases: UDSService

Source code in pyudskit/services/session_management/security_access.py 
 7
 8
 9
10
11
12
13
14
15
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
class SecurityAccess(UDSService):
    SID = 0x27
    NAME = "SecurityAccess"
    GROUP = "Session Management"

    def build_request_seed(self, level: int = 1) -> ServiceResult:
        ok, errors = self.validate(level=level)
        if not ok:
            return self._result(False, b"", {}, "Invalid security level", errors)
        sub = (level * 2) - 1
        data = bytes([self.SID, sub])
        return self._result(True, data, {"level": level, "type": "seed"}, "SecurityAccess seed request")

    def build_send_key(self, level: int, key_bytes: bytes) -> ServiceResult:
        ok, errors = self.validate(level=level)
        if not ok:
            return self._result(False, b"", {}, "Invalid security level", errors)
        sub = level * 2
        data = bytes([self.SID, sub]) + key_bytes
        return self._result(True, data, {"level": level, "type": "key"}, "SecurityAccess send key")

    def build_request(self, level: int = 1, key_bytes: bytes = b"") -> ServiceResult:
        if key_bytes:
            return self.build_send_key(level, key_bytes)
        return self.build_request_seed(level)

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        if len(data) < 2:
            return self._result(False, data, {}, "Response too short", ["response too short"])
        sub = data[1]
        fields = {"subfunction": sub}
        if sub % 2 == 1 and len(data) > 2:
            fields["seed_hex"] = bytes_to_hex(data[2:])
        return self._result(True, data, fields, "SecurityAccess response")

    def parse_seed(self, response_hex: str) -> bytes:
        data = parse_hex(response_hex)
        if len(data) >= 3 and data[0] == (self.SID + 0x40):
            return data[2:]
        return b""

    def validate(self, level: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not isinstance(level, int) or not (1 <= level <= 7):
            errors.append("level must be 1..7")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.communication_control.CommunicationControl

Bases: UDSService

Source code in pyudskit/services/session_management/communication_control.py 
 7
 8
 9
10
11
12
13
14
15
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
class CommunicationControl(UDSService):
    SID = 0x28
    NAME = "CommunicationControl"
    GROUP = "Session Management"

    CONTROL_TYPES = {
        "enableRxAndTx": 0x00,
        "enableRxAndDisableTx": 0x01,
        "disableRxAndEnableTx": 0x02,
        "disableRxAndTx": 0x03,
    }

    COMM_TYPES = {
        "normalCommunication": 0x01,
        "nmCommunication": 0x02,
        "networkManagement": 0x03,
    }

    def build_request(self, control_type: str = "disableRxAndTx", comm_type: str = "normalCommunication") -> ServiceResult:
        ok, errors = self.validate(control_type=control_type, comm_type=comm_type)
        if not ok:
            return self._result(False, b"", {}, "Invalid communication control", errors)
        data = bytes([self.SID, self.CONTROL_TYPES[control_type], self.COMM_TYPES[comm_type]])
        return self._result(True, data, {"control_type": control_type, "comm_type": comm_type}, "CommunicationControl request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"control_type": data[1] if len(data) > 1 else None}, "CommunicationControl response")

    def validate(self, control_type: str, comm_type: str = "normalCommunication", **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if control_type not in self.CONTROL_TYPES:
            errors.append("invalid control_type")
        if comm_type not in self.COMM_TYPES:
            errors.append("invalid comm_type")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.authentication.Authentication

Bases: UDSService

Source code in pyudskit/services/session_management/authentication.py 
 7
 8
 9
10
11
12
13
14
15
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
class Authentication(UDSService):
    SID = 0x29
    NAME = "Authentication"
    GROUP = "Session Management"

    SUBFUNCTIONS = {
        0x00: "deAuthenticate",
        0x01: "verifyCertificateUnidirectional",
        0x02: "verifyCertificateBidirectional",
        0x03: "proofOfOwnership",
        0x04: "transmitCertificate",
        0x05: "requestChallengeForAuthentication",
        0x06: "verifyProofOfOwnershipUnidirectional",
        0x07: "verifyProofOfOwnershipBidirectional",
        0x08: "authenticationConfiguration",
    }

    def build_request(self, subfunction: int, data_hex: str = "") -> ServiceResult:
        ok, errors = self.validate(subfunction=subfunction)
        if not ok:
            return self._result(False, b"", {}, "Invalid subfunction", errors)
        data = bytes([self.SID, subfunction]) + parse_hex(data_hex)
        return self._result(True, data, {"subfunction": subfunction}, "Authentication request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"subfunction": data[1] if len(data) > 1 else None}, "Authentication response")

    def validate(self, subfunction: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if subfunction not in self.SUBFUNCTIONS:
            errors.append("invalid subfunction")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.tester_present.TesterPresent

Bases: UDSService

Source code in pyudskit/services/session_management/tester_present.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
class TesterPresent(UDSService):
    SID = 0x3E
    NAME = "TesterPresent"
    GROUP = "Session Management"

    def build_request(self, suppress_response: bool = True) -> ServiceResult:
        sub = 0x80 if suppress_response else 0x00
        data = bytes([self.SID, sub])
        return self._result(True, data, {"suppress_response": suppress_response}, "TesterPresent request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"subfunction": data[1] if len(data) > 1 else None}, "TesterPresent response")

    def validate(self, **kwargs) -> tuple[bool, list[str]]:
        return True, []

pyudskit.services.session_management.access_timing_parameter.AccessTimingParameter

Bases: UDSService

Source code in pyudskit/services/session_management/access_timing_parameter.py 
 7
 8
 9
10
11
12
13
14
15
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
class AccessTimingParameter(UDSService):
    SID = 0x83
    NAME = "AccessTimingParameter"
    GROUP = "Session Management"

    SUBFUNCTIONS = {
        0x01: "readExtendedTimingParameterSet",
        0x02: "setTimingParametersToDefaultValues",
        0x03: "readCurrentlyActiveTimingParameters",
        0x04: "setTimingParametersToGivenValues",
    }

    def build_request(self, subfunction: int, timing_params: bytes = b"") -> ServiceResult:
        ok, errors = self.validate(subfunction=subfunction)
        if not ok:
            return self._result(False, b"", {}, "Invalid subfunction", errors)
        data = bytes([self.SID, subfunction]) + timing_params
        return self._result(True, data, {"subfunction": subfunction}, "AccessTimingParameter request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"subfunction": data[1] if len(data) > 1 else None, "data_hex": parse_hex(response_hex)[2:].hex().upper()}, "AccessTimingParameter response")

    def validate(self, subfunction: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if subfunction not in self.SUBFUNCTIONS:
            errors.append("invalid subfunction")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.secured_data_transmission.SecuredDataTransmission

Bases: UDSService

Source code in pyudskit/services/session_management/secured_data_transmission.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class SecuredDataTransmission(UDSService):
    SID = 0x84
    NAME = "SecuredDataTransmission"
    GROUP = "Session Management"

    def build_request(self, data_hex: str = "") -> ServiceResult:
        data = bytes([self.SID]) + parse_hex(data_hex)
        return self._result(True, data, {"data_hex": data_hex}, "SecuredDataTransmission request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"data_hex": data[1:].hex().upper()}, "SecuredDataTransmission response")

    def validate(self, **kwargs) -> tuple[bool, list[str]]:
        return True, []

pyudskit.services.session_management.control_dtc_setting.ControlDTCSetting

Bases: UDSService

Source code in pyudskit/services/session_management/control_dtc_setting.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
class ControlDTCSetting(UDSService):
    SID = 0x85
    NAME = "ControlDTCSetting"
    GROUP = "Session Management"

    SETTINGS = {"on": 0x01, "off": 0x02}

    def build_request(self, setting: str = "off", dtc_group: bytes = b"\xFF\xFF\xFF") -> ServiceResult:
        ok, errors = self.validate(setting=setting)
        if not ok:
            return self._result(False, b"", {}, "Invalid setting", errors)
        data = bytes([self.SID, self.SETTINGS[setting]]) + dtc_group
        return self._result(True, data, {"setting": setting}, "ControlDTCSetting request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"setting": data[1] if len(data) > 1 else None}, "ControlDTCSetting response")

    def validate(self, setting: str, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if setting not in self.SETTINGS:
            errors.append("setting must be on or off")
        return (len(errors) == 0, errors)

pyudskit.services.session_management.response_on_event.ResponseOnEvent

Bases: UDSService

Source code in pyudskit/services/session_management/response_on_event.py 
 7
 8
 9
10
11
12
13
14
15
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
class ResponseOnEvent(UDSService):
    SID = 0x86
    NAME = "ResponseOnEvent"
    GROUP = "Session Management"

    EVENT_TYPES = {
        0x00: "stopResponseOnEvent",
        0x01: "onDTCStatusChange",
        0x02: "onTimerInterrupt",
        0x03: "onChangeOfDataIdentifier",
        0x04: "reportActivatedEvents",
        0x05: "startResponseOnEvent",
        0x06: "clearResponseOnEvent",
        0x07: "onComparisonOfValues",
    }

    def build_request(
        self,
        event_type: int,
        event_window: int = 0x02,
        event_type_record: bytes = b"",
        service_to_respond: bytes = b"",
    ) -> ServiceResult:
        ok, errors = self.validate(event_type=event_type)
        if not ok:
            return self._result(False, b"", {}, "Invalid event_type", errors)
        data = bytes([self.SID, event_type, event_window]) + event_type_record + service_to_respond
        return self._result(True, data, {"event_type": event_type}, "ResponseOnEvent request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"event_type": data[1] if len(data) > 1 else None}, "ResponseOnEvent response")

    def validate(self, event_type: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if event_type not in self.EVENT_TYPES:
            errors.append("invalid event_type")
        return (len(errors) == 0, errors)

Bases: UDSService

Source code in pyudskit/services/session_management/link_control.py 
 7
 8
 9
10
11
12
13
14
15
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
class LinkControl(UDSService):
    SID = 0x87
    NAME = "LinkControl"
    GROUP = "Session Management"

    SUBFUNCTIONS = {
        0x01: "verifyBaudrateTransitionWithFixedBaudrate",
        0x02: "verifyBaudrateTransitionWithSpecificBaudrate",
        0x03: "transitionBaudrate",
    }

    FIXED_BAUDRATES = {
        0x01: 9600,
        0x02: 19200,
        0x03: 38400,
        0x04: 57600,
        0x05: 115200,
        0x10: 250000,
        0x11: 500000,
        0x12: 1000000,
    }

    def build_request(self, subfunction: int, baudrate: int = 0x10, specific_baudrate: int | None = None) -> ServiceResult:
        ok, errors = self.validate(subfunction=subfunction)
        if not ok:
            return self._result(False, b"", {}, "Invalid subfunction", errors)
        payload = [self.SID, subfunction]
        if subfunction == 0x02 and specific_baudrate is not None:
            payload.extend(int(specific_baudrate).to_bytes(3, "big"))
        else:
            payload.append(baudrate & 0xFF)
        data = bytes(payload)
        return self._result(True, data, {"subfunction": subfunction}, "LinkControl request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response SID", ["unexpected SID"])
        return self._result(True, data, {"subfunction": data[1] if len(data) > 1 else None}, "LinkControl response")

    def validate(self, subfunction: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if subfunction not in self.SUBFUNCTIONS:
            errors.append("invalid subfunction")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.read_data_by_identifier.ReadDataByIdentifier

Bases: UDSService

Source code in pyudskit/services/data_transmission/read_data_by_identifier.py 
 8
 9
10
11
12
13
14
15
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
class ReadDataByIdentifier(UDSService):
    SID = 0x22
    NAME = "ReadDataByIdentifier"
    GROUP = "Data Transmission"
    POSITIVE_RESPONSE_SID = 0x62

    def build_request(self, dids: list[int]) -> ServiceResult:
        ok, errors = self.validate(dids=dids)
        if not ok:
            return self._result(False, b"", {}, "Invalid DIDs", errors)
        payload = [self.SID]
        for did in dids:
            payload.extend([(did >> 8) & 0xFF, did & 0xFF])
        data = bytes(payload)
        return self._result(True, data, {"dids": dids}, "ReadDataByIdentifier request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != self.POSITIVE_RESPONSE_SID or len(data) < 3:
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        did = (data[1] << 8) | data[2]
        info = COMMON_DIDS.get(did, {})
        fields = {
            "did": did,
            "did_name": info.get("name"),
            "data_hex": bytes_to_hex(data[3:]),
        }
        return self._result(True, data, fields, "ReadDataByIdentifier response")

    def validate(self, dids: list[int], **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not dids:
            errors.append("dids list must not be empty")
        for did in dids:
            if not (0 <= did <= 0xFFFF):
                errors.append(f"invalid DID: {did}")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.read_memory_by_address.ReadMemoryByAddress

Bases: UDSService

Source code in pyudskit/services/data_transmission/read_memory_by_address.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class ReadMemoryByAddress(UDSService):
    SID = 0x23
    NAME = "ReadMemoryByAddress"
    GROUP = "Data Transmission"

    def build_request(self, address: int, length: int, address_length: int = 4, memory_size_length: int = 1) -> ServiceResult:
        ok, errors = self.validate(address=address, length=length)
        if not ok:
            return self._result(False, b"", {}, "Invalid address/length", errors)
        alfid = build_alfid(address_length, memory_size_length)
        data = bytes([self.SID, alfid]) + address.to_bytes(address_length, "big") + length.to_bytes(memory_size_length, "big")
        return self._result(True, data, {"address": address, "length": length}, "ReadMemoryByAddress request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {"data_hex": bytes_to_hex(data[1:])}, "ReadMemoryByAddress response")

    def validate(self, address: int, length: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if address < 0:
            errors.append("address must be >= 0")
        if length <= 0:
            errors.append("length must be > 0")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.read_scaling_data.ReadScalingDataByIdentifier

Bases: UDSService

Source code in pyudskit/services/data_transmission/read_scaling_data.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class ReadScalingDataByIdentifier(UDSService):
    SID = 0x24
    NAME = "ReadScalingDataByIdentifier"
    GROUP = "Data Transmission"

    def build_request(self, did: int) -> ServiceResult:
        ok, errors = self.validate(did=did)
        if not ok:
            return self._result(False, b"", {}, "Invalid DID", errors)
        data = bytes([self.SID, (did >> 8) & 0xFF, did & 0xFF])
        return self._result(True, data, {"did": did}, "ReadScalingDataByIdentifier request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40) or len(data) < 3:
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        did = (data[1] << 8) | data[2]
        info = COMMON_DIDS.get(did, {})
        fields = {"did": did, "did_name": info.get("name"), "scaling_hex": bytes_to_hex(data[3:])}
        return self._result(True, data, fields, "ReadScalingDataByIdentifier response")

    def validate(self, did: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not (0 <= did <= 0xFFFF):
            errors.append("DID must be 0..0xFFFF")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.read_data_by_periodic_id.ReadDataByPeriodicIdentifier

Bases: UDSService

Source code in pyudskit/services/data_transmission/read_data_by_periodic_id.py 
 7
 8
 9
10
11
12
13
14
15
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
class ReadDataByPeriodicIdentifier(UDSService):
    SID = 0x2A
    NAME = "ReadDataByPeriodicIdentifier"
    GROUP = "Data Transmission"

    RATES = {
        "slow": 0x01,
        "medium": 0x02,
        "fast": 0x03,
        "stop": 0x04,
    }

    def build_request(self, rate: str, periodic_dids: list[int]) -> ServiceResult:
        ok, errors = self.validate(rate=rate, periodic_dids=periodic_dids)
        if not ok:
            return self._result(False, b"", {}, "Invalid periodic request", errors)
        payload = [self.SID, self.RATES[rate]]
        for did in periodic_dids:
            payload.append(did & 0xFF)
        data = bytes(payload)
        return self._result(True, data, {"rate": rate, "periodic_dids": periodic_dids}, "ReadDataByPeriodicIdentifier request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {"rate": data[1] if len(data) > 1 else None}, "ReadDataByPeriodicIdentifier response")

    def validate(self, rate: str, periodic_dids: list[int], **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if rate not in self.RATES:
            errors.append("rate must be slow, medium, fast, or stop")
        if not periodic_dids:
            errors.append("periodic_dids must not be empty")
        for did in periodic_dids:
            if not (0 <= did <= 0xFF):
                errors.append("periodic DID must be 1 byte")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.dynamically_define_did.DynamicallyDefineDataIdentifier

Bases: UDSService

Source code in pyudskit/services/data_transmission/dynamically_define_did.py 
 7
 8
 9
10
11
12
13
14
15
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
class DynamicallyDefineDataIdentifier(UDSService):
    SID = 0x2C
    NAME = "DynamicallyDefineDataIdentifier"
    GROUP = "Data Transmission"

    SUBFUNCTIONS = {
        0x01: "defineByIdentifier",
        0x02: "defineByMemoryAddress",
        0x03: "clearDynamicallyDefinedDataIdentifier",
    }

    def build_define_by_identifier(self, dynamic_did: int, source_dids: list[tuple[int, int, int]]) -> ServiceResult:
        payload = [self.SID, 0x01, (dynamic_did >> 8) & 0xFF, dynamic_did & 0xFF]
        for source_did, position, size in source_dids:
            payload.extend([(source_did >> 8) & 0xFF, source_did & 0xFF, position & 0xFF, size & 0xFF])
        data = bytes(payload)
        return self._result(True, data, {"dynamic_did": dynamic_did}, "Define dynamic DID")

    def build_clear(self, dynamic_did: int) -> ServiceResult:
        data = bytes([self.SID, 0x03, (dynamic_did >> 8) & 0xFF, dynamic_did & 0xFF])
        return self._result(True, data, {"dynamic_did": dynamic_did}, "Clear dynamic DID")

    def build_request(self, subfunction: int, **kwargs) -> ServiceResult:
        ok, errors = self.validate(subfunction=subfunction, **kwargs)
        if not ok:
            return self._result(False, b"", {}, "Invalid dynamic DID request", errors)
        if subfunction == 0x01:
            return self.build_define_by_identifier(kwargs["dynamic_did"], kwargs["source_dids"])
        if subfunction == 0x03:
            return self.build_clear(kwargs["dynamic_did"])
        data = bytes([self.SID, subfunction])
        return self._result(True, data, {"subfunction": subfunction}, "Dynamic DID request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        fields = {"subfunction": data[1] if len(data) > 1 else None}
        if len(data) >= 4:
            fields["dynamic_did"] = (data[2] << 8) | data[3]
        return self._result(True, data, fields, "Dynamic DID response")

    def validate(self, subfunction: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if subfunction not in self.SUBFUNCTIONS:
            errors.append("invalid subfunction")
        if subfunction in (0x01, 0x03) and "dynamic_did" not in kwargs:
            errors.append("dynamic_did required")
        if subfunction == 0x01 and "source_dids" not in kwargs:
            errors.append("source_dids required")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.write_data_by_identifier.WriteDataByIdentifier

Bases: UDSService

Source code in pyudskit/services/data_transmission/write_data_by_identifier.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class WriteDataByIdentifier(UDSService):
    SID = 0x2E
    NAME = "WriteDataByIdentifier"
    GROUP = "Data Transmission"

    def build_request(self, did: int, data: bytes | str) -> ServiceResult:
        ok, errors = self.validate(did=did, data=data)
        if not ok:
            return self._result(False, b"", {}, "Invalid DID/data", errors)
        data_bytes = data if isinstance(data, (bytes, bytearray)) else parse_hex(str(data))
        payload = bytes([self.SID, (did >> 8) & 0xFF, did & 0xFF]) + bytes(data_bytes)
        return self._result(True, payload, {"did": did}, "WriteDataByIdentifier request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {"did": (data[1] << 8) | data[2] if len(data) >= 3 else None}, "WriteDataByIdentifier response")

    def validate(self, did: int, data, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not (0 <= did <= 0xFFFF):
            errors.append("DID must be 0..0xFFFF")
        if data is None:
            errors.append("data must not be None")
        return (len(errors) == 0, errors)

pyudskit.services.data_transmission.write_memory_by_address.WriteMemoryByAddress

Bases: UDSService

Source code in pyudskit/services/data_transmission/write_memory_by_address.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class WriteMemoryByAddress(UDSService):
    SID = 0x3D
    NAME = "WriteMemoryByAddress"
    GROUP = "Data Transmission"

    def build_request(self, address: int, data: bytes, address_length: int = 4) -> ServiceResult:
        ok, errors = self.validate(address=address, data=data)
        if not ok:
            return self._result(False, b"", {}, "Invalid address/data", errors)
        alfid = build_alfid(address_length, 0)
        payload = bytes([self.SID, alfid]) + address.to_bytes(address_length, "big") + bytes(data)
        return self._result(True, payload, {"address": address}, "WriteMemoryByAddress request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {}, "WriteMemoryByAddress response")

    def validate(self, address: int, data: bytes, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if address < 0:
            errors.append("address must be >= 0")
        if not data:
            errors.append("data must not be empty")
        return (len(errors) == 0, errors)

pyudskit.services.dtc_management.clear_diagnostic_info.ClearDiagnosticInformation

Bases: UDSService

Source code in pyudskit/services/dtc_management/clear_diagnostic_info.py 
 7
 8
 9
10
11
12
13
14
15
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
class ClearDiagnosticInformation(UDSService):
    SID = 0x14
    NAME = "ClearDiagnosticInformation"
    GROUP = "DTC Management"

    DTC_GROUPS = {
        "all": 0xFFFFFF,
        "powertrain": 0x000000,
        "chassis": 0x400000,
        "body": 0x800000,
        "network": 0xC00000,
    }

    def build_request(self, group: str | int = "all") -> ServiceResult:
        ok, errors = self.validate(group=group)
        if not ok:
            return self._result(False, b"", {}, "Invalid DTC group", errors)
        if isinstance(group, str):
            value = self.DTC_GROUPS[group]
        else:
            value = int(group)
        group_bytes = value.to_bytes(3, "big")
        data = bytes([self.SID]) + group_bytes
        return self._result(True, data, {"group": group}, "ClearDiagnosticInformation request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {}, "ClearDiagnosticInformation response")

    def validate(self, group, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if isinstance(group, str):
            if group not in self.DTC_GROUPS:
                errors.append("invalid DTC group")
        elif not (0 <= int(group) <= 0xFFFFFF):
            errors.append("group must be 0..0xFFFFFF")
        return (len(errors) == 0, errors)

pyudskit.services.dtc_management.read_dtc_information.ReadDTCInformation

Bases: UDSService

Source code in pyudskit/services/dtc_management/read_dtc_information.py 
  8
  9
 10
 11
 12
 13
 14
 15
 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
 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
126
127
128
129
130
131
class ReadDTCInformation(UDSService):
    SID = 0x19
    NAME = "ReadDTCInformation"
    GROUP = "DTC Management"

    SUBFUNCTIONS = {
        0x01: "reportNumberOfDTCByStatusMask",
        0x02: "reportDTCByStatusMask",
        0x03: "reportDTCSnapshotIdentification",
        0x04: "reportDTCSnapshotRecordByDTCNumber",
        0x05: "reportDTCSnapshotRecordByRecordNumber",
        0x06: "reportDTCExtendedDataRecordByDTCNumber",
        0x07: "reportNumberOfDTCBySeverityMaskRecord",
        0x08: "reportDTCBySeverityMaskRecord",
        0x09: "reportSeverityInformationOfDTC",
        0x0A: "reportSupportedDTC",
        0x0B: "reportFirstTestFailedDTC",
        0x0C: "reportFirstConfirmedDTC",
        0x0D: "reportMostRecentTestFailedDTC",
        0x0E: "reportMostRecentConfirmedDTC",
        0x0F: "reportMirrorMemoryDTCByStatusMask",
        0x10: "reportMirrorMemoryDTCExtendedDataRecordByDTCNumber",
        0x11: "reportNumberOfMirrorMemoryDTCByStatusMask",
        0x12: "reportNumberOfEmissionsOBDDTCByStatusMask",
        0x13: "reportEmissionsOBDDTCByStatusMask",
        0x14: "reportDTCFaultDetectionCounter",
        0x15: "reportDTCWithPermanentStatus",
    }

    def build_request_by_status_mask(self, status_mask: int = 0x08) -> ServiceResult:
        return self.build_request(0x02, status_mask=status_mask)

    def build_request_number_by_status(self, status_mask: int = 0x08) -> ServiceResult:
        return self.build_request(0x01, status_mask=status_mask)

    def build_request_snapshot_id(self) -> ServiceResult:
        return self.build_request(0x03)

    def build_request_snapshot_by_dtc(self, dtc: int, record_num: int = 0xFF) -> ServiceResult:
        return self.build_request(0x04, dtc=dtc, record_num=record_num)

    def build_request_snapshot_by_record(self, record_num: int) -> ServiceResult:
        return self.build_request(0x05, record_num=record_num)

    def build_request_extended_by_dtc(self, dtc: int, record_num: int = 0xFF) -> ServiceResult:
        return self.build_request(0x06, dtc=dtc, record_num=record_num)

    def build_request_by_severity(self, severity_mask: int, status_mask: int) -> ServiceResult:
        return self.build_request(0x08, severity_mask=severity_mask, status_mask=status_mask)

    def build_request_supported_dtcs(self) -> ServiceResult:
        return self.build_request(0x0A)

    def build_request_first_failed(self) -> ServiceResult:
        return self.build_request(0x0B)

    def build_request_first_confirmed(self) -> ServiceResult:
        return self.build_request(0x0C)

    def build_request_most_recent_failed(self) -> ServiceResult:
        return self.build_request(0x0D)

    def build_request_most_recent_confirmed(self) -> ServiceResult:
        return self.build_request(0x0E)

    def build_request_fault_counter(self) -> ServiceResult:
        return self.build_request(0x14)

    def build_request_permanent(self) -> ServiceResult:
        return self.build_request(0x15)

    def build_request(self, subfunction: int, **kwargs) -> ServiceResult:
        ok, errors = self.validate(subfunction=subfunction)
        if not ok:
            return self._result(False, b"", {}, "Invalid subfunction", errors)
        payload = [self.SID, subfunction]
        if subfunction in (0x01, 0x02):
            payload.append(kwargs.get("status_mask", 0x08) & 0xFF)
        elif subfunction in (0x04, 0x06):
            dtc = int(kwargs.get("dtc", 0))
            payload.extend(dtc.to_bytes(3, "big"))
            payload.append(kwargs.get("record_num", 0xFF) & 0xFF)
        elif subfunction == 0x05:
            payload.append(kwargs.get("record_num", 0xFF) & 0xFF)
        elif subfunction == 0x08:
            payload.append(kwargs.get("severity_mask", 0) & 0xFF)
            payload.append(kwargs.get("status_mask", 0) & 0xFF)
        data = bytes(payload)
        return self._result(True, data, {"subfunction": subfunction}, "ReadDTCInformation request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40) or len(data) < 2:
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        subfunction = data[1]
        fields: dict = {"subfunction": subfunction}
        payload = data[2:]
        if subfunction == 0x02 and payload:
            fields["dtcs"] = self.parse_dtc_list(payload)
        else:
            fields["data_hex"] = bytes_to_hex(payload)
        return self._result(True, data, fields, "ReadDTCInformation response")

    def parse_dtc_list(self, data: bytes) -> list[dict]:
        results: list[dict] = []
        i = 0
        while i + 3 < len(data):
            dtc_bytes = data[i : i + 3]
            status = data[i + 3]
            _, code = split_dtc(dtc_bytes)
            flags = [name for bit, name in DTC_STATUS_BITS.items() if status & bit]
            results.append({"dtc": code, "status_byte": status, "status_flags": flags})
            i += 4
        return results

    def validate(self, subfunction: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if subfunction not in self.SUBFUNCTIONS:
            errors.append("invalid subfunction")
        return (len(errors) == 0, errors)

pyudskit.services.io_control.input_output_control.InputOutputControlByIdentifier

Bases: UDSService

Source code in pyudskit/services/io_control/input_output_control.py 
 7
 8
 9
10
11
12
13
14
15
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
class InputOutputControlByIdentifier(UDSService):
    SID = 0x2F
    NAME = "InputOutputControlByIdentifier"
    GROUP = "I/O Control"

    CONTROL_OPTIONS = {
        "returnControlToECU": 0x00,
        "resetToDefault": 0x01,
        "freezeCurrentState": 0x02,
        "shortTermAdjustment": 0x03,
    }

    def build_request(
        self,
        did: int,
        control_option: str,
        control_enable_mask: bytes = b"",
        control_value: bytes = b"",
    ) -> ServiceResult:
        ok, errors = self.validate(did=did, control_option=control_option)
        if not ok:
            return self._result(False, b"", {}, "Invalid I/O control request", errors)
        payload = [self.SID, (did >> 8) & 0xFF, did & 0xFF, self.CONTROL_OPTIONS[control_option]]
        data = bytes(payload) + control_enable_mask + control_value
        return self._result(True, data, {"did": did, "control_option": control_option}, "InputOutputControl request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        fields = {}
        if len(data) >= 4:
            fields["did"] = (data[1] << 8) | data[2]
            fields["control_option"] = data[3]
        return self._result(True, data, fields, "InputOutputControl response")

    def validate(self, did: int, control_option: str, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not (0 <= did <= 0xFFFF):
            errors.append("DID must be 0..0xFFFF")
        if control_option not in self.CONTROL_OPTIONS:
            errors.append("invalid control_option")
        return (len(errors) == 0, errors)

pyudskit.services.routine_control.routine_control.RoutineControl

Bases: UDSService

Source code in pyudskit/services/routine_control/routine_control.py 
 7
 8
 9
10
11
12
13
14
15
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
class RoutineControl(UDSService):
    SID = 0x31
    NAME = "RoutineControl"
    GROUP = "Routine Control"

    SUBFUNCTIONS = {"start": 0x01, "stop": 0x02, "result": 0x03}

    KNOWN_ROUTINES = {
        0xFF00: "EraseFlashMemory",
        0xFF01: "CheckFlashMemory",
        0xFF02: "CheckProgrammingDependencies",
        0x0202: "CheckProgrammingPreConditions",
        0x0203: "EraseMemory",
        0xF000: "VariantCoding",
    }

    def build_request(self, routine_id: int, action: str = "start", option_record: bytes = b"") -> ServiceResult:
        ok, errors = self.validate(routine_id=routine_id, action=action)
        if not ok:
            return self._result(False, b"", {}, "Invalid routine request", errors)
        sub = self.SUBFUNCTIONS[action]
        data = bytes([self.SID, sub, (routine_id >> 8) & 0xFF, routine_id & 0xFF]) + option_record
        return self._result(True, data, {"routine_id": routine_id, "action": action}, "RoutineControl request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        fields = {}
        if len(data) >= 4:
            fields["routine_id"] = (data[2] << 8) | data[3]
            fields["action"] = data[1]
            fields["result_hex"] = data[4:].hex().upper() if len(data) > 4 else ""
        return self._result(True, data, fields, "RoutineControl response")

    def validate(self, routine_id: int, action: str, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not (0 <= routine_id <= 0xFFFF):
            errors.append("routine_id must be 0..0xFFFF")
        if action not in self.SUBFUNCTIONS:
            errors.append("action must be start, stop, or result")
        return (len(errors) == 0, errors)

pyudskit.services.upload_download.request_download.RequestDownload

Bases: UDSService

Source code in pyudskit/services/upload_download/request_download.py 
 7
 8
 9
10
11
12
13
14
15
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
class RequestDownload(UDSService):
    SID = 0x34
    NAME = "RequestDownload"
    GROUP = "Upload/Download"

    def build_request(
        self,
        address: int,
        size: int,
        address_length: int = 4,
        memory_size_length: int = 4,
        compression_method: int = 0x00,
        encrypting_method: int = 0x00,
    ) -> ServiceResult:
        ok, errors = self.validate(address=address, size=size)
        if not ok:
            return self._result(False, b"", {}, "Invalid request download", errors)
        dfi = ((compression_method & 0x0F) << 4) | (encrypting_method & 0x0F)
        alfid = build_alfid(address_length, memory_size_length)
        payload = bytes([self.SID, dfi, alfid])
        payload += address.to_bytes(address_length, "big")
        payload += size.to_bytes(memory_size_length, "big")
        return self._result(True, payload, {"address": address, "size": size}, "RequestDownload request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40) or len(data) < 2:
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        fields = {"length_format_identifier": data[1], "max_block_length_hex": data[2:].hex().upper()}
        return self._result(True, data, fields, "RequestDownload response")

    def validate(self, address: int, size: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if address < 0:
            errors.append("address must be >= 0")
        if size <= 0:
            errors.append("size must be > 0")
        return (len(errors) == 0, errors)

pyudskit.services.upload_download.request_upload.RequestUpload

Bases: UDSService

Source code in pyudskit/services/upload_download/request_upload.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class RequestUpload(UDSService):
    SID = 0x35
    NAME = "RequestUpload"
    GROUP = "Upload/Download"

    def build_request(self, address: int, size: int, address_length: int = 4, memory_size_length: int = 4) -> ServiceResult:
        ok, errors = self.validate(address=address, size=size)
        if not ok:
            return self._result(False, b"", {}, "Invalid request upload", errors)
        alfid = build_alfid(address_length, memory_size_length)
        payload = bytes([self.SID, 0x00, alfid])
        payload += address.to_bytes(address_length, "big")
        payload += size.to_bytes(memory_size_length, "big")
        return self._result(True, payload, {"address": address, "size": size}, "RequestUpload request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40) or len(data) < 2:
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        fields = {"length_format_identifier": data[1], "max_block_length_hex": data[2:].hex().upper()}
        return self._result(True, data, fields, "RequestUpload response")

    def validate(self, address: int, size: int, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if address < 0:
            errors.append("address must be >= 0")
        if size <= 0:
            errors.append("size must be > 0")
        return (len(errors) == 0, errors)

pyudskit.services.upload_download.transfer_data.TransferData

Bases: UDSService

Source code in pyudskit/services/upload_download/transfer_data.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
class TransferData(UDSService):
    SID = 0x36
    NAME = "TransferData"
    GROUP = "Upload/Download"

    def build_request(self, block_sequence_counter: int, data: bytes | str) -> ServiceResult:
        ok, errors = self.validate(block_sequence_counter=block_sequence_counter, data=data)
        if not ok:
            return self._result(False, b"", {}, "Invalid transfer data", errors)
        data_bytes = data if isinstance(data, (bytes, bytearray)) else parse_hex(str(data))
        payload = bytes([self.SID, block_sequence_counter & 0xFF]) + bytes(data_bytes)
        return self._result(True, payload, {"block_sequence_counter": block_sequence_counter}, "TransferData request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {"block_sequence_counter": data[1] if len(data) > 1 else None}, "TransferData response")

    def validate(self, block_sequence_counter: int, data, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if not (0 <= block_sequence_counter <= 0xFF):
            errors.append("block_sequence_counter must be 0..255")
        if data is None:
            errors.append("data must not be None")
        return (len(errors) == 0, errors)

pyudskit.services.upload_download.request_transfer_exit.RequestTransferExit

Bases: UDSService

Source code in pyudskit/services/upload_download/request_transfer_exit.py 
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
class RequestTransferExit(UDSService):
    SID = 0x37
    NAME = "RequestTransferExit"
    GROUP = "Upload/Download"

    def build_request(self, transfer_response_parameter: bytes = b"") -> ServiceResult:
        data = bytes([self.SID]) + transfer_response_parameter
        return self._result(True, data, {}, "RequestTransferExit request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        return self._result(True, data, {"data_hex": data[1:].hex().upper()}, "RequestTransferExit response")

    def validate(self, **kwargs) -> tuple[bool, list[str]]:
        return True, []

pyudskit.services.upload_download.request_file_transfer.RequestFileTransfer

Bases: UDSService

Source code in pyudskit/services/upload_download/request_file_transfer.py 
 7
 8
 9
10
11
12
13
14
15
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
class RequestFileTransfer(UDSService):
    SID = 0x38
    NAME = "RequestFileTransfer"
    GROUP = "Upload/Download"

    MODE_OF_OPERATION = {
        0x01: "addFile",
        0x02: "deleteFile",
        0x03: "replaceFile",
        0x04: "readFile",
        0x05: "readDir",
    }

    def build_request(
        self,
        mode: int,
        file_path: str,
        compression_method: int = 0x00,
        encrypting_method: int = 0x00,
        file_size: int = 0,
    ) -> ServiceResult:
        ok, errors = self.validate(mode=mode, file_path=file_path)
        if not ok:
            return self._result(False, b"", {}, "Invalid file transfer", errors)
        dfi = ((compression_method & 0x0F) << 4) | (encrypting_method & 0x0F)
        path_bytes = file_path.encode("ascii", errors="ignore")
        payload = bytes([self.SID, mode, dfi, len(path_bytes)]) + path_bytes
        if file_size:
            payload += int(file_size).to_bytes(4, "big")
        return self._result(True, payload, {"mode": mode, "file_path": file_path}, "RequestFileTransfer request")

    def parse_response(self, response_hex: str) -> ServiceResult:
        data = parse_hex(response_hex)
        if not data:
            return self._result(False, b"", {}, "Empty response", ["empty response"])
        if self.is_negative_response(data):
            return self._result(False, data, {}, "Negative response", ["negative response"])
        if data[0] != (self.SID + 0x40):
            return self._result(False, data, {}, "Unexpected response", ["unexpected response"])
        fields = {"mode": data[1] if len(data) > 1 else None, "data_hex": data[2:].hex().upper()}
        return self._result(True, data, fields, "RequestFileTransfer response")

    def validate(self, mode: int, file_path: str, **kwargs) -> tuple[bool, list[str]]:
        errors: list[str] = []
        if mode not in self.MODE_OF_OPERATION:
            errors.append("invalid mode")
        if not file_path:
            errors.append("file_path required")
        return (len(errors) == 0, errors)