<commit_before>#!/usr/bin/env python
import sys
import time
import cancat
import struct
import threading

import cancat.iso_tp as cisotp

NEG_RESP_CODES = {
        0x10:'GeneralReject',
        0x11:'ServiceNotSupported',
        0x12:'SubFunctionNotSupported',
        0x13:'IncorrectMesageLengthOrInvalidFormat',
        0x14:'ResponseTooLong',
        0x21:'BusyRepeatRequest',
        0x22:'ConditionsNotCorrect',
        0x24:'RequestSequenceError',
        0x25:'NoResponseFromSubnetComponent',
        0x26:'FailurePreventsExecutionOfRequestedAction',
        0x31:'RequestOutOfRange',
        0x33:'SecurityAccessDenied',
        0x35:'InvalidKey',
        0x36:'ExceedNumberOfAttempts',
        0x37:'RequiredTimeDelayNotExpired',
        0x70:'UploadDownloadNotAccepted',
        0x71:'TransferDataSuspended',
        0x72:'GeneralProgrammingFailure',
        0x73:'WrongBlockSequenceCounter',
        0x78:'RequestCorrectlyReceived-ResponsePending',
        0x7e:'SubFunctionNotSupportedInActiveSession',
        0x7f:'ServiceNotSupportedInActiveSession',
        0x81:'RpmTooHigh',
        0x82:'RpmTooLow',
        0x83:'EngineIsRunning',
        0x84:'EngineIsNotRunning',
        0x85:'EngineRunTimeTooLow',
        0x86:'TemperatureTooHigh',
        0x87:'TemperatureTooLow',
        0x88:'VehicleSpeedTooHigh',
        0x89:'VehicleSpeedTooLow',
        0x8a:'ThrottlePedalTooHigh',
        0x8b:'ThrottlePedalTooLow',
        0x8c:'TransmissionRangeNotInNeutral',
        0x8d:'TransmissionRangeNotInGear',
        0x8f:'BrakeSwitchsNotClosed',
        0x90:'ShifterLeverNotInPark',
        0x91:'TorqueConverterClutchLocked',
        0x92:'VoltageTooHigh',
        0x93:'VoltageTooLow',
        }

SVC_DIAGNOSTICS_SESSION_CONTROL =           0x10
SVC_ECU_RESET =                             0x11
SVC_CLEAR_DIAGNOSTICS_INFORMATION =         0x14
SVC_READ_DTC_INFORMATION =                  0x19
SVC_READ_DATA_BY_IDENTIFIER =               0x22
SVC_READ_MEMORY_BY_ADDRESS =                0x23
SVC_SECURITY_ACCESS =                       0x27
SVC_READ_DATA_BY_PERIODIC_IDENTIFIER =      0x2a
SVC_DYNAMICALLY_DEFINE_DATA_IDENTIFIER =    0x2c
SVC_WRITE_DATA_BY_IDENTIFIER =              0x2e
SVC_INPUT_OUTPUT_CONTROL_BY_IDENTIFIER =    0x2f
SVC_ROUTINE_CONTROL =                       0x31
SVC_REQUEST_DOWNLOAD =                      0x34
SVC_REQUEST_UPLOAD =                        0x35
SVC_TRANSFER_DATA =                         0x36
SVC_REQUEST_TRANSFER_EXIT =                 0x37
SVC_WRITE_MEMORY_BY_ADDRESS =               0x3d
SVC_TESTER_PRESENT =                        0x3e
SVC_CONTROL_DTC_SETTING =                   0x85

UDS_SVCS = { v:k for k,v in globals().items() if k.startswith('SVC_') }

POS_RESP_CODES = { (k|0x40) : "OK_" + v.lower() for k,v in UDS_SVCS.items() }
POS_RESP_CODES[0] = 'Success'

NEG_RESP_REPR = {}
for k,v in NEG_RESP_CODES.items():
    NEG_RESP_REPR[k] = 'ERR_' + v

RESP_CODES = {}
RESP_CODES.update(NEG_RESP_REPR)
RESP_CODES.update(POS_RESP_CODES)


