<commit_before># Utility functions for UDS

import time
import string
import struct
from contextlib import contextmanager
from cancat import uds
from cancat.uds.types import ECUAddress
from cancat.utils import log, _range_func

def enter_session(u, session, prereq_sessions=None):
    # Enter any required preq sessions
    if prereq_sessions:
        for prereq in prereq_sessions:
            #u.DiagnosticSessionControl(prereq)
            enter_session(u, prereq)

    msg = None
    # Handle 0x7f:'ServiceNotSupportedInActiveSession' errors
    for i in range(30):
        try:
            msg = u.DiagnosticSessionControl(session)
            break
        except uds.NegativeResponseException as e:
            if e.neg_code != 0x7f:
                raise e

        # Wait until 
        time.sleep(0.1)

    return msg

@contextmanager
def new_session(u, session, prereq_sessions=None, tester_present=False):
    try:
        # Enter any required preq sessions
        if prereq_sessions:
            for prereq in prereq_sessions:
                #u.DiagnosticSessionControl(prereq)
                enter_session(u, prereq)

        msg = None
        # Handle 0x7f:'ServiceNotSupportedInActiveSession' errors
        for i in range(30):
            try:
                msg = u.DiagnosticSessionControl(session)
                break
            except uds.NegativeResponseException as e:
                if e.neg_code != 0x7f:
                    raise e

            # Wait until 
            time.sleep(0.1)

        # Start tester present again
        if tester_present:
            u.StartTesterPresent(request_response=False)

        yield msg
    finally:
        if tester_present:
            u.StopTesterPresent()


def find_possible_resp(u, start_index, tx_arbid, service, subfunction=None, timeout=3.0):
    # Starting at the supplied starting index, find the service request, and
    # then look for possible responses until the supplied timeout
    
    if subfunction:
        tx_match_bytes = struct.pack('>BH', service, subfunction)
        rx_match_bytes = struct.pack('>BH', service + 0x40, subfunction)
        match_len = 3
    else:
        tx_match_bytes = struct.pack('>B', service)
        rx_match_bytes = struct.pack('>B', service + 0x40)
        match_len = 1

    for idx, _, _, msg in u.c.genCanMsgs(start=start_index, arbids=[tx_arbid], maxsecs=timeout):
        ftype = ord(msg[0]) >> 4
        # Our Tx arbid is probably a 0, but check for frame type 1 as well.
        if (ftype == 0 and msg[1:1+match_len] == tx_match_bytes) or \
                (ftype == 1 and msg[2:2+match_len] == tx_match_bytes):
            tx_index = idx
            tx_msg = msg
            break
    else
        err = 'Unable to find tx arbid {} starting at index {}'.format(hex(tx_arbid), start_index)
        raise RangeError(err)

    # Determine the proper rx_arbid range to look for
    if tx_arbid > 0x18db0000:
        base_arbid = tx_arbid & 0xFFFF00FF
        rx_range = _range_func(base_arbid, base_arbid + 0x10000, 0x100)
    else:
        rx_range = _range_func(0x700, 0x800)

    err_match = b'\x7F' + struct.pack('>B', service)

    possible_matches = []
    for idx, _, arbid, msg in u.c.genCanMsgs(start=tx_index+1, arbids=rx_range, maxsecs=timeout):
        ftype = ord(msg[0]) >> 4
        # Check for frame types 0 (positive and negative responses) and 1
        if (ftype == 0 and msg[1:1+match_len] == rx_match_bytes) or \
                (ftype == 0 and msg[1:3] == err_match) or \
                (ftype == 1 and msg[2:2+match_len] == rx_match_bytes):
            possible_matches.append((arbid, msg))
    else
        return tx_msg, None 

    return  tx_msg, possible_matches


