templates/python/set_devices/signal_diagnostics.py

"""Локальные гипотезы неисправностей по сравнению осциллограмм с эталоном."""  # Правила не требуют сети, модели ИИ или знания имени файла.
import math  # Конечность отсчётов и масштабы защищают признаки от пропусков и нулевого размаха.
import re  # Обозначения физических каналов позволяют отделить Uce, Uge и Ic.
from set_devices.signal_fault_models import identify_fault_models  # Параметрические модели конкурируют по BIC и ошибке контрольных наблюдений.


def _role(name):  # Название файла не используется как доказательство неисправности.
    match = re.match(r'^\s*(uce|uge|ic)\b', name, re.IGNORECASE)  # Узнаём только явные обозначения напряжений ключа, затвора и тока.
    return match.group(1).lower() if match else ''  # Неизвестные обозначения получают только общие статистические гипотезы.


class _Edges:  # Гистерезис отделяет полное повторное переключение от небольшого звона и шума.
    def __init__(self, low, high):  # Пороги задаются относительно размаха эталона в единицах канала.
        self.low, self.high = low, high  # Нижняя и верхняя границы определяют устойчивые состояния.
        self.state, self.start, self.previous = None, None, None  # Начальное состояние и разрыв не считаются фронтами.
        self.events, self.count = [], 0  # Полный счётчик дополняется ограниченным списком для оценки задержки.

    def reset(self):  # Через NaN или большой разрыв нельзя восстанавливать ненаблюдавшееся переключение.
        self.state, self.start, self.previous = None, None, None  # После разрыва ждём нового устойчивого состояния.

    def add(self, stamp, value):  # Последовательный обход не создаёт дополнительную копию многомиллионной записи.
        state = 0 if value <= self.low else 1 if value >= self.high else None  # Зона между порогами сохраняет предыдущее устойчивое состояние.
        if self.state is None:  # Первая наблюдаемая полка задаёт исходное состояние.
            self.state = state  # Попадание только в середину фронта ещё не определяет направление.
        elif state == self.state:  # Возврат к прежней полке отменяет незавершённый переход.
            self.start = None  # Шум внутри полосы не накапливает ложную длительность фронта.
        else:  # Сигнал покинул прежнюю полку и может достичь противоположной.
            if self.start is None:  # Запоминаем начало только один раз на каждый фронт.
                self.start = self.previous if self.previous is not None else stamp  # Шаг дискретизации ограничивает точность начала перехода.
            if state is not None:  # Полное пересечение гистерезиса подтверждает переключение.
                self.count += 1  # Считаем все наблюдаемые фронты независимо от длины записи.
                if len(self.events) < 256:  # Первых 256 переходов достаточно для сводки; память остаётся ограниченной.
                    self.events.append(((self.start + stamp) / 2., stamp - self.start, state))  # Сохраняем середину, длительность в мс и направление фронта.
                self.state, self.start = state, None  # Новая полка становится исходной для следующего перехода.
        self.previous = stamp  # Предыдущая временная метка нужна для оценки длительности на дискретной сетке.


