#!/usr/bin/env python3 import socket import struct import time import os import sys import threading from typing import Optional, Tuple, Dict, List from datetime import datetime # =================== Форматы структур =================== # Формат для входящей структуры from_dev_bcst_t (начальный пакет) INPUT_STRUCT_FORMAT = " Tuple[bytes, dict]: """Создание команды структуры from_serv_update_t""" vers = STR_VERSION magic_number = MAGIC_NUMER my_port = SU_PORT n_frame = self.counter_frames if ext_area is None: ext_area = bytes(256) self.counter_frames += 1 cmd_data_without_crc = struct.pack( CMD_STRUCT_FORMAT[:-1], vers, magic_number, my_port, n_frame, cmd, f_address, size, ext_area ) crc_struct = calculate_crc32(cmd_data_without_crc) cmd_data = struct.pack( CMD_STRUCT_FORMAT, vers, magic_number, my_port, n_frame, cmd, f_address, size, ext_area, crc_struct ) cmd_info = { 'vers': vers, 'magic_number': magic_number, 'my_port': my_port, 'n_frame': n_frame, 'cmd': cmd, 'cmd_name': get_command_name(cmd), 'f_address': f_address, 'size': size, 'ext_area_size': len(ext_area), 'crc_struct': crc_struct, 'device_ip': self.device_info['public_ip'], 'device_port': self.device_info.get('device_port', SU_PORT), 'device_local_ip': self.device_info.get('local_ip', ''), 'device_local_port': self.device_info.get('local_port', 0), 'device_n_frame': self.device_info['n_frame'], 'packet_size': len(cmd_data) } return cmd_data, cmd_info # =================== Вспомогательные функции =================== def calculate_crc32(data: bytes, init_value: int = 0xFFFFFFFF) -> int: """Вычисление CRC-32 по алгоритму из спецификации""" crc = init_value for byte in data: val = (crc ^ byte) & 0xFF for _ in range(8): if val & 1: val = (val >> 1) ^ CRC32_POLY else: val = val >> 1 crc = val ^ (crc >> 8) return crc ^ 0xFFFFFFFF def format_mac(mac_bytes: bytes) -> str: """Форматирование MAC-адреса в читаемый вид""" if isinstance(mac_bytes, bytes) and len(mac_bytes) == 6: return ":".join(f"{b:02X}" for b in mac_bytes) else: return f"Некорректный MAC: {mac_bytes!r}" def format_ip(ip_bytes: bytes) -> str: """Форматирование IP-адреса в читаемый вид""" if isinstance(ip_bytes, bytes) and len(ip_bytes) == 4: return ".".join(str(b) for b in ip_bytes) else: return f"Некорректный IP: {ip_bytes!r}" def parse_config_ext_area(ext_area: bytes) -> dict: """Парсинг ext_area для команды CMD_GET_CONFIG""" config = {} if not isinstance(ext_area, bytes) or len(ext_area) != 256: return {"error": "Некорректный формат ext_area"} try: if len(ext_area) >= 4: jumpers_mask = struct.unpack('> 0) & 1 config['bit0'] = bit0 config['bit0_description'] = "Нулевой бит установлен (1)" if bit0 == 1 else "Нулевой бит сброшен (0)" return config except Exception as e: return {"error": f"Ошибка парсинга конфигурации: {e}", "raw_data": ext_area.hex()} def get_result_description(get_cmd: int, get_result: int) -> str: """Получение описания результата выполнения команды""" result_codes = { OP_ERR: "OP_ERR - Общая ошибка выполнения", OP_OK: "OP_OK - Операция выполнена успешно", OP_FLASH_LOCK: "OP_FLASH_LOCK - Flash память заблокирована", OP_ADDRESS_ERR: "OP_ADDRESS_ERR - Ошибка адреса", OP_PAGE_NO_ERASE: "OP_PAGE_NO_ERASE - Не удалось стереть страницу", OP_PAGE_NO_WRITE: "OP_PAGE_NO_WRITE - Не удалось записать страницу", OP_CRC_BAD: "OP_CRC_BAD - Неверная контрольная сумма", OP_NOT_VALID_PASS: "OP_NOT_VALID_PASS - Неверный пароль", OP_OB_NO_ERASE: "OP_OB_NO_ERASE - Не удалось стереть Option Bytes", OP_OB_NO_WRITE: "OP_OB_NO_WRITE - Не удалось записать Option Bytes", OP_ERR_PASS: "OP_ERR_PASS - Ошибка пароля", OP_BL_LOCK_PIN: "OP_BL_LOCK_PIN - Bootloader заблокирован пин-кодом" } activate_bl_results = { OP_OK: "OK: Bootloader активирован", OP_FLASH_LOCK: "ОШИБКА: Flash память заблокирована", OP_NOT_VALID_PASS: "ОШИБКА: Неверный пароль", OP_ERR_PASS: "ОШИБКА: Ошибка пароля", OP_BL_LOCK_PIN: "ОШИБКА: Bootloader заблокирован пин-кодом", } base_desc = result_codes.get(get_result, f"Неизвестный код результата: {get_result}") if get_cmd == CMD_ACTIVATE_BL: specific_desc = activate_bl_results.get(get_result) if specific_desc: return specific_desc return base_desc def get_command_name(cmd_code: int) -> str: """Получение имени команды по коду""" cmd_names = { CMD_NOP: "CMD_NOP", CMD_ACTIVATE_BL: "CMD_ACTIVATE_BL", CMD_UNBLOCK_FLASH: "CMD_UNBLOCK_FLASH", CMD_ERASE_UP_FLASH: "CMD_ERASE_UP_FLASH", CMD_SET_PASS_BIN: "CMD_SET_PASS_BIN", CMD_WRITE_UP_FLASH: "CMD_WRITE_UP_FLASH", CMD_GETCRC_FLASH: "CMD_GETCRC_FLASH", CMD_JUMP_TO_APP: "CMD_JUMP_TO_APP", CMD_GET_CONFIG: "CMD_GET_CONFIG", CMD_ERASE_SECTOR_SPI: "CMD_ERASE_SECTOR_SPI", CMD_WRITE_PAGE_SPI: "CMD_WRITE_PAGE_SPI", CMD_READ_PAGE_SPI: "CMD_READ_PAGE_SPI", CMD_ID_CHIP_SPI: "CMD_ID_CHIP_SPI", } return cmd_names.get(cmd_code, f"Неизвестная команда ({cmd_code})") def find_firmware_file_by_crc(crc_value: int) -> Optional[dict]: """ Поиск файла прошивки по CRC значению. Поддерживает форматы: - .bct: одиночный файл прошивки (122880 байт) - .xct: контейнер с двумя прошивками (1171456 байт), из которого извлекается BCT часть и WCT часть Возвращает словарь с информацией о файле или None """ crc_hex = f"{crc_value:08X}" # Проверяем наличие .bct файла filename_bct = f"{crc_hex}.bct" filename_bct_upper = filename_bct.upper() if os.path.exists(filename_bct) or os.path.exists(filename_bct_upper): actual_filename = filename_bct if os.path.exists(filename_bct) else filename_bct_upper try: with open(actual_filename, 'rb') as f: file_data = f.read() print(f"Найден файл: {actual_filename} (тип: BCT контейнер)") print(f"Размер: {len(file_data)} байт") if len(file_data) != APPLICATION_SIZE: print(f"ОШИБКА: Размер файла {actual_filename} ({len(file_data)} байт) не соответствует ожидаемому ({APPLICATION_SIZE} байт)") return None file_crc = calculate_crc32(file_data) return { 'filename': actual_filename, 'firmware_data': file_data, # BCT данные для MCU Flash 'wct_data': None, # Нет WCT данных для .bct 'expected_crc': file_crc, 'actual_crc': file_crc, 'file_type': 'BCT', 'is_xct': False } except Exception as e: print(f"Ошибка при чтении файла .bct: {e}") return None # Проверяем наличие .xct файла filename_xct = f"{crc_hex}.xct" filename_xct_upper = filename_xct.upper() if os.path.exists(filename_xct) or os.path.exists(filename_xct_upper): actual_filename = filename_xct if os.path.exists(filename_xct) else filename_xct_upper try: with open(actual_filename, 'rb') as f: xct_data = f.read() print(f"Найден файл: {actual_filename} (тип: XCT контейнер)") print(f"Общий размер XCT: {len(xct_data)} байт") if len(xct_data) != XCT_SIZE: print(f"ОШИБКА: Размер XCT файла ({len(xct_data)} байт) не соответствует ожидаемому ({XCT_SIZE} байт)") return None # Проверяем CRC всего XCT контейнера xct_crc = calculate_crc32(xct_data) xct_crc_str = f"{xct_crc:08X}" if actual_filename.upper() == xct_crc_str + ".XCT": print(f"CRC32 XCT контейнера совпадает с именем: 0x{xct_crc_str}") else: print(f"ПРЕДУПРЕЖДЕНИЕ: Имя файла ({actual_filename}) не совпадает с вычисленным CRC32 XCT контейнера (0x{xct_crc_str})") # Извлекаем BCT часть (первые 122880 байт) bct_data = xct_data[:BCT_SIZE] # Извлекаем WCT часть (следующие 1048576 байт) wct_data = xct_data[BCT_SIZE:BCT_SIZE + WCT_SIZE] print(f"Извлечена BCT часть: {len(bct_data)} байт (CRC: 0x{calculate_crc32(bct_data):08X})") print(f"Извлечена WCT часть: {len(wct_data)} байт") print(f"Ожидаемое количество страниц MCU Flash: {TOTAL_PAGES} по {SIZE_PAGE_FLASH} байт") print(f"Ожидаемое количество страниц SPI Flash: {WCT_PAGES} по {SPI_FLASH_PAGE_SIZE} байт") # Вычисляем CRC BCT части - это реальный CRC прошивки MCU bct_crc = calculate_crc32(bct_data) print(f"CRC32 BCT части (прошивка MCU): 0x{bct_crc:08X}") print(f"Ожидаемый CRC (из имени файла): 0x{crc_value:08X}") return { 'filename': actual_filename, 'firmware_data': bct_data, # BCT данные для MCU Flash 'wct_data': wct_data, # WCT данные для SPI Flash 'expected_crc': crc_value, # CRC из имени файла (для сравнения ДО обновления) 'actual_crc': bct_crc, # Реальный CRC прошивки MCU 'file_type': 'XCT', 'is_xct': True } except Exception as e: print(f"Ошибка при чтении файла .xct: {e}") return None # Файл не найден print(f"Файлы {crc_hex}.bct и {crc_hex}.xct не найдены") return None def print_progress_bar(iteration: int, total: int, prefix: str = '', suffix: str = '', length: int = 40, fill: str = '=', print_end: str = "\r"): """ Вывод прогресс-бара """ percent = ("{0:.1f}").format(100 * (iteration / float(total))) filled_length = int(length * iteration // total) bar = fill * filled_length + '-' * (length - filled_length) print(f'\r{prefix} |{bar}| {percent}% {suffix}', end=print_end, flush=True) if iteration == total: print() # =================== Функции для работы со структурами =================== def parse_initial_packet(data: bytes) -> Optional[dict]: """Парсинг начального пакета from_dev_bcst_t""" if len(data) < INPUT_PACKET_SIZE: return None try: unpacked = struct.unpack(INPUT_STRUCT_FORMAT, data[:INPUT_PACKET_SIZE]) packet = { 'vers': unpacked[0], 'magic_number': unpacked[1], 'ver_hw': unpacked[2], 'ver_bl': unpacked[3], 'num_key': unpacked[4], 'crc_bl': unpacked[5], 'my_mac': unpacked[6], 'tmp': unpacked[7], 'my_ip': unpacked[8], 'my_port': unpacked[9], 'crc_uflash': unpacked[10], 'n_frame': unpacked[11], 'ext_area': unpacked[12], 'crc_struct': unpacked[13] } return packet except struct.error: return None def parse_response_packet(data: bytes) -> Optional[dict]: """Парсинг ответного пакета from_dev_update_t""" if len(data) < RESPONSE_PACKET_SIZE: if len(data) >= INPUT_PACKET_SIZE: initial_packet = parse_initial_packet(data) if initial_packet: return { 'vers': initial_packet['vers'], 'magic_number': initial_packet['magic_number'], 'my_mac': initial_packet['my_mac'], 'tmp': initial_packet['tmp'], 'my_ip': initial_packet['my_ip'], 'my_port': initial_packet['my_port'], 'n_frame': initial_packet['n_frame'], 'get_cmd': CMD_NOP, 'get_result': OP_ERR, 'time_proc': 0, 'ext_area': initial_packet['ext_area'], 'crc_struct': initial_packet['crc_struct'] } return None try: unpacked = struct.unpack(RESPONSE_STRUCT_FORMAT, data[:RESPONSE_PACKET_SIZE]) packet = { 'vers': unpacked[0], 'magic_number': unpacked[1], 'my_mac': unpacked[2], 'tmp': unpacked[3], 'my_ip': unpacked[4], 'my_port': unpacked[5], 'n_frame': unpacked[6], 'get_cmd': unpacked[7], 'get_result': unpacked[8], 'time_proc': unpacked[9], 'ext_area': unpacked[10], 'crc_struct': unpacked[11] } return packet except struct.error: return None # =================== TCP-функции =================== def send_command_tcp(dev_conn: DeviceConnection, cmd: int, cmd_name: str, custom_ext_area: bytes = None, verbose: bool = True) -> Tuple[bool, Optional[dict]]: """Отправка команды через TCP соединение""" cmd_data, cmd_info = dev_conn.create_command_packet( cmd, 0, 0, custom_ext_area ) if verbose: print(f"Отправка команды {cmd_name}...") try: # Отправляем команду dev_conn.conn.sendall(cmd_data) # Ждем ответ response_data = recv_exact(dev_conn.conn, RESPONSE_PACKET_SIZE, timeout=3.0) if not response_data: if verbose: print(f"Таймаут получения ответа на команду {cmd_name}") return False, None response_packet = parse_response_packet(response_data) if response_packet: data_without_crc = response_data[:RESPONSE_PACKET_SIZE-4] calculated_crc = calculate_crc32(data_without_crc) crc_valid = (response_packet['crc_struct'] == calculated_crc) if crc_valid: if response_packet['get_cmd'] == cmd: result = response_packet['get_result'] if result == OP_OK: if verbose: print(f"Команда {cmd_name} выполнена успешно") return True, response_packet else: result_desc = get_result_description(cmd, result) print(f"Ошибка {cmd_name}: {result_desc}") return False, response_packet else: if verbose: print(f"Получена неожиданная команда в ответе: {response_packet['get_cmd']}") return False, response_packet else: if verbose: print(f"Неверный CRC в ответе на команду {cmd_name}") return False, response_packet else: if verbose: print(f"Не удалось распарсить ответ на команду {cmd_name}") return False, None except socket.timeout: if verbose: print(f"Таймаут при отправке/получении команды {cmd_name}") return False, None except socket.error as e: if verbose: print(f"Ошибка сокета при выполнении команды {cmd_name}: {e}") return False, None except Exception as e: if verbose: print(f"Неожиданная ошибка при выполнении команды {cmd_name}: {e}") return False, None def recv_exact(conn, size: int, timeout: float = 3.0) -> Optional[bytes]: """Получение точного количества байт через TCP""" try: conn.settimeout(timeout) data = b'' remaining = size while remaining > 0: chunk = conn.recv(remaining) if not chunk: return None data += chunk remaining -= len(chunk) return data except socket.timeout: return None except socket.error: return None def process_flash_commands_tcp(dev_conn: DeviceConnection, config_response: dict, verbose: bool = True) -> Tuple[bool, bool]: """Обработка команд для работы с Flash памятью через TCP, если бит 0 установлен""" config_data = parse_config_ext_area(config_response['ext_area']) if 'error' in config_data: if verbose: print(f"Не удалось проанализировать конфигурации: {config_data['error']}") return False, False bit0 = config_data.get('bit0', 0) if bit0 == 1: if verbose: print("Перемычка SA1 снята, выполняем подготовку Flash...") unblock_success, unblock_response = send_command_tcp( dev_conn, CMD_UNBLOCK_FLASH, "CMD_UNBLOCK_FLASH", verbose=verbose ) if unblock_success and unblock_response and unblock_response['get_result'] == OP_OK: if verbose: print("Flash разблокирована") erase_success, erase_response = send_command_tcp( dev_conn, CMD_ERASE_UP_FLASH, "CMD_ERASE_UP_FLASH", verbose=verbose ) if erase_success and erase_response and erase_response['get_result'] == OP_OK: if verbose: print("Flash стерта") return True, True else: if verbose: print("Ошибка стирания Flash") return True, False else: if verbose: print("Ошибка разблокировки Flash") return False, False else: if verbose: print("Перемычка SA1 установлена, Flash защищена от записи") return False, False def send_write_up_flash_command_tcp(dev_conn: DeviceConnection, f_address: int, page_data: bytes, verbose: bool = False) -> Tuple[bool, Optional[dict]]: """Отправка команды CMD_WRITE_UP_FLASH через TCP""" try: ext_area = bytearray(256) ext_area[:len(page_data)] = page_data cmd_data, cmd_info = dev_conn.create_command_packet( CMD_WRITE_UP_FLASH, f_address, SIZE_PAGE_FLASH, bytes(ext_area) ) # Отправляем команду dev_conn.conn.sendall(cmd_data) # Ждем ответ response_data = recv_exact(dev_conn.conn, RESPONSE_PACKET_SIZE, timeout=3.0) if not response_data: return False, None response_packet = parse_response_packet(response_data) if response_packet: if response_packet['get_cmd'] == CMD_WRITE_UP_FLASH: if response_packet['get_result'] == OP_OK: return True, response_packet else: return False, response_packet return False, None except socket.timeout: return False, None except Exception: return False, None def program_flash_memory_tcp(dev_conn: DeviceConnection, firmware_data: bytes) -> Tuple[bool, int]: """Программирование Flash памяти через TCP соединение""" print(f"\nНачало программирования Flash:") print(f"Размер файла: {len(firmware_data)} байт") print(f"Количество страниц: {TOTAL_PAGES}") print(f"Начальный адрес: 0x{APPLICATION_BEGIN:08X}") prog_address = APPLICATION_BEGIN offset_address = 0 successful_pages = 0 total_pages = TOTAL_PAGES print_progress_bar(0, total_pages, prefix='Прогресс:', suffix='Начало', length=40, fill='=') for page_idx in range(total_pages): start_offset = offset_address end_offset = offset_address + SIZE_PAGE_FLASH page_data = firmware_data[start_offset:end_offset] success, response = send_write_up_flash_command_tcp( dev_conn, prog_address, page_data, verbose=False ) if success and response and response['get_result'] == OP_OK: successful_pages += 1 if (page_idx + 1) % 10 == 0 or (page_idx + 1) == total_pages: current_address = APPLICATION_BEGIN + (page_idx * SIZE_PAGE_FLASH) suffix = f"Страница {page_idx+1}/{total_pages} | Адр: 0x{current_address:08X}" print_progress_bar(page_idx + 1, total_pages, prefix='Прогресс:', suffix=suffix, length=40, fill='=') prog_address += SIZE_PAGE_FLASH offset_address += SIZE_PAGE_FLASH else: print(f"\nОшибка записи страницы {page_idx} по адресу 0x{prog_address:08X}") return False, successful_pages print(f"\nПрограммирование завершено!") print(f"Успешно записано страниц: {successful_pages}/{total_pages}") return True, successful_pages # =================== Функции для работы с SPI Data Flash =================== def send_spi_command_tcp(dev_conn: DeviceConnection, cmd: int, cmd_name: str, f_address: int = 0, size: int = 0, custom_ext_area: bytes = None, verbose: bool = True) -> Tuple[bool, Optional[dict]]: """Отправка SPI команды через TCP соединение""" cmd_data, cmd_info = dev_conn.create_command_packet( cmd, f_address, size, custom_ext_area ) if verbose: print(f"Отправка SPI команды {cmd_name}...") try: dev_conn.conn.sendall(cmd_data) # Ждем ответ response_data = recv_exact(dev_conn.conn, RESPONSE_PACKET_SIZE, timeout=3.0) if not response_data: if verbose: print(f"Таймаут получения ответа на SPI команду {cmd_name}") return False, None response_packet = parse_response_packet(response_data) if response_packet: data_without_crc = response_data[:RESPONSE_PACKET_SIZE-4] calculated_crc = calculate_crc32(data_without_crc) crc_valid = (response_packet['crc_struct'] == calculated_crc) if crc_valid: if response_packet['get_cmd'] == cmd: result = response_packet['get_result'] if result == OP_OK: if verbose: print(f"SPI команда {cmd_name} выполнена успешно") return True, response_packet else: result_desc = get_result_description(cmd, result) print(f"Ошибка SPI команды {cmd_name}: {result_desc}") return False, response_packet else: if verbose: print(f"Получена неожиданная команда в ответе на SPI команду: {response_packet['get_cmd']}") return False, response_packet else: if verbose: print(f"Неверный CRC в ответе на SPI команду {cmd_name}") return False, response_packet else: if verbose: print(f"Не удалось распарсить ответ на SPI команду {cmd_name}") return False, None except socket.timeout: if verbose: print(f"Таймаут при отправке/получении SPI команды {cmd_name}") return False, None except socket.error as e: if verbose: print(f"Ошибка сокета при выполнении SPI команды {cmd_name}: {e}") return False, None except Exception as e: if verbose: print(f"Неожиданная ошибка при выполнении SPI команды {cmd_name}: {e}") return False, None def erase_spi_wct_zone(dev_conn: DeviceConnection, verbose: bool = True) -> bool: """Стирание зоны WCT в SPI Flash (256 секторов начиная с сектора 256)""" # Расчет начального сектора: адрес 0x00100000 / 4KB = 256 сектор start_sector = WCT_START_ADDRESS // SPI_FLASH_SECTOR_SIZE # 1048576 / 4096 = 256 print(f"\nСтирание зоны WCT в SPI Flash:") print(f" Размер зоны: {WCT_SIZE} байт ({WCT_SECTORS} секторов по {SPI_FLASH_SECTOR_SIZE} байт)") print(f" Начальный сектор: {start_sector} (адрес: 0x{WCT_START_ADDRESS:08X})") for i in range(WCT_SECTORS): sector_number = start_sector + i # Упаковываем номер сектора в ext_area ext_area = bytearray(256) struct.pack_into(' Tuple[bool, int]: """Запись WCT данных в SPI Flash""" print(f"\nЗапись WCT в SPI Flash:") print(f" Размер WCT: {len(wct_data)} байт") print(f" Количество страниц: {WCT_PAGES}") print(f" Начальный адрес: 0x{WCT_START_ADDRESS:08X}") if len(wct_data) != WCT_SIZE: print(f"ОШИБКА: Размер WCT данных ({len(wct_data)} байт) не соответствует ожидаемому ({WCT_SIZE} байт)") return False, 0 successful_pages = 0 offset_address = WCT_START_ADDRESS print_progress_bar(0, WCT_PAGES, prefix='Запись SPI:', suffix='Начало', length=40, fill='=') for page_idx in range(WCT_PAGES): start_offset = page_idx * SPI_FLASH_PAGE_SIZE end_offset = start_offset + SPI_FLASH_PAGE_SIZE page_data = wct_data[start_offset:end_offset] # Упаковываем данные страницы в ext_area ext_area = bytearray(256) ext_area[:len(page_data)] = page_data success, response = send_spi_command_tcp( dev_conn, CMD_WRITE_PAGE_SPI, f"WRITE_PAGE_SPI", f_address=offset_address, size=SPI_FLASH_PAGE_SIZE, custom_ext_area=bytes(ext_area), verbose=False ) if success and response and response['get_result'] == OP_OK: successful_pages += 1 # Показываем прогресс каждые 100 страниц или в конце if (page_idx + 1) % 100 == 0 or (page_idx + 1) == WCT_PAGES: current_addr = WCT_START_ADDRESS + (page_idx * SPI_FLASH_PAGE_SIZE) suffix = f"Страница {page_idx+1}/{WCT_PAGES} | Адр: 0x{current_addr:08X}" print_progress_bar(page_idx + 1, WCT_PAGES, prefix='Запись SPI:', suffix=suffix, length=40, fill='=') offset_address += SPI_FLASH_PAGE_SIZE else: print(f"\nОшибка записи страницы {page_idx} по адресу 0x{offset_address:08X}") return False, successful_pages print(f"\nЗапись WCT в SPI Flash завершена!") print(f"Успешно записано страниц: {successful_pages}/{WCT_PAGES}") return True, successful_pages def get_spi_chip_id(dev_conn: DeviceConnection, verbose: bool = True) -> Optional[int]: """Получение ID чипа SPI Flash (диагностическая функция)""" success, response = send_spi_command_tcp( dev_conn, CMD_ID_CHIP_SPI, "CMD_ID_CHIP_SPI", verbose=verbose ) if success and response and response['get_result'] == OP_OK: ext_area = response['ext_area'] if len(ext_area) >= 4: chip_id = struct.unpack('= 4: flags = struct.unpack('> 0) & 1 go_to_up1 = (flags >> 1) & 1 dhcp_configured = (flags >> 2) & 1 global_server_configured = (flags >> 3) & 1 device_info = { 'public_ip': device_ip, 'local_ip': device_local_ip, 'local_port': device_local_port, 'device_port': addr[1], 'n_frame': device_n_frame, 'mac': initial_packet['my_mac'], 'address': addr, 'crc_bl': initial_packet['crc_bl'], 'programming_possible': programming_possible, 'go_to_up1': go_to_up1, 'dhcp_configured': dhcp_configured, 'global_server_configured': global_server_configured } print(f"\nУСТРОЙСТВО ПОДКЛЮЧИЛОСЬ:") print(f" Публичный IP: {device_ip}") print(f" Порт: {addr[1]}") print(f" Локальный адрес устройства: {device_local_ip}:{device_local_port}") print(f" MAC: {format_mac(initial_packet['my_mac'])}") print(f" CRC бутлоадера: 0x{initial_packet['crc_bl']:08X}") print(f" Запрашиваемый CRC файла: 0x{required_crc:08X}") print(f" Используемый файл: {filename} (тип: {file_type})") print(f" Ожидаемый CRC прошивки: 0x{expected_crc:08X}") if file_type == 'XCT': print(f" Реальный CRC прошивки: 0x{actual_crc:08X}") print(f" Флаги состояния BL:") print(f" • Программирование возможно: {'ДА' if programming_possible else 'НЕТ'} (бит 0)") print(f" • Переход на UP1: {'ДА' if go_to_up1 else 'НЕТ'} (бит 1)") print(f" • IP адрес получен по DHCP: {'ДА' if dhcp_configured else 'НЕТ'} (бит 2)") print(f" • Настройки для GSU: {'ДА' if global_server_configured else 'НЕТ'} (бит 3)") if is_xct_file and wct_data: print(f" Файл является XCT контейнером, будет выполнена прошивка SPI Data Flash") # Создаем объект соединения с устройством (с локальным счетчиком кадров) dev_conn = DeviceConnection(conn, addr, device_info) # Обрабатываем устройство results = process_device_tcp(dev_conn, initial_packet, device_info, firmware_data, wct_data, expected_crc, actual_crc, is_xct_file) # Выводим итоги print_device_summary(results) except socket.timeout as e: print(f"Таймаут при обработке соединения от {addr}: {e}") except Exception as e: print(f"Ошибка обработки соединения от {addr}: {e}") import traceback traceback.print_exc() finally: try: conn.close() except: pass print(f"Соединение с {addr} закрыто") def process_device_tcp(dev_conn: DeviceConnection, initial_packet: dict, device_info: dict, firmware_data: bytes, wct_data: Optional[bytes], expected_crc: int, actual_crc: int, is_xct_file: bool) -> dict: """Обработка одного устройства через TCP""" device_key = f"{device_info['public_ip']}_{format_mac(device_info['mac'])}" print(f"\n{'='*80}") print(f"ОБРАБОТКА УСТРОЙСТВА: {device_info['public_ip']} ({format_mac(device_info['mac'])})") print(f"{'='*80}") # Проверяем CRC устройства и ОЖИДАЕМЫЙ CRC device_crc = initial_packet['crc_uflash'] print(f"Проверка CRC устройства:") print(f" CRC устройства: 0x{device_crc:08X}") print(f" Ожидаемый CRC: 0x{expected_crc:08X}") if expected_crc != actual_crc: print(f" Реальный CRC: 0x{actual_crc:08X}") # Проверяем флаг принудительного обновления global force_update # Если CRC совпадают и НЕ включен принудительный режим - пропускаем # Если включен принудительный режим - выполняем обновление даже при совпадении CRC skip_programming = (device_crc == actual_crc) and not force_update if skip_programming: print(f"\nУстройство уже имеет актуальное ПО UP1 (CRC: 0x{device_crc:08X})") print("Обслуживание устройства ЗАВЕРШЕНО - обновление не требуется") # Записываем время обработки processed_devices[device_key] = time.time() return { 'device_ip': device_info['public_ip'], 'device_mac': format_mac(device_info['mac']), 'device_crc': device_crc, 'device_crc_bl': device_info.get('crc_bl', 0), 'expected_crc': expected_crc, 'actual_crc': actual_crc, 'skip_programming': True, 'forced': False, 'initial_packet': True, 'activate_bl': False, 'get_config': False, 'unblock_flash': False, 'erase_flash': False, 'set_pass_bin': False, 'write_up_flash': False, 'write_up_flash_pages': 0, 'getcrc_flash': False, 'getcrc_value': 0, 'jump_to_app': False, 'spi_flash_programmed': False, 'errors': [] } # Если включен принудительный режим и CRC совпадают, показываем предупреждение if force_update and (device_crc == actual_crc): print(f"\nВНИМАНИЕ: Включен режим ПРИНУДИТЕЛЬНОГО обновления!") print(f" CRC устройства совпадает с ожидаемым (0x{device_crc:08X})") print(f" Но будет выполнено принудительное обновление UP1 и SPI Flash") print(f"Требуется обновление ПО UP1:") print(f" CRC устройства: 0x{device_crc:08X}") print(f" Ожидаемый CRC: 0x{expected_crc:08X}") # Только если CRC разные, выполняем остальные команды results = { 'device_ip': device_info['public_ip'], 'device_mac': format_mac(device_info['mac']), 'device_crc': device_crc, 'device_crc_bl': device_info.get('crc_bl', 0), 'expected_crc': expected_crc, 'actual_crc': actual_crc, 'skip_programming': False, 'forced': force_update and (device_crc == actual_crc), 'initial_packet': True, 'activate_bl': False, 'get_config': False, 'unblock_flash': False, 'erase_flash': False, 'set_pass_bin': False, 'write_up_flash': False, 'write_up_flash_pages': 0, 'getcrc_flash': False, 'getcrc_value': 0, 'jump_to_app': False, 'spi_flash_programmed': False, 'spi_flash_pages': 0, 'errors': [] } try: # Очищаем буфер сокета перед началом работы try: dev_conn.conn.settimeout(0.1) while True: junk = dev_conn.conn.recv(1024) if not junk: break except socket.timeout: pass except: pass dev_conn.conn.settimeout(15.0) # ============ ШАГ 2: Отправка команды CMD_ACTIVATE_BL ============ print("\nПодготовка устройства...") activate_bl_success, activate_response = send_command_tcp( dev_conn, CMD_ACTIVATE_BL, "CMD_ACTIVATE_BL", verbose=True ) results['activate_bl'] = activate_bl_success and activate_response and activate_response['get_result'] == OP_OK if not results['activate_bl']: results['errors'].append("CMD_ACTIVATE_BL не выполнена") print(f"Ошибка активации bootloader.") processed_devices[device_key] = time.time() return results # ============ ШАГ 3: Отправка команды CMD_GET_CONFIG ============ get_config_success, config_response = send_command_tcp( dev_conn, CMD_GET_CONFIG, "CMD_GET_CONFIG", verbose=True ) results['get_config'] = get_config_success and config_response and config_response['get_result'] == OP_OK if not results['get_config']: results['errors'].append("CMD_GET_CONFIG не выполнена") print(f"Ошибка получения конфигурации.") processed_devices[device_key] = time.time() return results # ============ ШАГ 4: Отправка команд для работы с Flash памятью ============ unblock_success, erase_success = process_flash_commands_tcp( dev_conn, config_response, verbose=True ) results['unblock_flash'] = unblock_success results['erase_flash'] = erase_success # ============ ШАГ 5: Отправка команды CMD_SET_PASS_BIN ============ PASS_BIN_LOAD = 0x184745bf ext_area_bytes = bytearray(256) ext_area_bytes[0:4] = struct.pack('= 4: device_crc_after = struct.unpack(' 0: print(f" CRC MCU после: 0x{results['getcrc_value']:08X}") print(f" Ожидаемый CRC MCU: 0x{results['expected_crc']:08X}") if results['expected_crc'] != results['actual_crc']: print(f" Реальный CRC MCU: 0x{results['actual_crc']:08X}") # Вывод информации о SPI Flash, если была прошивка if results.get('spi_flash_programmed', False): print(f"\n SPI DATA FLASH:") print(f" Статус: ПРОШИТА") print(f" Записано страниц: {results.get('spi_flash_pages', 0)}/{WCT_PAGES}") if not results['skip_programming']: print(f" CMD_ACTIVATE_BL: {'OK' if results['activate_bl'] else 'ОШИБКА'}") print(f" CMD_GET_CONFIG: {'OK' if results['get_config'] else 'ОШИБКА'}") if results['unblock_flash'] or results['erase_flash']: print(f" CMD_UNBLOCK_FLASH: {'OK' if results['unblock_flash'] else 'ОШИБКА'}") print(f" CMD_ERASE_UP_FLASH: {'OK' if results['erase_flash'] else 'ОШИБКА'}") print(f" CMD_SET_PASS_BIN: {'OK' if results['set_pass_bin'] else 'ОШИБКА'}") print(f" CMD_WRITE_UP_FLASH: {'OK' if results['write_up_flash'] else 'ОШИБКА'}") if results['write_up_flash']: print(f" Записано страниц MCU: {results['write_up_flash_pages']}/{TOTAL_PAGES}") print(f" CMD_GETCRC_FLASH: {'OK' if results['getcrc_flash'] else 'ОШИБКА'}") # Вывод информации о SPI операциях if results.get('spi_flash_programmed', False): print(f" SPI DATA FLASH: ПРОШИТ") elif results.get('spi_flash_programmed') is False and not results['skip_programming']: print(f" SPI DATA FLASH: НЕ ТРЕБУЕТСЯ (BCT файл)") print(f" CMD_JUMP_TO_APP: {'OK' if results['jump_to_app'] else 'ОШИБКА'}") if results['errors']: print(f"\nОшибки:") for error in results['errors']: print(f" - {error}") print(f"{'='*80}") def tcp_server(): """TCP сервер для массового обновления устройств""" print("\n" + "="*80) print("TCP СЕРВЕР ОБНОВЛЕНИЯ ПО TALENTUM GSU1AB-PY-TCP") print("="*80) print(f"Ожидание TCP подключений на порту {SU_PORT}...") global force_update if force_update: print("*** РЕЖИМ ПРИНУДИТЕЛЬНОГО ОБНОВЛЕНИЯ ВКЛЮЧЕН ***") print("*** Все устройства будут перепрошиты независимо от CRC ***") else: print("*** ОБЫЧНЫЙ РЕЖИМ: обновляются только устройства с отличающимся CRC ***") print(f"Поддерживаются форматы файлов: .bct и .xct") print(f"Файлы будут загружаться динамически по запросу устройства") # Выводим список доступных файлов bct_files = [f for f in os.listdir('.') if f.lower().endswith('.bct')] xct_files = [f for f in os.listdir('.') if f.lower().endswith('.xct')] if bct_files: print(f"\nДоступные файлы .bct в текущей директории ({len(bct_files)}):") for i, file in enumerate(sorted(bct_files)): file_size = os.path.getsize(file) crc_part = file.upper().replace('.BCT', '') print(f" {i+1:2d}. {file} ({file_size} байт, CRC: 0x{crc_part})") if xct_files: print(f"\nДоступные файлы .xct в текущей директории ({len(xct_files)}):") for i, file in enumerate(sorted(xct_files)): file_size = os.path.getsize(file) crc_part = file.upper().replace('.XCT', '') print(f" {i+1:2d}. {file} ({file_size} байт, CRC контейнера: 0x{crc_part})") if file_size != XCT_SIZE: print(f" ВНИМАНИЕ: Неправильный размер XCT файла! Ожидается {XCT_SIZE} байт") if not bct_files and not xct_files: print("\nВНИМАНИЕ: Нет файлов .bct или .xct в текущей директории!") print("="*80) print("Для остановки сервера нажмите Ctrl+C") print("="*80) server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.settimeout(1.0) # Таймаут для accept чтобы можно было прервать try: server_socket.bind(('0.0.0.0', SU_PORT)) server_socket.listen(5) print(f"TCP сервер запущен...") while True: try: conn, addr = server_socket.accept() print(f"\nПолучено подключение от {addr}") # Создаем поток для обработки клиента client_thread = threading.Thread( target=handle_client_connection, args=(conn, addr), daemon=True ) client_thread.start() except socket.timeout: # Таймаут accept - просто продолжаем цикл continue except KeyboardInterrupt: print("\n\nОстановка TCP сервера по запросу пользователя...") break except Exception as e: print(f"Ошибка accept: {e}") continue except KeyboardInterrupt: print("\n\nСервер остановлен по запросу пользователя") except Exception as e: print(f"Ошибка TCP сервера: {e}") import traceback traceback.print_exc() finally: server_socket.close() print("TCP сервер остановлен") # =================== Основная функция =================== def main(): """Основная функция - TCP сервер массового обновления""" global force_update # Парсинг аргументов командной строки if len(sys.argv) > 1: if sys.argv[1].lower() in ['-force', '-f', '--force']: force_update = True print("="*80) print("ВНИМАНИЕ: Включен режим ПРИНУДИТЕЛЬНОГО обновления!") print("Устройства будут перепрошиты даже при совпадении CRC") print("="*80) elif sys.argv[1].lower() in ['-help', '-h', '--help']: print("Использование: python GSU_1_TCP.py [опции]") print("") print("Опции:") print(" -force, -f, --force Режим принудительного обновления") print(" (перепрошивать устройства даже если CRC совпадает)") print(" -help, -h, --help Показать эту справку") print("") print("Примеры:") print(" python GSU_1_TCP.py # Обычный режим") print(" python GSU_1_TCP.py -force # Принудительное обновление") return print("Инициализация TCP сервера...") if force_update: print("РЕЖИМ: ПРИНУДИТЕЛЬНОЕ ОБНОВЛЕНИЕ (перезапись всех устройств)") else: print("РЕЖИМ: ОБЫЧНЫЙ (только устройства с отличающимся CRC)") # Проверяем наличие файлов прошивок bct_files = [f for f in os.listdir('.') if f.lower().endswith('.bct')] xct_files = [f for f in os.listdir('.') if f.lower().endswith('.xct')] if not bct_files and not xct_files: print("ПРЕДУПРЕЖДЕНИЕ: В текущей директории нет файлов .bct или .xct!") print("Сервер будет работать, но устройства не смогут обновиться без файлов прошивки") response = input("Продолжить? (y/n): ") if response.lower() != 'y': print("Выход...") return tcp_server() if __name__ == "__main__": main()