def ecu_did_scan(c, udsclass, arb_id_range, ext=0, did=0xf190, timeout=3.0, delay=None, verbose_flag=False):
    scan_type = ''
    if ext:
        scan_type = ' ext'

    log.debug('Starting{} DID read ECU scan for range: {}'.format(scan_type, arb_id_range))
    c.placeCanBookmark('ecu_did_scan({}, ext={}, did={}, timeout={}, delay={})'.format(
        arb_id_range, ext, did, timeout, delay))
    ecus = []
    for i in arb_id_range:  
        if ext and i == 0xF1:
            # Skip i == 0xF1 because in that case the sender and receiver IDs 
            # are the same
            log.detail('Skipping 0xF1 on ext ECU scan: invalid ECU address')
            continue
        elif ext == False and i >= 0xF8:
            # For non-extended scans the valid range goes from 0x00 to 0xFF, but 
            # stop the scan at 0xF7 because at that time the response is the 
            # largest possible valid value
            log.detail('Stopping std ECU scan at 0xF7: last valid ECU address')
            break
        elif ext:
            arb_id = 0x18db00f1 + (i << 8)
            resp_id = 0x18dbf100 + i
        else:
            arb_id = 0x700 + i
            resp_id = 0x700 + i + 8

        addr = ECUAddress(arb_id, resp_id, ext)

        u = udsclass(c, addr.tx_arbid, addr.rx_arbid, extflag=addr.extflag,
                verbose=verbose_flag, timeout=timeout)
        log.detail('Trying {}'.format(addr))
        try:
            start_index = u.c.getCanMsgCount()
            msg = u.ReadDID(did)
            if msg is not None:
                log.debug('{} DID {}: {}'.format(addr, hex(did), repr(msg)))
                log.msg('found {}'.format(addr))

                ecus.append(addr)
            else:
                tx_msg, responses = find_possible_resp(u, start_index, arb_id,
                        uds.SVC_READ_DATA_BY_IDENTIFIER, did, timeout)
                if responses:
                    log.warn('Possible non-standard responses for {} found:'.format(addr))
                    log.debug('tx msg:'.format(tx_msg.encode('hex')))
                    for rx_arbid, msg in responses:
                        log.warn('{}: {}'.format(hex(rx_arbid), msg.encode('hex')))
        except uds.NegativeResponseException as e:
            log.debug('{} DID {}: {}'.format(addr, hex(did), e))
            log.msg('found {}'.format(addr))

            # If a negative response happened, that means an ECU is present 
            # to respond at this address.
            ecus.append(addr)

        if delay:
            time.sleep(delay)

    return ecus


def ecu_session_scan(c, udsclass, arb_id_range, ext=0, session=1, verbose_flag=False, timeout=3.0, delay=None):
    scan_type = ''
    if ext:
        scan_type = ' ext'

    log.debug('Starting{} Session ECU scan for range: {}'.format(scan_type, arb_id_range))
    c.placeCanBookmark('ecu_session_scan({}, ext={}, session={}, timeout={}, delay={})'.format(
        arb_id_range, ext, session, timeout, delay))

    ecus = []
    for i in arb_id_range:  
        if ext and i == 0xF1:
            # Skip i == 0xF1 because in that case the sender and receiver IDs 
            # are the same
            continue
        elif ext == False and i >= 0xF8:
            # For non-extended scans the valid range goes from 0x00 to 0xFF, but 
            # stop the scan at 0xF7 because at that time the response is the 
            # largest possible valid value
            break
        elif ext:
            arb_id = 0x18db00f1 + (i << 8)
            resp_id = 0x18dbf100 + i
        else:
            arb_id = 0x700 + i
            resp_id = 0x700 + i + 8

        addr = ECUAddress(arb_id, resp_id, ext)
        u = udsclass(c, addr.tx_arbid, addr.rx_arbid, extflag=addr.extflag, verbose=verbose_flag, timeout=timeout)
        log.detail('Trying {}'.format(addr))
        try:
            with new_session(u, sess) as msg:
                if msg is not None:
                    log.debug('{} session {}: {}'.format(addr, session, repr(msg)))
                    log.msg('found {}'.format(addr))

                    ecus.append(addr)
                else:
                    tx_msg, responses = find_possible_resp(u, start_index, arb_id,
                            uds.SVC_DIAGNOSTICS_SESSION_CONTROL, session, timeout)
                    if responses:
                        log.warn('Possible non-standard responses for {} found:'.format(addr))
                        log.debug('tx msg:'.format(tx_msg.encode('hex')))
                        for rx_arbid, msg in responses:
                            log.warn('{}: {}'.format(hex(rx_arbid), msg.encode('hex')))
        except uds.NegativeResponseException as e:
            log.debug('{} session {}: {}'.format(addr, session, e))
            log.msg('found {}'.format(addr))

            # If a negative response happened, that means an ECU is present 
            # to respond at this address.
            ecus.append(addr)

        if delay:
            time.sleep(delay)

    return ecus


def try_read_did(u, did):
    data = None
    try:<commit_msg>added re-scanning of possible non-standard addresses<commit_after># Utility functions for UDS

import time
import string
import struct
from contextlib import contextmanager
from cancat import uds
from cancat.uds.types import ECUAddress
from cancat.utils import log, _range_func

def enter_session(u, session, prereq_sessions=None):
    # Enter any required preq sessions
    if prereq_sessions:
        for prereq in prereq_sessions:
            #u.DiagnosticSessionControl(prereq)
            enter_session(u, prereq)

    msg = None
    # Handle 0x7f:'ServiceNotSupportedInActiveSession' errors
    for i in range(30):
        try:
            msg = u.DiagnosticSessionControl(session)
            break
        except uds.NegativeResponseException as e:
            if e.neg_code != 0x7f:
                raise e

        # Wait until 
        time.sleep(0.1)

    return msg

@contextmanager
def new_session(u, session, prereq_sessions=None, tester_present=False):
    try:
        # Enter any required preq sessions
        if prereq_sessions:
            for prereq in prereq_sessions:
                #u.DiagnosticSessionControl(prereq)
                enter_session(u, prereq)

        msg = None
        # Handle 0x7f:'ServiceNotSupportedInActiveSession' errors
        for i in range(30):
            try:
                msg = u.DiagnosticSessionControl(session)
                break
            except uds.NegativeResponseException as e:
                if e.neg_code != 0x7f:
                    raise e

            # Wait until 
            time.sleep(0.1)

        # Start tester present again
        if tester_present:
            u.StartTesterPresent(request_response=False)

        yield msg
    finally:
        if tester_present:
            u.StopTesterPresent()


def find_possible_resp(u, start_index, tx_arbid, service, subfunction=None, timeout=3.0):
    # Starting at the supplied starting index, find the service request, and
    # then look for possible responses until the supplied timeout
    
    if subfunction:
        tx_match_bytes = struct.pack('>BH', service, subfunction)
        rx_match_bytes = struct.pack('>BH', service + 0x40, subfunction)
        match_len = 3
    else:
        tx_match_bytes = struct.pack('>B', service)
        rx_match_bytes = struct.pack('>B', service + 0x40)
        match_len = 1

    for idx, _, _, msg in u.c.genCanMsgs(start=start_index, arbids=[tx_arbid], maxsecs=timeout):
        ftype = ord(msg[0]) >> 4
        # Our Tx arbid is probably a 0, but check for frame type 1 as well.
        if (ftype == 0 and msg[1:1+match_len] == tx_match_bytes) or \
                (ftype == 1 and msg[2:2+match_len] == tx_match_bytes):
            tx_index = idx
            tx_msg = msg
            break
    else
        err = 'Unable to find tx arbid {} starting at index {}'.format(hex(tx_arbid), start_index)
        raise RangeError(err)

    # Determine the proper rx_arbid range to look for
    if tx_arbid > 0x18db0000:
        base_arbid = tx_arbid & 0xFFFF00FF
        rx_range = _range_func(base_arbid, base_arbid + 0x10000, 0x100)
    else:
        rx_range = _range_func(0x700, 0x800)

    err_match = b'\x7F' + struct.pack('>B', service)

    possible_matches = []
    for idx, _, arbid, msg in u.c.genCanMsgs(start=tx_index+1, arbids=rx_range, maxsecs=timeout):
        ftype = ord(msg[0]) >> 4
        # Check for frame types 0 (positive and negative responses) and 1
        if (ftype == 0 and msg[1:1+match_len] == rx_match_bytes) or \
                (ftype == 0 and msg[1:3] == err_match) or \
                (ftype == 1 and msg[2:2+match_len] == rx_match_bytes):
            possible_matches.append((arbid, msg))
    else
        return tx_msg, None 

    return  tx_msg, possible_matches


