Files
Balsam_167/tests/can_scheduler/run_test.py
Andrey 31f92e89d9 Обновить сборку прошивки 166, TESTLAMP и кодировку исходников
Перенести настройки выпуска в JSON, добавить переносимый сценарий сборки и проверки BIN. Обновить конфигурации CCS, исходники в UTF-8, обработку TESTLAMP, образы прошивки и регрессионные проверки.
2026-10-17 14:36:59 +03:00

78 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
from pathlib import Path # Проверяет исходники подготовленной прошивки и отдельно исходную версию для чувствительности регрессии.
import re # Выделяет настоящие тела C-функций без комментариев.
import subprocess # Компилирует и выполняет поведенческую проверку C-кода.
import os # Добавляет библиотеки MinGW только в окружение дочерних процессов.
import tempfile # Изолирует сгенерированные файлы GCC от исходников прошивки.
import sys # Позволяет выбрать старую версию планировщика для воспроизведения ошибки.
stage = Path(__file__).resolve().parents[2] # Все результаты проверки сохраняются в каталоге этой конкретной правки.
root = Path(sys.argv[1]) if len(sys.argv) > 1 else stage # По умолчанию проверяет подготовленное исправление.
def clean(path): # Комментарии не участвуют в поиске вложенных блоков алгоритма.
return re.sub(r'/\*.*?\*/|//[^\n]*', '', path.read_bytes().decode('utf-8', errors='replace'), flags=re.S) # Замена старых комментариев не меняет ASCII-код алгоритма.
def block(source, start): # Возвращает выбранный блок вместе со всеми вложенными условиями.
cursor = source.index('{', start) + 1 # Начинает счётчик после открывающей скобки выбранного определения.
depth = 1 # Отслеживает незакрытый внешний блок.
while depth: # Завершает поиск только на парной внешней закрывающей скобке.
depth += (source[cursor] == '{') - (source[cursor] == '}') # Учитывает все вложенные ветви production-кода.
cursor += 1 # Включает последнюю закрывающую скобку в извлечённый фрагмент.
return source[start:cursor] # Не поддерживает отдельную копию проверяемого алгоритма.
ecan = clean(root / 'Source/Internal/ecan.c') # Передатчик проверяется в той же версии, что выбранный планировщик.
functions = [] # Модель включает штатную отправку и действующий обработчик тайм-аутов.
for name in ('CAN_send', 'CANa_reset_err'): # Оба алгоритма должны сохранять согласованный жизненный цикл mailbox.
match = re.search(r'(?:interrupt\s+)?(?:void|int)\s+' + name + r'\([^;{}]*\)\s*\{', ecan) # Поддерживает старую void-версию и исправленную int-версию функции отправки.
assert match, name # Отсутствие функции делает проверку неуспешной.
functions.append(block(ecan, match.start())) # Использует реальное тело, включая все аппаратные условия отказа.
main = clean(root / 'Source/Internal/main.c') # Планировщик извлекается непосредственно из фонового цикла платы.
scheduler = block(main, main.index('if(CanGO)')) # Изолирует только обслуживание очередных периодов CAN.
model = (stage / 'tests/can_timeout/model.inc').read_text(encoding='utf-8') # Повторно использует уже проверенную модель регистров eCAN.
prefix = model.split('/* PRODUCTION */')[0] # Типы регистров остаются общими для проверки тайм-аутов и очереди.
helpers = model[model.index('#define CHECK'):model.index('int main(void)')] # Проверки условий и сброс модели совпадают с существующим тестом.
body = r'''
static int CanGO, modbus[131]; /* Запас двух слов повторяет поведение трёхсловного пакета на верхней границе банка. */
static unsigned int cancount[2], cancell[2], Maska[2][8]; /* Два потока и восемь 16-битных масок повторяют main и Init_packMask. */
#define Cancount (modbus+0x60) /* R96 и R97 задают периоды быстрого и медленного потоков. */
static void step(void) { int i, mask; /* Локальные переменные соответствуют исходному main.c. */
/* SCHEDULER */
} /* Один вызов обслуживает очередной базовый интервал CANPOWSE. */
static void prepare(unsigned fast, unsigned slow) { /* Задаёт именно карту задатчика, без обращения к физической CAN-шине. */
reset_model(); memset(modbus,0,sizeof(modbus)); memset(cancount,0,sizeof(cancount)); memset(cancell,0,sizeof(cancell)); memset(Maska,0,sizeof(Maska)); /* Начинает независимый сценарий. */
Maska[0][1]=1; Maska[1][0]=15; Maska[1][6]=0x1FF; Maska[1][7]=0xE000; /* R16, R0–R3, параметры R96–R104 и R125–R127 соответствуют Init_packMask(dsk_EPLT). */
Cancount[0]=(int)fast; Cancount[1]=(int)slow; modbus[0]=9; modbus[1]=12; /* Использует периоды сценария и задание со снимка пользователя. */
} /* Завершает подготовку начального состояния. */
static void tick(void) { CanGO=1; step(); } /* Таймер даёт ровно один базовый интервал планировщика. */
static void complete(void) { ECanaRegs.CANTRS.all=0; ECanaRegs.CANTRR.all=0; ECanaRegs.CANTA.all=1; } /* Имитирует аппаратное завершение предыдущего кадра до следующего интервала. */
int main(void) { /* Проверяет именно потерю числового фрагмента при одновременном запуске двух потоков. */
unsigned n, scenario, seen[128], address; /* Счётчики наблюдают кадры, действительно поставленные в MBOX0. */
const unsigned periods[][2]={{10,50},{10,101},{10,8},{12,50}}; /* Проверяет текущие defaults, прежний делитель и несовпадающие периоды из EEPROM. */
prepare(10,50); cancount[0]=9; cancount[1]=49; tick(); /* Воспроизводит совпадение быстрого и медленного сроков отправки. */
CHECK(((ECanaMboxes.MBOX0.MDH.all>>16)&0x1FFF)==16); /* Быстрый поток успел поставить кнопки в штатный mailbox. */
CHECK(cancell[1]==0 && cancount[1]==50); /* Занятость сохраняет неподтверждённый адрес R0 и готовность повторной попытки. */
complete(); tick(); /* Следующий свободный интервал должен доставить отложенное задание. */
CHECK(((ECanaMboxes.MBOX0.MDH.all>>16)&0x1FFF)==0 && cancell[1]==3 && cancount[1]==0); /* Адрес продвигается только после принятого mailbox кадра R0–R2. */
CHECK((ECanaMboxes.MBOX0.MDH.all&0xFFFF)==9 && (ECanaMboxes.MBOX0.MDL.all>>16)==12); /* В кадре находятся реальные R0/R1 платы, а не значения из GUI. */
prepare(10,50); ECanaRegs.CANTRS.all=1; wait=1; /* Длительно занятый mailbox не должен вызывать переполнение счётчиков. */
for(n=0;n<100000;n++) tick(); /* Сто тысяч интервалов превышают полный диапазон 16-битного счётчика C28x. */
CHECK(cancount[0]==10 && cancount[1]==50 && cancell[1]==0); /* Оба периода насыщаются, не теряя неподтверждённое задание. */
complete(); tick(); complete(); tick(); /* Последовательно освобождает mailbox для обоих ожидающих потоков. */
CHECK(((ECanaMboxes.MBOX0.MDH.all>>16)&0x1FFF)==0); /* После восстановления шины повторяется прежде отложенный R0. */
for(scenario=0;scenario<sizeof(periods)/sizeof(periods[0]);scenario++) { /* Проверяет устойчивость при разных параметрах EEPROM. */
prepare(periods[scenario][0],periods[scenario][1]); memset(seen,0,sizeof(seen)); /* Сбрасывает статистику каждого сценария. */
for(n=0;n<3000;n++) { /* Одна минута модельного времени при базовом интервале 20 мс. */
complete(); tick(); /* Предыдущий кадр завершается, затем выполняется настоящий планировщик. */
if(ECanaRegs.CANTRS.all&1) { address=(ECanaMboxes.MBOX0.MDH.all>>16)&0x1FFF; CHECK(address<128); seen[address]++; } /* Учитывает только фактически поставленные в передатчик адреса. */
} /* Обновления R16 не должны вытеснять числовую индикацию из медленного потока. */
CHECK(seen[0]>0 && seen[3]>0 && seen[16]>0 && seen[99]>0 && seen[125]>0); /* Все группы из пользовательского журнала, включая пропавший R0, достигают mailbox. */
} /* Завершает матрицу периодов передачи. */
prepare(0,0); tick(); CHECK(ECanaRegs.CANTRS.all==0); /* Нулевые периоды продолжают отключать рассылку без изменения настроек EEPROM. */
puts("PASS scheduler: R0 retried after collision, counters saturated, all VPU groups sent for 4 period pairs"); /* Указывает программные гарантии без заявления об испытании на плате. */
return 0; /* Все сценарии завершены успешно. */
} /* Завершает поведенческую проверку. */
''' # C-модель исполняет извлечённый production-код вместо упрощённого пересказа алгоритма.
code = prefix + '\n'.join(functions) + '\n' + helpers + body.replace('/* SCHEDULER */', scheduler) # Отделяет директиву CHECK переводом строки после последней извлечённой функции.
compiler = Path('C:/msys64/mingw64/bin/gcc.exe') # Использует уже установленный компилятор без загрузок и установки зависимостей.
environment = dict(os.environ, PATH=str(compiler.parent)+os.pathsep+os.environ.get('PATH','')) # Подключает DLL компилятора только дочерней проверке.
temporary = tempfile.TemporaryDirectory(prefix='can-scheduler-') # Удаляет только временные результаты проверки.
cpath, exe = Path(temporary.name) / 'scheduler-test.c', Path(temporary.name) / 'scheduler-test.exe' # Артефакты проверки не смешиваются с образом для микроконтроллера.
cpath.write_text(code, encoding='utf-8') # Сохраняет исходник модели с русскими комментариями в UTF-8.
subprocess.run([str(compiler),'-std=c99','-O3','-Wall','-Wextra','-Wno-unused-parameter','-Werror',*(['-Wno-error=sign-compare'] if root != stage else []),str(cpath),'-o',str(exe)],env=environment,check=True) # Исправленный код требует чистой сборки; старый допускает только известное предупреждение знака для проверки самой потери R0.
subprocess.run([str(exe)],env=environment,check=True,timeout=10) # Бесконечное ожидание или пропущенный R0 немедленно проваливают регрессию.