def analyze_comparison(result, cancel=lambda: False, *, circuit=None, channels=None):  # Подбор получает необязательные параметры электрического эквивалента и связанные физические каналы.
    metrics = result.metrics  # Покрытие и допуски уже рассчитаны по полным исходным данным.
    measured_role, reference_role = _role(result.measured.name), _role(result.reference.name)  # Физические роли важнее совпадения единицы «В».
    findings, notes = [], []  # Несколько одновременно наблюдаемых признаков сохраняются в одном отчёте.
    features = {}  # Численные признаки экспортируются для проверки и воспроизведения решения.
    model_report = {}  # До проверки входных данных никакая математическая модель не считается идентифицированной.
    def finish(summary, attention=True):  # Краткая строка и подробности всегда относятся к одному результату.
        details = [summary, f'Каналы: {result.measured.name} ↔ {result.reference.name}.',  # Обозначаем фактически выбранные измерение и эталон.
                   f'Сравнено {metrics["count"]} отсчётов; покрытие {metrics["coverage_pct"]:.4g}%; за допуском {metrics["outside_tolerance"]}.',  # Не смешиваем непокрытые точки с превышениями.
                   f'RMSE {metrics["rmse"]:.7g}; максимум |Δ| {metrics["max_abs"]:.7g} в единицах канала.',  # Ошибка сохраняет физические единицы выбранной пары.
                   *[item['text'] for item in findings if item['text'] != summary], *notes,  # Совместные причины и ограничения сохраняются без повторения единственного краткого вывода.
                   'Диагностические гипотезы основаны на параметрических моделях. BIC и контрольная ошибка оценивают согласие модели с данными, но не доказывают отказ конкретного элемента.']  # Отделяем идентификацию динамики от причинного диагноза оборудования.
        if features:  # Наблюдаемые численные признаки доступны только после полного прохода по данным.
            details.insert(4, f'Полных переходов: эталон {features["reference_edges"]}, измерение {features["measured_edges"]}; смен знака после выключения {features["turn_off_reversals"]}; выброс после выключения {features["turn_off_overshoot"]:.7g} в единицах канала.')  # Объясняем гипотезы дребезга и снаббера фактическими числами.
        for model in model_report.get('models', []):  # Подробности показывают также конкурентов выбранного объяснения.
            details.append(f'Модель {model["code"]}: {model["formula"]}\nПараметры: {model["parameters"]}\nRMSE={model["rmse"]:.6g}; контроль={model["validation_rmse"]:.6g}; BIC={model["bic"]:.6g}; ΔBIC={model["delta_bic"]:.6g}; объяснено ошибки {model["explained_error_pct"]:.3g}%.')  # Каждый вывод сопровождается фактическими параметрами и штрафом за сложность.
        for pair in model_report.get('ringing', []):  # Отклики измерения и эталона имеют отдельные динамические оценки.
            for side in ('measured', 'reference'):  # Не выдаём качество одного сигнала за проверку другой стороны сравнения.
                model = pair.get(side)  # Эталон может не содержать пригодного колебательного участка.
                if model:  # Отсутствующая оценка не форматируется как нулевая частота или затухание.
                    details.append(f'RLC {side}: {model["parameters"]}\nR²={model["r_squared"]}; контроль RMSE={model["validation_rmse"]:.6g}; ΔBIC={model["delta_bic"]:.6g}; модель подтверждена: {model["supported"]}.')  # Приводим качество именно подбора затухающего отклика.
            details.append(f'Эквивалентный контур: {pair["equivalent"]}')  # Введённые элементы, восстановленные значения и неопределённость сохраняются в отчёте.
        joint = model_report.get('joint', {})  # Совместная электрическая проверка не должна теряться в одном только JSON.
        if joint:  # Даже отсутствие каналов является существенным ограничением модели.
            details.append(f'Совместные модели Uce/Uge/Ic: {joint}')  # Сохраняем результаты ВАХ, остаточного тока, совпадения провалов и причины неидентифицируемости.
        details.extend(model_report.get('assumptions', []))  # Предположения BIC, выборки и эквивалентной топологии видны рядом с результатами.
        if model_report.get('criteria'):  # Критерии поддержки модели не должны оставаться неявными константами кода.
            details.append(f'Начальные критерии принятия моделей: {model_report["criteria"]}. Это настройки анализа, не паспортные нормы оборудования.')  # Отделяем математическую идентификацию от выбранной чувствительности диагностического решения.
        return dict(summary=summary, details='\n\n'.join(details), attention=attention, findings=findings, features=features, model_report=model_report, version=2)  # Новый отчёт сохраняет совместимые поля и добавляет структурированные математические модели.
    def finding(code, text):  # Коды не зависят от русской формулировки и позволяют проверять сценарии.
        findings.append(dict(code=code, text=text))  # Сохраняем причину вместе с объясняющим наблюдением.
    if measured_role and reference_role and measured_role != reference_role:  # Uce и Uge нельзя диагностировать как один физический сигнал даже при одинаковых единицах.
        return finish('Выбраны разные физические каналы; сопоставьте одинаковые Uce, Uge или Ic.')  # Не выдаём ложную неисправность при ошибке выбора на экране.
    units = [re.search(r'\(([^()]*)\)\s*$', trace.name) for trace in (result.measured, result.reference)]  # Проверяем явные единицы также при независимом вызове ядра без Qt.
    if all(units) and units[0].group(1) != units[1].group(1):  # Одинаковые обозначения с разными единицами не позволяют строить аппаратные гипотезы.
        return finish('Единицы каналов различаются; сопоставьте сигналы с одинаковыми единицами.')  # Численные показатели доступны, но физическая причина не определяется.
    if metrics['coverage_pct'] < 100.:  # Любое неполное покрытие ограничивает вывод областью реально наблюдаемых данных.
        notes.append('Покрытие неполное: пропуски и время вне эталона исключены; состояние всей записи не определено.')  # Даже нулевая ошибка на остатке не подтверждает исправность всей записи.
    if metrics['count'] < 16:  # Минимум 16 общих точек нужен для распознавания формы, а не для вычисления RMSE.
        return finish('Недостаточно общих отсчётов для гипотезы неисправности.')  # Короткий набор не должен давать уверенное заключение.
    low, high = metrics['reference_min'], metrics['reference_max']  # Пределы относятся к интерполированному эталону выбранного интервала.
    span = high - low  # Размах определяет относительные пороги переключения.
    scale = max(span, abs(low), abs(high), 1e-12)  # Для постоянной ненулевой полки используем её уровень; 10^-12 исключает деление на ноль.
    switching_span = span >= .5 * scale  # Полный переход должен занимать не менее половины масштаба; малый звон около 400 В не является включением ключа.
    expected_edges = _Edges(low + .3 * span, low + .7 * span)  # Уровни 30% и 70% подавляют малые колебания на полках.
    measured_edges = _Edges(expected_edges.low, expected_edges.high)  # Одинаковые пороги позволяют сравнивать число и длительность переходов.
    minimum, maximum, step = math.inf, -math.inf, math.inf  # Границы измерения и минимальный шаг рассчитываются без дополнительного массива.
    previous, residual_sign, reversals = None, 0, 0  # Знак значимого остатка нужен для распознавания затухающего звона.
    off_sign, off_reversals, off_overshoot = 0, 0, 0.  # Снаббер проверяется по колебаниям после выключения Uce, отдельно от помех на включённой полке.
    band = max(.02 * scale, metrics['absolute_tolerance'])  # 2% масштаба и абсолютный допуск отсекают небольшую погрешность и фоновый шум.
    for index, (stamp, expected, delta) in enumerate(zip(result.difference.times, result.expected, result.difference.values)):  # Анализируем каждый сравниваемый отсчёт без экранного прореживания.
        if index % 16384 == 0 and cancel():  # Проверка отмены порциями по 16384 точки сохраняет отзывчивость больших записей.
            raise InterruptedError('Анализ отменён')  # Устаревшие гипотезы не публикуются после смены записи.
        if not math.isfinite(expected) or not math.isfinite(delta):  # Пропуск обрывает и фронт, и последовательность знаков остатка.
            expected_edges.reset(); measured_edges.reset()  # Обе стороны имеют общую область достоверного наблюдения.
            previous, residual_sign = None, 0  # Через NaN нельзя считать смену знака или длительность фронта.
            off_sign = 0  # Пропуск также прерывает подтверждение звона после выключения.
            continue  # Некорректные значения не влияют на признаки.
        if previous is not None:  # Первый корректный отсчёт ещё не определяет шаг времени.
            distance = stamp - previous  # Внутренняя единица времени — миллисекунда.
            if distance > 5. * step:  # Разрыв более пяти минимальных шагов не интерполируется диагностикой.
                expected_edges.reset(); measured_edges.reset()  # Не считаем фронт через ненаблюдавшийся промежуток времени.
                residual_sign = 0  # Звон после большого разрыва начинается новой последовательностью.
                off_sign = 0  # Колебания по разные стороны временного разрыва не образуют одну последовательность.
            step = min(step, distance)  # Минимальный наблюдаемый шаг задаёт нижнюю границу временной точности.
        value = expected + delta  # Восстанавливаем амплитуду на общей сетке без поиска по исходному массиву.
        minimum, maximum = min(minimum, value), max(maximum, value)  # Выброс сравнивается с огибающей эталона.
        if switching_span:  # Небольшие колебания отдельной полки не должны считаться полными переключениями транзистора.
            expected_edges.add(stamp, expected); measured_edges.add(stamp, value)  # Считаем переходы обеих кривых на одной временной сетке.
        sign = 1 if delta > band else -1 if delta < -band else 0  # Малая ошибка внутри полосы не образует новую полуволну звона.
        if sign:  # Нулевая зона между полуволнами сохраняет последний значимый знак.
            reversals += int(residual_sign != 0 and residual_sign != sign)  # Смена положительного и отрицательного отклонения подтверждает колебание.
            residual_sign = sign  # Следующая значимая полуволна сравнивается с текущей.
        if expected_edges.count and expected_edges.state == 1 and expected >= expected_edges.high:  # Для Uce высокая полка после наблюдаемого фронта соответствует выключенному транзистору.
            off_overshoot = max(off_overshoot, value - high, low - value)  # Выброс на включённой полке не является признаком неисправности снаббера.
            if sign:  # Для звона учитываем только значимые знакопеременные отклонения на выключенной полке.
                off_reversals += int(off_sign != 0 and off_sign != sign)  # Пересчитываем смены знака отдельно от всех помех записи.
                off_sign = sign  # Последняя полуволна сохраняется до следующего выключенного отсчёта.
        else:  # Включённая полка и область фронта прерывают проверяемую последовательность звона.
            off_sign = 0  # Шум между переключениями не связывается со следующим выключением.
        previous = stamp  # Временная непрерывность проверяется только по корректным точкам.
    if cancel():  # Проверяем отмену также после последней неполной порции.
        raise InterruptedError('Анализ отменён')  # Не возвращаем результат закрытой панели.
    features.update(reference_edges=expected_edges.count, measured_edges=measured_edges.count, residual_reversals=reversals,  # Количественные признаки объясняют гипотезы дребезга и звона.
                    overshoot=max(0., maximum - high, low - minimum), turn_off_reversals=off_reversals, turn_off_overshoot=off_overshoot, sample_step_ms=step if math.isfinite(step) else None)  # Общий выброс отделён от признаков после выключения; шаг выражен в мс.
    physical = bool(measured_role and measured_role == reference_role)  # Электрические гипотезы допустимы только для одинаковых явно обозначенных физических каналов.
    model_report, findings = identify_fault_models(result, expected_edges.events, features, physical, measured_role, circuit, channels, cancel)  # Конкуренты подбираются по данным и проверяются на отложенных точках.
    if metrics['outside_tolerance'] == 0 and not findings:  # Совпадение выбранного Uce не скрывает нарушение, подтверждённое другими синхронными каналами.
        summary = 'Отклонений от эталона в заданных допусках не обнаружено.' if metrics['absolute_tolerance'] > 0. or metrics['relative_tolerance_pct'] > 0. else 'Сигнал совпадает с эталоном; допуски не заданы.'  # Совпадение формы не является полной проверкой исправности устройства.
        return finish(summary + (' Покрытие неполное.' if notes else ''), bool(notes))  # Сохраняем ограничение неполного покрытия и параметры найденных моделей.
    if metrics['absolute_tolerance'] == 0. and metrics['relative_tolerance_pct'] == 0.:  # Нулевой коридор должен быть явно отличим от заданного норматива.
        notes.append('Допуски не заданы: любое ненулевое отклонение считается превышением; задайте погрешность измерения.')  # Обычный шум не должен восприниматься как доказанный аппаратный отказ.
    summary = ' '.join(item['text'] for item in findings)  # Разные модели совместного дефекта остаются видимыми в одной строке.
    return finish(summary + (' Покрытие неполное.' if metrics['coverage_pct'] < 100. else ''))  # Краткий вывод связан с тем же покрытием, что полный математический отчёт.