def ecu_did_scan(c, udsclass, arb_id_range, ext=0, did=0xf190, timeout=3.0, delay=None, verbose_flag=False):
    scan_type = ''
    if ext:
        scan_type = ' ext'

    log.debug('Starting{} DID read ECU scan for range: {}'.format(scan_type, arb_id_range))
    c.placeCanBookmark('ecu_did_scan({}, ext={}, did={}, timeout={}, delay={})'.format(
        arb_id_range, ext, did, timeout, delay))
    ecus = []
    possible_ecus = []
    for i in arb_id_range:  
        if ext and i == 0xF1:
            # Skip i == 0xF1 because in that case the sender and receiver IDs 
            # are the same
            log.detail('Skipping 0xF1 on ext ECU scan: invalid ECU address')
            continue
        elif ext == False and i >= 0xF8:
            # For non-extended scans the valid range goes from 0x00 to 0xFF, but 
            # stop the scan at 0xF7 because at that time the response is the 
            # largest possible valid value
            log.detail('Stopping std ECU scan at 0xF7: last valid ECU address')
            break
        elif ext:
            arb_id = 0x18db00f1 + (i << 8)
            resp_id = 0x18dbf100 + i
        else:
            arb_id = 0x700 + i
            resp_id = 0x700 + i + 8

        addr = ECUAddress(arb_id, resp_id, ext)

        u = udsclass(c, addr.tx_arbid, addr.rx_arbid, extflag=addr.extflag,
                verbose=verbose_flag, timeout=timeout)
        log.detail('Trying {}'.format(addr))
        try:
            start_index = u.c.getCanMsgCount()
            msg = u.ReadDID(did)
            if msg is not None:
                log.debug('{} DID {}: {}'.format(addr, hex(did), repr(msg)))
                log.msg('found {}'.format(addr))

                ecus.append(addr)
            else:
                tx_msg, responses = find_possible_resp(u, start_index, arb_id,
                        uds.SVC_READ_DATA_BY_IDENTIFIER, did, timeout)
                if responses:
                    log.warn('Possible non-standard responses for {} found:'.format(addr))
                    log.debug('tx msg:'.format(tx_msg.encode('hex')))
                    for rx_arbid, msg in responses:
                        log.warn('{}: {}'.format(hex(rx_arbid), msg.encode('hex')))
                        possible_ecus.append(ECUAddress(arb_id, rx_arbid, ext))
        except uds.NegativeResponseException as e:
            log.debug('{} DID {}: {}'.format(addr, hex(did), e))
            log.msg('found {}'.format(addr))

            # If a negative response happened, that means an ECU is present 
            # to respond at this address.
            ecus.append(addr)

        if delay:
            time.sleep(delay)

    # Double check any non-standard responses that were found
    for addr in possible_ecus:  
        u = udsclass(c, addr.tx_arbid, addr.rx_arbid, extflag=addr.extflag,
                verbose=verbose_flag, timeout=timeout)
        log.detail('Trying {}'.format(addr))
        try:
            msg = u.ReadDID(did)
            if msg is not None:
                log.debug('{} DID {}: {}'.format(addr, hex(did), repr(msg)))
                log.msg('found {}'.format(addr))
                ecus.append(addr)
        except uds.NegativeResponseException as e:
            log.debug('{} DID {}: {}'.format(addr, hex(did), e))
            log.msg('found {}'.format(addr))

            # If a negative response happened, that means an ECU is present 
            # to respond at this address.
            ecus.append(addr)

        if delay:
            time.sleep(delay)

    return ecus