class NegativeResponseException(Exception):
    def __init__(self, neg_code, svc, msg):
        self.neg_code = neg_code
        self.msg = msg
        self.svc = svc

    def __repr__(self):
        negresprepr = NEG_RESP_CODES.get(self.neg_code)
        return "NEGATIVE RESPONSE to 0x%x (%s):   ERROR 0x%x: %s" % \
            (self.svc, UDS_SVCS.get(self.svc), self.neg_code, negresprepr, self.msg.encode('hex'))

    def __str__(self):
        negresprepr = NEG_RESP_CODES.get(self.neg_code)
        return "NEGATIVE RESPONSE to 0x%x (%s):   ERROR 0x%x: %s   \tmsg: %s" % \
            (self.svc, UDS_SVCS.get(self.svc), self.neg_code, negresprepr, self.msg.encode('hex'))


class UDS:
    def __init__(self, c, tx_arbid, rx_arbid=None, verbose=True, extflag=0):
        self.c = c
        self.verbose = verbose
        self.extflag = extflag

        if rx_arbid == None:
            rx_arbid = tx_arbid + 8 # by UDS spec

        self.tx_arbid = tx_arbid
        self.rx_arbid = rx_arbid

    def xmit_recv(self, data, extflag=0, timeout=3, count=1, service=None):
        msg = self.c.ISOTPxmit_recv(self.tx_arbid, self.rx_arbid, data, extflag, timeout, count, service)

        # check if the response is something we know about and can help out
        if msg != None and len(msg):
            svc = ord(data[0])
            svc_resp = ord(msg[0])
            errcode = ord(msg[2])

            if svc_resp == svc + 0x40:
                if self.verbose: 
                    print "Positive Response!"

            negresprepr = NEG_RESP_CODES.get(errcode)
            if negresprepr != None and svc_resp != svc + 0x40:<commit_msg>With IsoTP update we can no longer rely on messages being at least a certain length<commit_after>#!/usr/bin/env python
import sys
import time
import cancat
import struct
import threading

import cancat.iso_tp as cisotp

NEG_RESP_CODES = {
        0x10:'GeneralReject',
        0x11:'ServiceNotSupported',
        0x12:'SubFunctionNotSupported',
        0x13:'IncorrectMesageLengthOrInvalidFormat',
        0x14:'ResponseTooLong',
        0x21:'BusyRepeatRequest',
        0x22:'ConditionsNotCorrect',
        0x24:'RequestSequenceError',
        0x25:'NoResponseFromSubnetComponent',
        0x26:'FailurePreventsExecutionOfRequestedAction',
        0x31:'RequestOutOfRange',
        0x33:'SecurityAccessDenied',
        0x35:'InvalidKey',
        0x36:'ExceedNumberOfAttempts',
        0x37:'RequiredTimeDelayNotExpired',
        0x70:'UploadDownloadNotAccepted',
        0x71:'TransferDataSuspended',
        0x72:'GeneralProgrammingFailure',
        0x73:'WrongBlockSequenceCounter',
        0x78:'RequestCorrectlyReceived-ResponsePending',
        0x7e:'SubFunctionNotSupportedInActiveSession',
        0x7f:'ServiceNotSupportedInActiveSession',
        0x81:'RpmTooHigh',
        0x82:'RpmTooLow',
        0x83:'EngineIsRunning',
        0x84:'EngineIsNotRunning',
        0x85:'EngineRunTimeTooLow',
        0x86:'TemperatureTooHigh',
        0x87:'TemperatureTooLow',
        0x88:'VehicleSpeedTooHigh',
        0x89:'VehicleSpeedTooLow',
        0x8a:'ThrottlePedalTooHigh',
        0x8b:'ThrottlePedalTooLow',
        0x8c:'TransmissionRangeNotInNeutral',
        0x8d:'TransmissionRangeNotInGear',
        0x8f:'BrakeSwitchsNotClosed',
        0x90:'ShifterLeverNotInPark',
        0x91:'TorqueConverterClutchLocked',
        0x92:'VoltageTooHigh',
        0x93:'VoltageTooLow',
        }

SVC_DIAGNOSTICS_SESSION_CONTROL =           0x10
SVC_ECU_RESET =                             0x11
SVC_CLEAR_DIAGNOSTICS_INFORMATION =         0x14
SVC_READ_DTC_INFORMATION =                  0x19
SVC_READ_DATA_BY_IDENTIFIER =               0x22
SVC_READ_MEMORY_BY_ADDRESS =                0x23
SVC_SECURITY_ACCESS =                       0x27
SVC_READ_DATA_BY_PERIODIC_IDENTIFIER =      0x2a
SVC_DYNAMICALLY_DEFINE_DATA_IDENTIFIER =    0x2c
SVC_WRITE_DATA_BY_IDENTIFIER =              0x2e
SVC_INPUT_OUTPUT_CONTROL_BY_IDENTIFIER =    0x2f
SVC_ROUTINE_CONTROL =                       0x31
SVC_REQUEST_DOWNLOAD =                      0x34
SVC_REQUEST_UPLOAD =                        0x35
SVC_TRANSFER_DATA =                         0x36
SVC_REQUEST_TRANSFER_EXIT =                 0x37
SVC_WRITE_MEMORY_BY_ADDRESS =               0x3d
SVC_TESTER_PRESENT =                        0x3e
SVC_CONTROL_DTC_SETTING =                   0x85

UDS_SVCS = { v:k for k,v in globals().items() if k.startswith('SVC_') }

POS_RESP_CODES = { (k|0x40) : "OK_" + v.lower() for k,v in UDS_SVCS.items() }
POS_RESP_CODES[0] = 'Success'

NEG_RESP_REPR = {}
for k,v in NEG_RESP_CODES.items():
    NEG_RESP_REPR[k] = 'ERR_' + v

RESP_CODES = {}
RESP_CODES.update(NEG_RESP_REPR)
RESP_CODES.update(POS_RESP_CODES)


class NegativeResponseException(Exception):
    def __init__(self, neg_code, svc, msg):
        self.neg_code = neg_code
        self.msg = msg
        self.svc = svc

    def __repr__(self):
        negresprepr = NEG_RESP_CODES.get(self.neg_code)
        return "NEGATIVE RESPONSE to 0x%x (%s):   ERROR 0x%x: %s" % \
            (self.svc, UDS_SVCS.get(self.svc), self.neg_code, negresprepr, self.msg.encode('hex'))

    def __str__(self):
        negresprepr = NEG_RESP_CODES.get(self.neg_code)
        return "NEGATIVE RESPONSE to 0x%x (%s):   ERROR 0x%x: %s   \tmsg: %s" % \
            (self.svc, UDS_SVCS.get(self.svc), self.neg_code, negresprepr, self.msg.encode('hex'))


class UDS:
    def __init__(self, c, tx_arbid, rx_arbid=None, verbose=True, extflag=0):
        self.c = c
        self.verbose = verbose
        self.extflag = extflag

        if rx_arbid == None:
            rx_arbid = tx_arbid + 8 # by UDS spec

        self.tx_arbid = tx_arbid
        self.rx_arbid = rx_arbid

    def xmit_recv(self, data, extflag=0, timeout=3, count=1, service=None):
        msg = self.c.ISOTPxmit_recv(self.tx_arbid, self.rx_arbid, data, extflag, timeout, count, service)

        # check if the response is something we know about and can help out
        if msg != None and len(msg):
            svc = ord(data[0])
            svc_resp = ord(msg[0])
            errcode = 0
            if len(msg) >= 3:
                errcode = ord(msg[2])

            if svc_resp == svc + 0x40:
                if self.verbose: 
                    print "Positive Response!"

            negresprepr = NEG_RESP_CODES.get(errcode)
            if negresprepr != None and svc_resp != svc + 0x40:
{'fraction_non_alphanumeric': 0.09378315298111606, 'fraction_numerical': 0.03967748779970295, 'mean_word_length': 2.5031586770717205, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2915691', 'n_tokens_mistral': 3525, 'n_tokens_neox': 3357, 'n_words': 539}