Files
Balsam_167/tests/can_telemetry/run_test.py
T
Andrey 13f1a2c5fd Добавить heartbeat и передачу CAN по изменению и таймеру
Для всех Mode передавать пакеты при обновлении данных и повторять каждый через 1300 мс от последнего CAN ACK. Отдельно отправлять heartbeat раз в секунду. Сохранить маски рассылки, отключение потоков нулевым периодом и неблокирующее восстановление после ошибок TX.

Проверены границы периодов, все Mode, занятый mailbox, отмена без CANAA, отсутствие ACK, переполнение Timer1, первый снимок и справедливость потоков. Изменённые модули ecan.c и main.c скомпилированы компилятором TI C2000.
2026-10-16 23:17:00 +03:00

23 lines
3.9 KiB
Python

from pathlib import Path # Находим production-исходники относительно установленного теста либо явно выбранного проверяемого снимка.
import os # Подключаем runtime GCC только дочернему процессу теста.
import re # Извлекаем настоящий ISR с сохранением вложенных условий и без отдельной копии алгоритма ошибки.
import subprocess # Компилируем и выполняем поведенческую модель передатчика.
import sys # Разрешаем проверить изолированный снимок коммита вместо текущего рабочего дерева.
import tempfile # Создаём исполняемый файл проверки только во временном каталоге теста.
test = Path(__file__).resolve().parent # Производственная установка хранит этот файл в tests/can_telemetry.
root = Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else test.parents[1] # При обычном запуске проверяет текущий проект прошивки.
source = (root / 'Source/Internal/ecan.c').read_text(encoding='utf-8') # Изменённый production-файл и русские комментарии обязаны быть корректным UTF-8.
clean = re.sub(r'/\*.*?\*/|//[^\n]*', '', source, flags=re.S) # Скобки комментариев не влияют на выделение ISR.
match = re.search(r'interrupt\s+void\s+CANa_reset_err\(void\)\s*\{', clean) # Отличаем определение обработчика от прототипа и регистрации IRQ.
assert match # Отсутствующий обработчик не считается успешной проверкой передачи.
end, depth = match.end(), 1 # Начинаем поиск после внешней открывающей скобки функции.
while depth: depth += (clean[end] == '{') - (clean[end] == '}'); end += 1 # Сохраняем все вложенные аппаратные проверки оригинала.
model = (test / 'model.c').read_text(encoding='utf-8').replace('/* TIMEOUT_ISR */', clean[match.start():end]) # Сама реализация телеметрии подключается напрямую через production-header.
compiler = os.environ.get('CC', 'C:/msys64/mingw64/bin/gcc.exe') # Поддерживаем установленный GCC и явный выбор другого совместимого компилятора.
environment = dict(os.environ); environment['PATH'] = str(Path(compiler).parent) + os.pathsep + environment.get('PATH', '') # Не меняем системный PATH пользователя или настройки CCS.
with tempfile.TemporaryDirectory(prefix='can-telemetry-166-') as directory: # Временные файлы не попадают в источник прошивки или Git-коммит.
cfile, exe = Path(directory) / 'test.c', Path(directory) / 'test.exe' # Отделяем исходник модели и её исполняемый результат.
cfile.write_text(model, encoding='utf-8') # Тестовые комментарии сохраняются в той же кодировке UTF-8.
subprocess.run([compiler, '-std=c99', '-O3', '-Wall', '-Wextra', '-Werror', '-I' + str(root / 'Source/Internal/Include'), str(cfile), '-o', str(exe)], env=environment, check=True) # Все предупреждения нового алгоритма считаются ошибками до создания коммита.
subprocess.run([str(exe)], env=environment, check=True, timeout=10) # Зависание очереди или нарушенный тайминг завершают проверку ошибкой.