def ecu_session_scan(c, udsclass, arb_id_range, ext=0, session=1, verbose_flag=False, timeout=3.0, delay=None):
    scan_type = ''
    if ext:
        scan_type = ' ext'

    log.debug('Starting{} Session ECU scan for range: {}'.format(scan_type, arb_id_range))
    c.placeCanBookmark('ecu_session_scan({}, ext={}, session={}, timeout={}, delay={})'.format(
        arb_id_range, ext, session, timeout, delay))

    ecus = []
    possible_ecus = []
    for i in arb_id_range:  
        if ext and i == 0xF1:
            # Skip i == 0xF1 because in that case the sender and receiver IDs 
            # are the same
            continue
        elif ext == False and i >= 0xF8:
            # For non-extended scans the valid range goes from 0x00 to 0xFF, but 
            # stop the scan at 0xF7 because at that time the response is the 
            # largest possible valid value
            break
        elif ext:
            arb_id = 0x18db00f1 + (i << 8)
            resp_id = 0x18dbf100 + i
        else:
            arb_id = 0x700 + i
            resp_id = 0x700 + i + 8

        addr = ECUAddress(arb_id, resp_id, ext)
        u = udsclass(c, addr.tx_arbid, addr.rx_arbid, extflag=addr.extflag, verbose=verbose_flag, timeout=timeout)
        log.detail('Trying {}'.format(addr))
        try:
            with new_session(u, sess) as msg:
                if msg is not None:
                    log.debug('{} session {}: {}'.format(addr, session, repr(msg)))
                    log.msg('found {}'.format(addr))

                    ecus.append(addr)
                else:
                    tx_msg, responses = find_possible_resp(u, start_index, arb_id,
                            uds.SVC_DIAGNOSTICS_SESSION_CONTROL, session, timeout)
                    if responses:
                        log.warn('Possible non-standard responses for {} found:'.format(addr))
                        log.debug('tx msg:'.format(tx_msg.encode('hex')))
                        for rx_arbid, msg in responses:
                            log.warn('{}: {}'.format(hex(rx_arbid), msg.encode('hex')))
                            possible_ecus.append(ECUAddress(arb_id, rx_arbid, ext))
        except uds.NegativeResponseException as e:
            log.debug('{} session {}: {}'.format(addr, session, e))
            log.msg('found {}'.format(addr))

            # If a negative response happened, that means an ECU is present 
            # to respond at this address.
            ecus.append(addr)

        if delay:
            time.sleep(delay)

    # Double check any non-standard responses that were found
    for addr in possible_ecus:  
        u = udsclass(c, addr.tx_arbid, addr.rx_arbid, extflag=addr.extflag,
                verbose=verbose_flag, timeout=timeout)
        log.detail('Trying {}'.format(addr))
        try:
            msg = u.ReadDID(did)
            if msg is not None:
                log.debug('{} DID {}: {}'.format(addr, hex(did), repr(msg)))
                log.msg('found {}'.format(addr))
                ecus.append(addr)
        except uds.NegativeResponseException as e:
            log.debug('{} DID {}: {}'.format(addr, hex(did), e))
            log.msg('found {}'.format(addr))

            # If a negative response happened, that means an ECU is present 
            # to respond at this address.
            ecus.append(addr)

        if delay:
            time.sleep(delay)

    return ecus


def try_read_did(u, did):
    data = None
    try:
{'fraction_non_alphanumeric': 0.10290896123083193, 'fraction_numerical': 0.016506189821182942, 'mean_word_length': 1.9751439830251591, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5480830', 'n_tokens_mistral': 6303, 'n_tokens_neox': 5825, 'n_words': 1636}