diff --git a/software/script/chameleon_cli_main.py b/software/script/chameleon_cli_main.py index 107b514..d069bfd 100755 --- a/software/script/chameleon_cli_main.py +++ b/software/script/chameleon_cli_main.py @@ -10,7 +10,7 @@ import pathlib import prompt_toolkit from prompt_toolkit.formatted_text import ANSI from prompt_toolkit.history import FileHistory -from chameleon_utils import CR, CG, CB, CC, CY, CM, C0 +from chameleon_utils import CR, CG, CY, C0 ULTRA = r""" ╦ ╦╦ ╔╦╗╦═╗╔═╗ @@ -43,7 +43,8 @@ class ChameleonCLI: def __init__(self): self.completer = chameleon_utils.CustomNestedCompleter.from_clitree(chameleon_cli_unit.root) self.session = prompt_toolkit.PromptSession(completer=self.completer, - history=FileHistory(pathlib.Path.home() / ".chameleon_history")) + history=FileHistory(str(pathlib.Path.home() / + ".chameleon_history"))) # new a device communication instance(only communication) self.device_com = chameleon_com.ChameleonCom() diff --git a/software/script/chameleon_cli_unit.py b/software/script/chameleon_cli_unit.py index c6d07de..d172114 100644 --- a/software/script/chameleon_cli_unit.py +++ b/software/script/chameleon_cli_unit.py @@ -10,6 +10,7 @@ from datetime import datetime import serial.tools.list_ports import threading import struct +from typing import Union from pathlib import Path from platform import uname @@ -17,7 +18,7 @@ import chameleon_com import chameleon_cmd from chameleon_utils import ArgumentParserNoExit, ArgsParserError, UnexpectedResponseError from chameleon_utils import CLITree -from chameleon_utils import CR, CG, CB, CC, CY, CM, C0 +from chameleon_utils import CR, CG, CB, CC, CY, C0 from chameleon_enum import Command, Status, SlotNumber, TagSenseType, TagSpecificType from chameleon_enum import MifareClassicWriteMode, MifareClassicPrngType, MifareClassicDarksideStatus, MfcKeyType from chameleon_enum import AnimationMode, ButtonType, ButtonPressFunction @@ -52,11 +53,12 @@ def check_tools(): class BaseCLIUnit: def __init__(self): # new a device command transfer and receiver instance(Send cmd and receive response) - self._device_com: chameleon_com.ChameleonCom | None = None - self._device_cmd: chameleon_cmd.ChameleonCMD = chameleon_cmd.ChameleonCMD(self._device_com) + self._device_com: Union[chameleon_com.ChameleonCom, None] = None + self._device_cmd: Union[chameleon_cmd.ChameleonCMD, None] = None @property def device_com(self) -> chameleon_com.ChameleonCom: + assert self._device_com is not None return self._device_com @device_com.setter @@ -66,6 +68,7 @@ class BaseCLIUnit: @property def cmd(self) -> chameleon_cmd.ChameleonCMD: + assert self._device_cmd is not None return self._device_cmd def args_parser(self) -> ArgumentParserNoExit: @@ -112,6 +115,7 @@ class BaseCLIUnit: def thread_read_output(self): while self._process.poll() is None: + assert self._process.stdout is not None data = self._process.stdout.read(1024) if len(data) > 0: self.output += data.decode(encoding="utf-8") @@ -635,7 +639,7 @@ class HF14AInfo(ReaderRequiredUnit): def on_exec(self, args: argparse.Namespace): scan = HF14AScan() scan.device_com = self.device_com - scan.scan(deep=1) + scan.scan(deep=True) @hf_mf.command('nested') @@ -665,7 +669,7 @@ class HFMFNested(ReaderRequiredUnit): if nt_level == 2: return 'HardNested' - def recover_a_key(self, block_known, type_known, key_known, block_target, type_target) -> str or None: + def recover_a_key(self, block_known, type_known, key_known, block_target, type_target) -> Union[str, None]: """ recover a key from key known. @@ -740,12 +744,11 @@ class HFMFNested(ReaderRequiredUnit): block_known = args.blk # default to A type_known = MfcKeyType.B if args.b else MfcKeyType.A - key_known: str = args.key if not re.match(r"^[a-fA-F0-9]{12}$", key_known): print("key must include 12 HEX symbols") return - key_known: bytearray = bytearray.fromhex(key_known) + key_known_bytes = bytes.fromhex(key_known) block_target = args.tblk # default to A type_target = MfcKeyType.B if args.b else MfcKeyType.A @@ -753,7 +756,7 @@ class HFMFNested(ReaderRequiredUnit): print(f"{CR}Target key already known{C0}") return print(f" - {C0}Nested recover one key running...{C0}") - key = self.recover_a_key(block_known, type_known, key_known, block_target, type_target) + key = self.recover_a_key(block_known, type_known, key_known_bytes, block_target, type_target) if key is None: print(f"{CY}No key found, you can retry.{C0}") else: @@ -862,8 +865,8 @@ class HFMFWRBL(MF1AuthArgsUnit): param = self.get_param(args) if not re.match(r"^[a-fA-F0-9]{32}$", args.data): raise ArgsParserError("Data must include 32 HEX symbols") - param.data = bytearray.fromhex(args.data) - resp = self.cmd.mf1_write_one_block(param.block, param.type, param.key, param.data) + data = bytearray.fromhex(args.data) + resp = self.cmd.mf1_write_one_block(param.block, param.type, param.key, data) if resp: print(f" - {CG}Write done.{C0}") else: @@ -1411,7 +1414,7 @@ class HWSlotList(DeviceRequiredUnit): def get_slot_name(self, slot, sense): try: - name = self.cmd.get_slot_tag_nick(slot, sense).decode(encoding="utf8") + name = self.cmd.get_slot_tag_nick(slot, sense) return {'baselen': len(name), 'metalen': len(CC+C0), 'name': f'{CC}{name}{C0}'} except UnexpectedResponseError: return {'baselen': 0, 'metalen': 0, 'name': ''} @@ -1666,10 +1669,7 @@ class HWSlotNick(SlotIndexArgsUnit, SenseTypeArgsUnit): sense_type = TagSenseType.HF if args.name is not None: name: str = args.name - encoded_name = name.encode(encoding="utf8") - if len(encoded_name) > 32: - raise ValueError("Your tag nick name too long.") - self.cmd.set_slot_tag_nick(slot_num, sense_type, encoded_name) + self.cmd.set_slot_tag_nick(slot_num, sense_type, name) print(f' - Set tag nick name for slot {slot_num} {sense_type.name}: {name}') elif args.delete: self.cmd.delete_slot_tag_nick(slot_num, sense_type) @@ -1677,7 +1677,7 @@ class HWSlotNick(SlotIndexArgsUnit, SenseTypeArgsUnit): else: res = self.cmd.get_slot_tag_nick(slot_num, sense_type) print(f' - Get tag nick name for slot {slot_num} {sense_type.name}' - f': {res.decode(encoding="utf8")}') + f': {res}') @hw_slot.command('store') @@ -1919,9 +1919,9 @@ class HWSettingsBLEKey(DeviceRequiredUnit): return parser def on_exec(self, args: argparse.Namespace): - resp = self.cmd.get_ble_pairing_key() + key = self.cmd.get_ble_pairing_key() print(" - The current key of the device(ascii): " - f"{CG}{resp.decode(encoding='ascii')}{C0}") + f"{CG}{key}{C0}") if args.key is not None: if len(args.key) != 6: diff --git a/software/script/chameleon_cmd.py b/software/script/chameleon_cmd.py index 130d657..15356fe 100644 --- a/software/script/chameleon_cmd.py +++ b/software/script/chameleon_cmd.py @@ -1,5 +1,6 @@ import struct import ctypes +from typing import Union import chameleon_com from chameleon_utils import expect_response @@ -28,7 +29,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_APP_VERSION) if resp.status == Status.SUCCESS: - resp.data = struct.unpack('!BB', resp.data) + resp.parsed = struct.unpack('!BB', resp.data) # older protocol, must upgrade! if resp.status == 0 and resp.data == b'\x00\x01': print("Chameleon does not understand new protocol. Please update firmware") @@ -43,7 +44,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_DEVICE_CHIP_ID) if resp.status == Status.SUCCESS: - resp.data = resp.data.hex() + resp.parsed = resp.data.hex() return resp @expect_response(Status.SUCCESS) @@ -53,21 +54,21 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_DEVICE_ADDRESS) if resp.status == Status.SUCCESS: - resp.data = resp.data.hex() + resp.parsed = resp.data.hex() return resp @expect_response(Status.SUCCESS) - def get_git_version(self) -> str: + def get_git_version(self): resp = self.device.send_cmd_sync(Command.GET_GIT_VERSION) if resp.status == Status.SUCCESS: - resp.data = resp.data.decode('utf-8') + resp.parsed = resp.data.decode('utf-8') return resp @expect_response(Status.SUCCESS) def get_device_mode(self): resp = self.device.send_cmd_sync(Command.GET_DEVICE_MODE) if resp.status == Status.SUCCESS: - resp.data, = struct.unpack('!?', resp.data) + resp.parsed, = struct.unpack('!?', resp.data) return resp def is_device_reader_mode(self) -> bool: @@ -113,7 +114,7 @@ class ChameleonCMD: ats, = struct.unpack_from(f'!{atslen}s', resp.data, offset) offset += struct.calcsize(f'!{atslen}s') data.append({'uid': uid, 'atqa': atqa, 'sak': sak, 'ats': ats}) - resp.data = data + resp.parsed = data return resp def mf1_detect_support(self): @@ -134,7 +135,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.MF1_DETECT_PRNG) if resp.status == Status.HF_TAG_OK: - resp.data = resp.data[0] + resp.parsed = resp.data[0] return resp @expect_response(Status.HF_TAG_OK) @@ -148,7 +149,7 @@ class ChameleonCMD: resp = self.device.send_cmd_sync(Command.MF1_DETECT_NT_DIST, data) if resp.status == Status.HF_TAG_OK: uid, dist = struct.unpack('!II', resp.data) - resp.data = {'uid': uid, 'dist': dist} + resp.parsed = {'uid': uid, 'dist': dist} return resp @expect_response(Status.HF_TAG_OK) @@ -160,12 +161,12 @@ class ChameleonCMD: data = struct.pack('!BB6sBB', type_known, block_known, key_known, type_target, block_target) resp = self.device.send_cmd_sync(Command.MF1_NESTED_ACQUIRE, data) if resp.status == Status.HF_TAG_OK: - resp.data = [{'nt': nt, 'nt_enc': nt_enc, 'par': par} - for nt, nt_enc, par in struct.iter_unpack('!IIB', resp.data)] + resp.parsed = [{'nt': nt, 'nt_enc': nt_enc, 'par': par} + for nt, nt_enc, par in struct.iter_unpack('!IIB', resp.data)] return resp @expect_response(Status.HF_TAG_OK) - def mf1_darkside_acquire(self, block_target, type_target, first_recover: int or bool, sync_max): + def mf1_darkside_acquire(self, block_target, type_target, first_recover: Union[int, bool], sync_max): """ Collect the key parameters needed for Darkside decryption. @@ -180,9 +181,9 @@ class ChameleonCMD: if resp.status == Status.HF_TAG_OK: if resp.data[0] == MifareClassicDarksideStatus.OK: darkside_status, uid, nt1, par, ks1, nr, ar = struct.unpack('!BIIQQII', resp.data) - resp.data = (darkside_status, {'uid': uid, 'nt1': nt1, 'par': par, 'ks1': ks1, 'nr': nr, 'ar': ar}) + resp.parsed = (darkside_status, {'uid': uid, 'nt1': nt1, 'par': par, 'ks1': ks1, 'nr': nr, 'ar': ar}) else: - resp.data = (resp.data[0],) + resp.parsed = (resp.data[0],) return resp @expect_response([Status.HF_TAG_OK, Status.MF_ERR_AUTH]) @@ -197,7 +198,7 @@ class ChameleonCMD: """ data = struct.pack('!BB6s', type_value, block, key) resp = self.device.send_cmd_sync(Command.MF1_AUTH_ONE_KEY_BLOCK, data) - resp.data = resp.status == Status.HF_TAG_OK + resp.parsed = resp.status == Status.HF_TAG_OK return resp @expect_response(Status.HF_TAG_OK) @@ -211,7 +212,9 @@ class ChameleonCMD: :return: """ data = struct.pack('!BB6s', type_value, block, key) - return self.device.send_cmd_sync(Command.MF1_READ_ONE_BLOCK, data) + resp = self.device.send_cmd_sync(Command.MF1_READ_ONE_BLOCK, data) + resp.parsed = resp.data + return resp @expect_response(Status.HF_TAG_OK) def mf1_write_one_block(self, block, type_value, key, block_data): @@ -226,7 +229,7 @@ class ChameleonCMD: """ data = struct.pack('!BB6s16s', type_value, block, key, block_data) resp = self.device.send_cmd_sync(Command.MF1_WRITE_ONE_BLOCK, data) - resp.data = resp.status == Status.HF_TAG_OK + resp.parsed = resp.status == Status.HF_TAG_OK return resp @expect_response(Status.HF_TAG_OK) @@ -270,7 +273,9 @@ class ChameleonCMD: f'must be between {((len(data) - 1) * 8 )+1} and {len(data) * 8} included') data = bytes(cs)+struct.pack(f'!HH{len(data)}s', resp_timeout_ms, bitlen, bytearray(data)) - return self.device.send_cmd_sync(Command.HF14A_RAW, data, timeout=(resp_timeout_ms / 1000) + 1) + resp = self.device.send_cmd_sync(Command.HF14A_RAW, data, timeout=(resp_timeout_ms // 1000) + 1) + resp.parsed = resp.data + return resp @expect_response(Status.HF_TAG_OK) def mf1_static_nested_acquire(self, block_known, type_known, key_known, block_target, type_target): @@ -281,7 +286,7 @@ class ChameleonCMD: data = struct.pack('!BB6sBB', type_known, block_known, key_known, type_target, block_target) resp = self.device.send_cmd_sync(Command.MF1_STATIC_NESTED_ACQUIRE, data) if resp.status == Status.HF_TAG_OK: - resp.data = { + resp.parsed = { 'uid': struct.unpack('!I', resp.data[0:4])[0], 'nts': [ { @@ -299,7 +304,9 @@ class ChameleonCMD: :return: """ - return self.device.send_cmd_sync(Command.EM410X_SCAN) + resp = self.device.send_cmd_sync(Command.EM410X_SCAN) + resp.parsed = resp.data + return resp @expect_response(Status.LF_TAG_OK) def em410x_write_to_t55xx(self, id_bytes: bytes): @@ -325,8 +332,8 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_SLOT_INFO) if resp.status == Status.SUCCESS: - resp.data = [{'hf': hf, 'lf': lf} - for hf, lf in struct.iter_unpack('!HH', resp.data)] + resp.parsed = [{'hf': hf, 'lf': lf} + for hf, lf in struct.iter_unpack('!HH', resp.data)] return resp @expect_response(Status.SUCCESS) @@ -338,7 +345,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_ACTIVE_SLOT) if resp.status == Status.SUCCESS: - resp.data = resp.data[0] + resp.parsed = resp.data[0] return resp @expect_response(Status.SUCCESS) @@ -425,7 +432,9 @@ class ChameleonCMD: """ Get the simulated EM410x card id """ - return self.device.send_cmd_sync(Command.EM410X_GET_EMU_ID) + resp = self.device.send_cmd_sync(Command.EM410X_GET_EMU_ID) + resp.parsed = resp.data + return resp @expect_response(Status.SUCCESS) def mf1_set_detection_enable(self, enabled: bool): @@ -447,7 +456,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.MF1_GET_DETECTION_COUNT) if resp.status == Status.SUCCESS: - resp.data, = struct.unpack('!I', resp.data) + resp.parsed, = struct.unpack('!I', resp.data) return resp @expect_response(Status.SUCCESS) @@ -476,7 +485,7 @@ class ChameleonCMD: 'ar': ar.hex() }) pos += struct.calcsize('!BB4s4s4s4s') - resp.data = result_list + resp.parsed = result_list return resp @expect_response(Status.SUCCESS) @@ -498,7 +507,9 @@ class ChameleonCMD: Gets data for selected block range """ data = struct.pack('!BB', block_start, block_count) - return self.device.send_cmd_sync(Command.MF1_READ_EMU_BLOCK_DATA, data) + resp = self.device.send_cmd_sync(Command.MF1_READ_EMU_BLOCK_DATA, data) + resp.parsed = resp.data + return resp @expect_response(Status.SUCCESS) def hf14a_set_anti_coll_data(self, uid: bytes, atqa: bytes, sak: bytes, ats: bytes = b''): @@ -515,7 +526,7 @@ class ChameleonCMD: return self.device.send_cmd_sync(Command.HF14A_SET_ANTI_COLL_DATA, data) @expect_response(Status.SUCCESS) - def set_slot_tag_nick(self, slot: SlotNumber, sense_type: TagSenseType, name: bytes): + def set_slot_tag_nick(self, slot: SlotNumber, sense_type: TagSenseType, name: str): """ Set the nick name of the slot. @@ -524,8 +535,11 @@ class ChameleonCMD: :param name: Card slot nickname :return: """ + encoded_name = name.encode(encoding="utf8") + if len(encoded_name) > 32: + raise ValueError("Your tag nick name too long.") # SlotNumber() will raise error for us if slot not in slot range - data = struct.pack(f'!BB{len(name)}s', SlotNumber.to_fw(slot), sense_type, name) + data = struct.pack(f'!BB{len(encoded_name)}s', SlotNumber.to_fw(slot), sense_type, encoded_name) return self.device.send_cmd_sync(Command.SET_SLOT_TAG_NICK, data) @expect_response(Status.SUCCESS) @@ -539,7 +553,9 @@ class ChameleonCMD: """ # SlotNumber() will raise error for us if slot not in slot range data = struct.pack('!BB', SlotNumber.to_fw(slot), sense_type) - return self.device.send_cmd_sync(Command.GET_SLOT_TAG_NICK, data) + resp = self.device.send_cmd_sync(Command.GET_SLOT_TAG_NICK, data) + resp.parsed = resp.data.decode(encoding="utf8") + return resp @expect_response(Status.SUCCESS) def delete_slot_tag_nick(self, slot: SlotNumber, sense_type: TagSenseType): @@ -569,11 +585,11 @@ class ChameleonCMD: resp = self.device.send_cmd_sync(Command.MF1_GET_EMULATOR_CONFIG) if resp.status == Status.SUCCESS: b1, b2, b3, b4, b5 = struct.unpack('!????B', resp.data) - resp.data = {'detection': b1, - 'gen1a_mode': b2, - 'gen2_mode': b3, - 'block_anti_coll_mode': b4, - 'write_mode': b5} + resp.parsed = {'detection': b1, + 'gen1a_mode': b2, + 'gen2_mode': b3, + 'block_anti_coll_mode': b4, + 'write_mode': b5} return resp @expect_response(Status.SUCCESS) @@ -630,7 +646,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_ANIMATION_MODE) if resp.status == Status.SUCCESS: - resp.data = resp.data[0] + resp.parsed = resp.data[0] return resp @expect_response(Status.SUCCESS) @@ -640,7 +656,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_ENABLED_SLOTS) if resp.status == Status.SUCCESS: - resp.data = [{'hf': hf, 'lf': lf} for hf, lf in struct.iter_unpack('!BB', resp.data)] + resp.parsed = [{'hf': hf, 'lf': lf} for hf, lf in struct.iter_unpack('!BB', resp.data)] return resp @expect_response(Status.SUCCESS) @@ -657,7 +673,7 @@ class ChameleonCMD: Reset settings stored in flash memory """ resp = self.device.send_cmd_sync(Command.RESET_SETTINGS) - resp.data = resp.status == Status.SUCCESS + resp.parsed = resp.status == Status.SUCCESS return resp @expect_response(Status.SUCCESS) @@ -666,7 +682,7 @@ class ChameleonCMD: Store settings to flash memory """ resp = self.device.send_cmd_sync(Command.SAVE_SETTINGS) - resp.data = resp.status == Status.SUCCESS + resp.parsed = resp.status == Status.SUCCESS return resp @expect_response(Status.SUCCESS) @@ -675,7 +691,7 @@ class ChameleonCMD: Reset to factory settings """ resp = self.device.send_cmd_sync(Command.WIPE_FDS) - resp.data = resp.status == Status.SUCCESS + resp.parsed = resp.status == Status.SUCCESS self.device.close() return resp @@ -686,7 +702,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_BATTERY_INFO) if resp.status == Status.SUCCESS: - resp.data = struct.unpack('!HB', resp.data) + resp.parsed = struct.unpack('!HB', resp.data) return resp @expect_response(Status.SUCCESS) @@ -697,7 +713,7 @@ class ChameleonCMD: data = struct.pack('!B', button) resp = self.device.send_cmd_sync(Command.GET_BUTTON_PRESS_CONFIG, data) if resp.status == Status.SUCCESS: - resp.data = resp.data[0] + resp.parsed = resp.data[0] return resp @expect_response(Status.SUCCESS) @@ -716,7 +732,7 @@ class ChameleonCMD: data = struct.pack('!B', button) resp = self.device.send_cmd_sync(Command.GET_LONG_BUTTON_PRESS_CONFIG, data) if resp.status == Status.SUCCESS: - resp.data = resp.data[0] + resp.parsed = resp.data[0] return resp @expect_response(Status.SUCCESS) @@ -746,7 +762,9 @@ class ChameleonCMD: """ Get config of ble connect key """ - return self.device.send_cmd_sync(Command.GET_BLE_PAIRING_KEY) + resp = self.device.send_cmd_sync(Command.GET_BLE_PAIRING_KEY) + resp.parsed = resp.data.decode(encoding='ascii') + return resp @expect_response(Status.SUCCESS) def delete_all_ble_bonds(self): @@ -768,7 +786,7 @@ class ChameleonCMD: status=Status.NOT_IMPLEMENTED) else: if resp.status == Status.SUCCESS: - resp.data = [x[0] for x in struct.iter_unpack('!H', resp.data)] + resp.parsed = [x[0] for x in struct.iter_unpack('!H', resp.data)] return resp @expect_response(Status.SUCCESS) @@ -781,7 +799,7 @@ class ChameleonCMD: resp = self.device.send_cmd_sync(Command.GET_DEVICE_MODEL) if resp.status == Status.SUCCESS: - resp.data = resp.data[0] + resp.parsed = resp.data[0] return resp @expect_response(Status.SUCCESS) @@ -809,14 +827,14 @@ class ChameleonCMD: settings_version, animation_mode, btn_press_A, btn_press_B, btn_long_press_A, \ btn_long_press_B, ble_pairing_enable, ble_pairing_key = \ struct.unpack('!BBBBBBB6s', resp.data) - resp.data = {'settings_version': settings_version, - 'animation_mode': animation_mode, - 'btn_press_A': btn_press_A, - 'btn_press_B': btn_press_B, - 'btn_long_press_A': btn_long_press_A, - 'btn_long_press_B': btn_long_press_B, - 'ble_pairing_enable': ble_pairing_enable, - 'ble_pairing_key': ble_pairing_key} + resp.parsed = {'settings_version': settings_version, + 'animation_mode': animation_mode, + 'btn_press_A': btn_press_A, + 'btn_press_B': btn_press_B, + 'btn_long_press_A': btn_long_press_A, + 'btn_long_press_B': btn_long_press_B, + 'ble_pairing_enable': ble_pairing_enable, + 'ble_pairing_key': ble_pairing_key} return resp @expect_response(Status.SUCCESS) @@ -836,7 +854,7 @@ class ChameleonCMD: offset += struct.calcsize(f'!{uidlen}s2s1sB') ats, = struct.unpack_from(f'!{atslen}s', resp.data, offset) offset += struct.calcsize(f'!{atslen}s') - resp.data = {'uid': uid, 'atqa': atqa, 'sak': sak, 'ats': ats} + resp.parsed = {'uid': uid, 'atqa': atqa, 'sak': sak, 'ats': ats} return resp @expect_response(Status.SUCCESS) @@ -848,7 +866,7 @@ class ChameleonCMD: """ resp = self.device.send_cmd_sync(Command.GET_BLE_PAIRING_ENABLE) if resp.status == Status.SUCCESS: - resp.data, = struct.unpack('!?', resp.data) + resp.parsed, = struct.unpack('!?', resp.data) return resp @expect_response(Status.SUCCESS) diff --git a/software/script/chameleon_com.py b/software/script/chameleon_com.py index 34b6327..cf11728 100644 --- a/software/script/chameleon_com.py +++ b/software/script/chameleon_com.py @@ -3,7 +3,8 @@ import struct import threading import time import serial -from chameleon_utils import CR, CG, CB, CC, CY, CM, C0 +from typing import Union +from chameleon_utils import CR, CG, CC, CY, C0 from chameleon_enum import Command, Status # each thread is waiting for its data for 100 ms before looping again @@ -36,10 +37,11 @@ class Response: Chameleon Response Data """ - def __init__(self, cmd, status, data=b''): + def __init__(self, cmd, status, data=b'', parsed=None): self.cmd = cmd self.status = status - self.data: bytearray = data + self.data: bytes = data + self.parsed = parsed class ChameleonCom: @@ -55,20 +57,20 @@ class ChameleonCom: """ Create a chameleon device instance """ - self.serial_instance: serial.Serial | None = None + self.serial_instance: Union[serial.Serial, None] = None self.send_data_queue = queue.Queue() self.wait_response_map = {} self.event_closing = threading.Event() - def isOpen(self): + def isOpen(self) -> bool: """ Chameleon is connected and init. :return: """ - return self.serial_instance is not None and self.serial_instance.isOpen() + return self.serial_instance is not None and self.serial_instance.is_open - def open(self, port): + def open(self, port) -> "ChameleonCom": """ Open chameleon port to communication And init some variables @@ -86,8 +88,9 @@ class ChameleonCom: finally: if error is not None: raise OpenFailException(error) + assert self.serial_instance is not None try: - self.serial_instance.dtr = 1 # must make dtr enable + self.serial_instance.dtr = True # must make dtr enable except Exception: # not all serial support dtr, e.g. virtual serial over BLE pass @@ -102,7 +105,7 @@ class ChameleonCom: threading.Thread(target=self.thread_check_timeout).start() return self - def check_open(self): + def check_open(self) -> None: """ :return: @@ -111,7 +114,7 @@ class ChameleonCom: raise NotOpenException("Please call open() function to start device.") @staticmethod - def lrc_calc(array): + def lrc_calc(array: Union[bytearray, bytes]) -> int: """ Calc lrc and auto cut byte. @@ -133,6 +136,7 @@ class ChameleonCom: """ self.event_closing.set() try: + assert self.serial_instance is not None self.serial_instance.close() except Exception: pass @@ -156,6 +160,7 @@ class ChameleonCom: while self.isOpen(): # receive try: + assert self.serial_instance is not None data_bytes = self.serial_instance.read() except Exception as e: if not self.event_closing.is_set(): @@ -197,8 +202,8 @@ class ChameleonCom: # ok, lrc for data is correct. # and we are receive completed # print(f"Buffer data = {data_buffer.hex()}") - data_response = data_buffer[struct.calcsize('!BBHHHB'): - struct.calcsize(f'!BBHHHB{data_length}s')] + data_response = bytes(data_buffer[struct.calcsize('!BBHHHB'): + struct.calcsize(f'!BBHHHB{data_length}s')]) if DEBUG: try: command = Command(data_cmd) @@ -263,6 +268,7 @@ class ChameleonCom: self.wait_response_map[task_cmd]['end_time'] = start_time + task_timeout self.wait_response_map[task_cmd]['is_timeout'] = False try: + assert self.serial_instance is not None # send to device self.serial_instance.write(task['frame']) except Exception as e: @@ -292,7 +298,7 @@ class ChameleonCom: self.wait_response_map[task_cmd]['is_timeout'] = True time.sleep(THREAD_BLOCKING_TIMEOUT) - def make_data_frame_bytes(self, cmd: int, data: bytearray = None, status: int = 0) -> bytearray: + def make_data_frame_bytes(self, cmd: int, data: Union[bytes, None] = None, status: int = 0) -> bytes: """ Make data frame @@ -308,9 +314,9 @@ class ChameleonCom: frame[struct.calcsize('!BBHHH')] = self.lrc_calc(frame[:struct.calcsize('!BBHHH')]) # lrc3 frame[struct.calcsize(f'!BBHHHB{len(data)}s')] = self.lrc_calc(frame[:struct.calcsize(f'!BBHHHB{len(data)}s')]) - return frame + return bytes(frame) - def send_cmd_auto(self, cmd: int, data: bytearray = None, status: int = 0, callback=None, timeout: int = 3, + def send_cmd_auto(self, cmd: int, data: Union[bytes, None] = None, status: int = 0, callback=None, timeout: int = 3, close: bool = False): """ Send cmd to device @@ -342,9 +348,8 @@ class ChameleonCom: if callable(callback): task['callback'] = callback self.send_data_queue.put(task) - return self - def send_cmd_sync(self, cmd: int, data: bytearray or bytes or list or int = None, status: int = 0, + def send_cmd_sync(self, cmd: int, data: Union[bytes, None] = None, status: int = 0, timeout: int = 3) -> Response: """ Send cmd to device, and block receive data. @@ -355,13 +360,11 @@ class ChameleonCom: :param timeout: wait response timeout :return: response data """ - if isinstance(data, int): - data = [data] # warp array. if len(self.commands): # check if chameleon can understand this command if cmd not in self.commands: - raise CMDInvalidException(f"This device doesn't declare that it can support this command: {cmd}.\nMake " - f"sure firmware is up to date and matches client") + raise CMDInvalidException(f"This device doesn't declare that it can support this command: {cmd}.\n" + f"Make sure firmware is up to date and matches client") # first to send cmd, no callback mode(sync) self.send_cmd_auto(cmd, data, status, None, timeout) # wait cmd start process diff --git a/software/script/chameleon_utils.py b/software/script/chameleon_utils.py index 9bf2a02..eef07ce 100644 --- a/software/script/chameleon_utils.py +++ b/software/script/chameleon_utils.py @@ -1,7 +1,8 @@ import argparse import colorama from functools import wraps -from typing import Union +# once Python3.10 is mainstream, we can replace Union[str, None] by str | None +from typing import Union, Callable, Any from prompt_toolkit.completion import Completer, NestedCompleter, WordCompleter from prompt_toolkit.completion.base import Completion from prompt_toolkit.document import Document @@ -38,12 +39,12 @@ class ArgumentParserNoExit(argparse.ArgumentParser): we must raise exception to stop parse """ - def __init__(self, **args): - super().__init__(*args) + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) self.add_help = False self.description = "Please enter correct parameters" - def exit(self, status: int = ..., message: str or None = ...): + def exit(self, status: int = 0, message: Union[str, None] = None): if message: raise ParserExitIntercept(message) @@ -97,7 +98,7 @@ class ArgumentParserNoExit(argparse.ArgumentParser): print('\n'.join(lines)) -def expect_response(accepted_responses: Union[int, list[int]]): +def expect_response(accepted_responses: Union[int, list[int]]) -> Callable[..., Any]: """ Decorator for wrapping a Chameleon CMD function to check its response for expected return codes and throwing an exception otherwise @@ -116,7 +117,7 @@ def expect_response(accepted_responses: Union[int, list[int]]): status_string = f"Unexpected response and unknown status {ret.status}" raise UnexpectedResponseError(status_string) - return ret.data + return ret.parsed return error_throwing_func @@ -133,11 +134,12 @@ class CLITree: :param cls: A BaseCLIUnit instance handling the command """ - def __init__(self, name=None, help_text=None, fullname=None, children=None, cls=None, root=False) -> None: - self.name: str = name - self.help_text: str = help_text - self.fullname: str = fullname if fullname else name - self.children: list[CLITree] = children if children else list() + def __init__(self, name: str = "", help_text: Union[str, None] = None, fullname: Union[str, None] = None, + children: Union[list["CLITree"], None] = None, cls=None, root=False) -> None: + self.name = name + self.help_text = help_text + self.fullname = fullname if fullname else name + self.children = children if children else list() self.cls = cls self.root = root if self.help_text is None and not root: