Files
Balsam_167/tests/can_vpu_rx/run_test.py
Andrey ee79c7a376 Ускорить обратную связь R0–R3 задатчика для ПМ67
Перенести регистры индикации VPU в быстрый поток CAN, чтобы R3 не ожидал медленного обхода параметров. Сохранить управление светодиодами и настройки периодов EEPROM.

Добавить проверку настоящих масок и планировщика: интервал R3 составляет 120 мс при быстром делителе 2 и 600 мс при делителе 10. Сохранить изменённый исходник в UTF-8.
2026-10-16 16:46:15 +03:00

84 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.
"""Проверяет частоту подтверждения R3 на настоящих масках и планировщике 166.""" # Проверка выполняется на ПК без отправки команд плате.
from pathlib import Path # Находит исходники проекта и отдельно подготовленный вариант маски.
import os # Передаёт библиотеки GCC только дочерним процессам проверки.
import re # Извлекает определения и блоки C без комментариев.
import subprocess # Компилирует и исполняет реальные алгоритмы прошивки в модели времени.
import sys # Позволяет проверить старый и исправленный файлы одинаковым сценарием.
import tempfile # Ограничивает результаты компиляции временным каталогом.
root = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(__file__).resolve().parents[2] # Установленная проверка по умолчанию использует свой проект прошивки.
measure = Path(sys.argv[2]) if len(sys.argv) > 2 else root / 'Source/Internal/measure.c' # Подмена только этого файла позволяет воспроизвести ошибку до применения исправления.
def clean(path): # Удаляет комментарии перед поиском парных скобок C.
raw = path.read_bytes() # Сохраняет возможность чтения старых исходников Windows-1251.
source = raw.decode('utf-8') if _utf8(raw) else raw.decode('cp1251') # Не теряет русские комментарии исходника при выборе кодировки.
return re.sub(r'/\*.*?\*/|//[^\n]*', '', source, flags=re.S) # Скобки в пояснениях не меняют границы блока алгоритма.
def _utf8(raw): # Отличает старую Windows-кодировку от требуемого UTF-8 без замены символов.
try: # Корректный UTF-8 имеет приоритет перед совместимой однобайтовой кодировкой.
raw.decode('utf-8'); return True # Успешное декодирование разрешает чтение современного исходника.
except UnicodeDecodeError: # Старый файл будет прочитан в его исходной кодировке.
return False # Не выдаёт повреждённые комментарии за исходный текст.
def block(source, start): # Выделяет настоящий блок с вложенными условиями.
cursor, depth = source.index('{', start) + 1, 1 # Начинает после открывающей скобки внешнего блока.
while depth: # Ищет соответствующую закрывающую скобку.
depth += (source[cursor] == '{') - (source[cursor] == '}') # Учитывает только скобки очищенного C-кода.
cursor += 1 # Включает завершающую скобку в извлечённый текст.
return source[start:cursor] # Проверка не поддерживает отдельную копию алгоритма прошивки.
package = clean(root / 'Source/Internal/Include/package.h') # Значения режимов и индексов потоков берутся из проекта 166.
names = ('dsk_COMM', 'dsk_BKSD', 'dsk_BKST', 'dsk_EPLT', 'dsk_SHKF', 'dsk_LOAD', 'm_FAST', 'm_SLOW') # Модели нужны только константы, используемые масками.
defines = '\n'.join(re.search(r'^#define\s+' + name + r'\s+\d+', package, re.M).group() for name in names) # Не дублирует номера плат в тестовой карте.
measure_source = clean(measure) # Выбирает проверяемую версию Init_packMask.
mask_function = block(measure_source, measure_source.index('void Init_packMask(')) # Состав быстрой и медленной рассылки полностью берётся из прошивки.
main = clean(root / 'Source/Internal/main.c') # Последовательность передачи берётся из текущего рабочего main.c.
scheduler = block(main, main.index('if(CanGO)')) # Проверяет настоящий учёт занятого mailbox и продвижение адресов.
base_tick = int(re.search(r'^#define\s+CANPOWSE\s+(\d+)', clean(root / 'Source/Internal/Include/measure.h'), re.M).group(1)) # При READY_FREQ=1000 Гц делитель CANPOWSE равен числу миллисекунд базового шага.
model = r'''
#include <stdio.h> /* Выводит измеренные моделью интервалы подтверждения. */
#include <string.h> /* Сбрасывает состояние перед каждым сочетанием периодов EEPROM. */
/* DEFINES */
static int Desk, Kurrent, tpl_cans, CanGO, modbus[131], busy; /* Запас слов повторяет трёхсловное чтение у границы R127. */
static unsigned int Maska[2][8], cancount[2], cancell[2], now, received_at, max_gap, received_count; /* Каждое состояние соответствует маскам, очередям и статистике RX. */
#define Cancount (modbus+0x60) /* R96/R97 задают периоды fast/slow в базовых интервалах CANPOWSE. */
#define CHECK(condition) do { if (!(condition)) { printf("FAIL line %d: %s\n", __LINE__, #condition); return 1; } } while (0) /* Ошибка остаётся видимой и при включённой оптимизации. */
static int CAN_send(int port, int data[], int address) { /* Моделирует только принятие кадра свободным mailbox, не аппаратный ACK. */
(void)port; /* Номер порта не влияет на порядок технологических кадров. */
if(busy) return 0; /* Второй поток в том же тике должен повторить попытку позднее. */
busy=1; /* Один передатчик одновременно принимает только один кадр. */
if(address<=3 && address+2>=3) { /* R3 передаётся и отдельным кадром, и внутри трёхсловного фрагмента. */
if(data[3] != (int)now) return 0; /* Проверяет передачу свежего слова, изменённого перед текущим шагом. */
if(now-received_at>max_gap) max_gap=now-received_at; /* Наблюдает худшее ожидание следующего реального кадра с R3. */
received_at=now; received_count++; /* Фиксирует только принятый передатчиком регистр. */
} /* Завершает наблюдение очередного фрагмента. */
return 1; /* Успешная постановка позволяет настоящему main продвинуть адрес. */
} /* Модель намеренно не заменяет проверку физического CAN на плате. */
/* MASK */
static void step(void) { int i, mask; /* Локальные переменные повторяют контекст фонового цикла main.c. */
/* SCHEDULER */
} /* Выполняет один базовый интервал настоящего планировщика. */
int main(void) { /* Покрывает штатные периоды 166 и варианты из EEPROM, включая отключённый медленный поток. */
const unsigned periods[][2]={{2,101},{10,101},{12,50},{2,0}}; /* Периоды заданы в тиках CANPOWSE: 2 — штатный быстрый делитель VPU-166, 101 — медленный. */
unsigned scenario, tick; /* Сценарии независимы, каждый охватывает сто секунд модельного времени. */
Desk=dsk_EPLT; Init_packMask(); /* Строит реальную карту задатчика выбранной версии прошивки. */
CHECK((Maska[m_FAST][0]&15)==15 && !(Maska[m_SLOW][0]&15)); /* R0–R3 должны подтверждаться быстрым потоком независимо от периода параметров. */
CHECK(Maska[m_FAST][1]==1 && Maska[m_SLOW][6]==0x1FF && Maska[m_SLOW][7]==0xE000); /* Кнопки R16 и медленные параметры R96–R104/R125–R127 сохраняют состав. */
for(scenario=0;scenario<sizeof(periods)/sizeof(periods[0]);scenario++) { /* Проверяет передачу R3 при разных сохранённых настройках платы. */
memset(cancount,0,sizeof(cancount)); memset(cancell,0,sizeof(cancell)); /* Начинает каждый сценарий с первого адреса и нулевого срока. */
received_at=max_gap=received_count=now=0; /* Старые измерения не сокращают наблюдаемую задержку нового сценария. */
Cancount[0]=(int)periods[scenario][0]; Cancount[1]=(int)periods[scenario][1]; /* Использует заданные периоды без сброса EEPROM к defaults. */
for(tick=1;tick<=5000;tick++) { /* Пять тысяч шагов по 20 мс включают много полных обходов медленной очереди. */
now=tick; modbus[3]=(int)tick; busy=0; CanGO=1; step(); /* Меняет R3 перед каждым шагом и завершает предыдущую аппаратную передачу. */
} /* В статистику входят начальное ожидание и все последующие интервалы. */
CHECK(received_count>0 && max_gap<=3*(periods[scenario][0]+1)); /* Три быстрых кадра R0–R2/R3–R5/R16–R18 ограничивают ожидание независимо от slow. */
printf("PASS fast=%u slow=%u: R3 max=%u ms, updates=%u\n",periods[scenario][0],periods[scenario][1],max_gap*BASE_TICK,received_count); /* Переводит тики в миллисекунды базового таймера платы. */
} /* Все сочетания периодов обязаны давать частое подтверждение R3. */
return 0; /* Успех означает прохождение модели, а не испытание физического устройства. */
} /* Завершает регрессию задержки обратной связи VPU. */
''' # Вставки ниже исполняют текущий код прошивки, а не его переписанную модель.
model = model.replace('/* DEFINES */', defines + f'\n#define BASE_TICK {base_tick} /* Длительность базового шага в миллисекундах при таймере 1000 Гц. */').replace('/* MASK */', mask_function).replace('/* SCHEDULER */', scheduler) # Связывает реальные маски и планировщик с моделью одного занятого передатчика.
compiler = Path(os.environ.get('CC', 'C:/msys64/mingw64/bin/gcc.exe')) # Использует установленный GCC без загрузки инструментов.
environment = dict(os.environ, PATH=str(compiler.parent) + os.pathsep + os.environ.get('PATH', '')) # Делает библиотеки компилятора доступными только дочерним программам.
with tempfile.TemporaryDirectory(prefix='vpu-r3-rx-') as directory: # Удаляет только собственные временные результаты после проверки.
source, executable = Path(directory) / 'test.c', Path(directory) / 'test.exe' # Разделяет C-код проверки и исполняемую программу.
source.write_text(model, encoding='utf-8') # Все создаваемые исходники и русские комментарии сохраняются в UTF-8.
subprocess.run([str(compiler), '-std=c99', '-O3', '-Wall', '-Wextra', '-Werror', str(source), '-o', str(executable)], env=environment, check=True) # Требует чистой компиляции настоящих алгоритмов при оптимизации.
subprocess.run([str(executable)], env=environment, check=True, timeout=10) # Зависание либо нарушение ограничения задержки проваливает регрессию.