"""Streaming PWM demodulation; sample coordinates, no GUI dependencies.""" import math class PwmChannel: def __init__(self, samplerate, active_high=True, carrier_hz=0, smoothing_ms=0): if not math.isfinite(samplerate) or samplerate <= 0: raise ValueError('Samplerate must be positive.') if not math.isfinite(carrier_hz) or carrier_hz < 0 or carrier_hz > samplerate / 2: raise ValueError('Carrier must be 0 (automatic) or <= samplerate / 2.') if not math.isfinite(smoothing_ms) or smoothing_ms < 0: raise ValueError('Smoothing must be nonnegative.') self.rate = samplerate self.active_high = active_high self.window = samplerate / carrier_hz if carrier_hz else None self.tau = smoothing_ms * samplerate / 1000 self.last = self.level = self.start = self.fall = None self.origin = None self.window_index = 1 self.high = 0 self.previous = None self.carrier_start = self.carrier_period = None # Два действительных активных фронта нужны для измерения несущей; начальный уровень не считается фронтом. self.carrier_count = self.carrier_samples = 0 # Счётчик полных периодов и их суммарная длина в отсчётах позволяют усреднять несущую без хранения записи. self.field_previous = self.field_crossing = self.field_pending = None # Интерполированное пересечение уровня 50% отделено от подтверждённого полного периода поля. self.field_armed = False # До отрицательной полуволны начальный фрагмент огибающей не может открыть измерение поля. self.field_carrier = None # Снимок счётчиков несущей привязывает среднюю частоту к последнему завершённому периоду поля. def next_boundary(self): if self.window is None or self.origin is None: return None return round(self.origin + self.window_index * self.window) def _emit(self, start, end, high): duty = high / (end - start) value = 2 * duty - 1 midpoint = (start + end) // 2 previous = self.previous if previous is not None and self.tau: alpha = -math.expm1(-(midpoint - previous[0]) / self.tau) value = previous[1] + alpha * (value - previous[1]) self.previous = (midpoint, value) carrier = self.rate / self.carrier_period if self.carrier_period and end - self.carrier_start <= 2 * self.carrier_period else None # После двух периодов без фронтов прежняя несущая больше не считается текущим измерением. field = self._field(midpoint, value) # Частота поля определяется по той же нормированной огибающей, которая выводится на график. return dict(start=start, end=end, duty=duty, value=value, frequency=self.rate / (end - start), midpoint=midpoint, previous=previous, carrier_frequency=carrier, field=field) # Частота окна остаётся совместимой со старым контрактом, измеренная несущая и полный период поля передаются отдельно. def _field(self, sample, value): # Потоковый детектор измеряет электрическую частоту между одноимёнными пересечениями уровня заполнения 50%. previous = self.field_previous # Предыдущая точка нужна для линейной интерполяции времени пересечения между серединами периодов ШИМ. self.field_previous = (sample, value) # Храним только последнюю точку, поэтому память не зависит от длительности захвата. if value <= -0.02: # Гистерезис 0,02 нормированной шкалы соответствует одному процентному пункту заполнения ниже 50%. self.field_armed = True # Отрицательная полуволна разрешает одно следующее положительное пересечение. if self.field_armed and previous is not None and previous[1] <= 0 < value: # Учитываем только восходящий переход, чтобы не удвоить электрическую частоту. self.field_pending = previous[0] - previous[1] * (sample - previous[0]) / (value - previous[1]) # Интерполяция устраняет квантование периода поля шагом несущей. if value <= 0: # Возврат через нуль до положительного порога означает неподтверждённое пересечение. self.field_pending = None # Колебания внутри полосы гистерезиса не создают отдельные периоды. if not self.field_armed or self.field_pending is None or value < 0.02: # Положительный порог также отстоит от 50% на один процентный пункт заполнения. return None # Постоянное заполнение и неполные полуволны не получают выдуманную частоту поля. crossing = self.field_pending # Подтверждение использует время нулевого пересечения, а не более позднее достижение порога. last = self.field_crossing # Два подтверждённых восходящих пересечения ограничивают один полный электрический период. counters = (self.carrier_count, self.carrier_samples) # Снимок относится только к действительно наблюдавшимся активным фронтам. old_counters = self.field_carrier # Разность накопленных счётчиков даёт среднюю несущую за текущий период поля. self.field_crossing, self.field_carrier = crossing, counters # Следующий период начинается с текущего подтверждённого пересечения. self.field_armed, self.field_pending = False, None # Новое измерение разрешается только после следующей отрицательной полуволны. if last is None: # Первое пересечение задаёт начало, но ещё не содержит полного периода. return None # Начальный неполный фрагмент записи исключён из результата. count, elapsed = counters[0] - old_counters[0], counters[1] - old_counters[1] # Длительность измеренных периодов несущей выражена в отсчётах, а не миллисекундах. return dict(start=round(last), end=round(crossing), frequency=self.rate / (crossing - last), carrier_frequency=self.rate * count / elapsed if elapsed else None) # Полный период поля получает частоту в Гц и среднюю измеренную несущую, если за интервал были фронты. def update(self, sample, pin): level = bool(pin) == self.active_high if self.last is None: self.last, self.level = sample, level if self.window is not None: self.origin = self.start = sample return [] if sample < self.last: raise ValueError('Samples must be monotonic.') events = [] if level != self.level and level and self.window is None: # В автоматическом режиме новый активный фронт одновременно завершает текущий период ШИМ. self._carrier_edge(sample) # Обновляем измерение до выдачи завершённого периода, чтобы частота относилась именно к нему. if self.window is not None: boundary = self.next_boundary() while boundary <= sample: self.high += (boundary - self.last) * self.level events.append(self._emit(self.start, boundary, self.high)) self.start = self.last = boundary self.high = 0 self.window_index += 1 boundary = self.next_boundary() self.high += (sample - self.last) * self.level if level != self.level and level: # В режиме окон частота несущей всё равно измеряется по фронтам, независимо от заданного размера окна. self._carrier_edge(sample) # Фронт на границе окна относится к следующему окну и не меняет уже выданное измерение. elif level != self.level: if level: if self.start is not None and self.fall is not None: events.append(self._emit(self.start, sample, self.fall - self.start)) self.start, self.fall = sample, None elif self.start is not None: self.fall = sample self.last, self.level = sample, level return events def _carrier_edge(self, sample): # Оба режима восстановления используют единый счётчик фактических периодов несущей. if self.carrier_start is not None and sample > self.carrier_start: # Первый фронт только задаёт начало; нулевой период не допускается. self.carrier_period = sample - self.carrier_start # Интервал активный фронт — активный фронт задаёт полный период ШИМ в отсчётах. self.carrier_count += 1 # Счётчик нужен для среднего числа полных периодов за электрический цикл. self.carrier_samples += self.carrier_period # Суммируем длительности, чтобы переменная несущая усреднялась как число периодов за время. self.carrier_start = sample # Следующее измерение начинается с текущего реального фронта. class NpcChannel(PwmChannel): # Один вход NPC объединяет четыре затвора S1…S4, расположенные сверху вниз в фазной стойке. STATES = {(1, 1, 0, 0): 1, (0, 1, 1, 0): 0, (0, 0, 1, 1): -1} # Допустимые состояния классического NPC дают +Udc/2, нейтраль и −Udc/2 соответственно. def __init__(self, samplerate, active_high=True, carrier_hz=0, smoothing_ms=0): # Настройки времени и полярности совпадают с двухуровневым режимом. super().__init__(samplerate, active_high, carrier_hz, smoothing_ms) # Общий расчёт частоты поля и фильтра применяется к среднему знаковому уровню NPC. self.integral = self.known = 0 # Интеграл уровня выражен в отсчётах; known хранит длительность достоверных состояний. self.last_valid = self.pulse_sign = None # Пропуск dead-time не должен ошибочно создавать новый импульс той же полярности. def state(self, gates): # Преобразование использует все четыре затвора, а не только два внешних ключа. return self.STATES.get(tuple(int(bool(pin) == self.active_high) for pin in gates)) # Неизвестное состояние, включая все выключенные ключи, возвращает None вместо выдуманного нуля напряжения. def _accumulate(self, sample): # Между цифровыми событиями уровень стойки считается постоянным только при допустимой комбинации затворов. if self.level is not None: # Во время dead-time напряжение зависит от тока и диодов и не определяется одними управляющими сигналами. self.integral += (sample - self.last) * self.level # Знаковая площадь сохраняет отличие положительных, нулевых и отрицательных участков. self.known += sample - self.last # В среднем используются лишь участки, однозначно восстановленные по затворам. self.last = sample # Новый интервал начинается после полностью учтённой части времени. def _npc_emit(self, start, end): # Среднее по допустимым состояниям является оценкой огибающей, а не измерением напряжения в dead-time. coverage = self.known / (end - start) # Доля известного времени позволяет отделить короткие коммутационные паузы от повреждённой записи. if coverage < 0.95: # При более чем 5% неизвестного времени в периоде оценка считается недостаточно определённой и пропускается. self.previous = self.field_previous = self.field_crossing = self.field_pending = self.field_carrier = None # Разрыв не должен соединяться графиком или измерением поля с предыдущим достоверным участком. self.field_armed = False # Новое поле должно заново подтвердить отрицательную полуволну. return [] # Неизвестный интервал не превращается в фиктивный уровень нуля. high = (self.integral / self.known + 1) * (end - start) / 2 # Общий _emit ожидает эквивалентную активную длительность; преобразование сохраняет средний уровень −1…+1. event = self._emit(start, end, high) # Общая обработка формирует середину периода, частоты и фильтрованное значение. event['valid_fraction'] = coverage # Точность определения уровня видна в числовой аннотации и экспорте. return [event] # Формат списка совместим с обновлением двухуровневого канала. def update(self, sample, gates): # Одно событие получает атомарный снимок четырёх цифровых входов. level = self.state(gates) # Только три штатные комбинации имеют определённый нормированный уровень стойки. if self.last is None: # Начальный уровень не является обнаруженным фронтом несущей. self.last, self.level, self.last_valid = sample, level, level # Состояние начинается с текущего снимка без выдуманного предыдущего импульса. if self.window is not None: # В режиме заданной несущей интегрирование начинается с области декодирования. self.origin = self.start = sample # Границы окон совпадают с прежним контрактом PwmChannel. return [] # Начальная точка ещё не содержит полного периода. if sample < self.last: # Равные номера допустимы для одновременных фронтов разных каналов. raise ValueError('Samples must be monotonic.') # Отрицательные интервалы нельзя интегрировать или выдавать в DSView. entry = level in (-1, 1) and level != self.last_valid # Начало ненулевого импульса распознаётся с учётом последнего достоверного состояния до dead-time. events = [] # Результаты содержат лишь полные периоды или окна. if self.window is not None: # Фиксированные окна сохраняют 0/100% и работают при отсутствии переключений. boundary = self.next_boundary() # Граница рассчитывается от начала, поэтому дробный период не накапливает ошибку округления. while boundary <= sample: # Одно цифровое событие может завершить несколько окон постоянного уровня. self._accumulate(boundary) # Учёт предыдущего состояния заканчивается точно на границе окна. events.extend(self._npc_emit(self.start, boundary)) # Неизвестные участки проверяются до формирования результата. self.start = boundary # Следующее окно начинается без перекрытия с предыдущим. self.integral = self.known = 0 # Площадь и известное время относятся только к новому окну. self.window_index += 1 # Номер следующей границы задаёт её положение в общей шкале отсчётов. boundary = self.next_boundary() # Повторное вычисление сохраняет точность дробной частоты. self._accumulate(sample) # Оставшаяся часть предыдущего уровня учитывается до изменения затворов. if entry: # Частота несущей измеряется по началам последовательных ненулевых импульсов одинакового знака. if self.pulse_sign == level: # Переход положительной полуволны в отрицательную не является коротким периодом несущей. self._carrier_edge(sample) # Измеренный период не подменяется частотой заданного окна. if self.window is None and self.start is not None: # Автоматический режим завершает период на одноимённом фронте следующего импульса. events.extend(self._npc_emit(self.start, sample)) # Огибающая получается из интеграла всех трёх уровней за полный период. else: # Первое появление другой полярности заново синхронизирует несущую этой полуволны. self.carrier_start, self.carrier_period = sample, None # Интервал через смену знака не принимается за измеренную частоту. self.pulse_sign = level # Следующий ненулевой импульс сравнивается с текущей полярностью. if self.window is None: # В автоматическом режиме начало и интеграл обновляются только на опорных фронтах. self.start = sample # Начальный неполный фрагмент полуволны исключается. self.integral = self.known = 0 # Новый период не наследует площадь предыдущего. self.level = level # Новое состояние действует только после текущего цифрового события. if level is not None: # Неизвестная коммутационная пауза не стирает информацию о последнем допустимом уровне. self.last_valid = level # Это исключает ложные импульсы при кратком выключении и возврате к прежней комбинации. return events # Адаптер получает тот же поток результатов, что для обычного ШИМ. class PhaseAnalysis: # Согласование времён A/B/C отделено от конкретного числа затворов в фазе. def __init__(self, samplerate): # Частота нужна для контроля устаревших оценок фазового периода. self.rate = samplerate # Все временные координаты остаются в отсчётах захвата. self.points, self.fields = {}, {} # Для интерполяции хранятся две последние точки каждой огибающей и последний полный период поля. self.last_vector = None # Одинаковый общий момент времени не должен выдавать повторный угол. def add(self, channel, event): # Каждый восстановленный период обновляет только свою фазу. points = self.points.setdefault(channel, []) # Отсутствующая фаза не получает нулевой или выдуманный сигнал. if event.get('previous', True) is None: # Ядро явно разрывает огибающую после недостаточно определённого окна NPC. points.clear() # Интерполяция не должна соединять новую точку с данными до неизвестного участка. self.fields.pop(channel, None) # Межфазные углы также должны заново дождаться полного достоверного периода. points.append((event['midpoint'], event['value'])) # Угол вычисляется по восстановленным средним уровням, а не по мгновенным коммутациям. del points[:-2] # Ограниченная история сохраняет постоянный расход памяти на длинных записях. if event['field'] is not None: # Межфазные углы требуют наблюдавшихся полных электрических периодов. field = event['field'] # Результат содержит последнее восходящее пересечение и частоту данной фазы. self.fields[channel] = (field['end'], self.rate / field['frequency']) # Период хранится в отсчётах для вычисления сдвига в градусах. if any(len(self.points.get(ch, [])) < 2 for ch in range(3)): # Пространственный вектор нельзя восстановить из неподключённых или ещё не заполненных A/B/C. return None # Начальный участок записи не получает ложного θ. sample = min(self.points[ch][-1][0] for ch in range(3)) # Общая точка не выходит за наблюдавшееся время ни одной фазы. if sample <= max(self.points[ch][0][0] for ch in range(3)) or self.last_vector is not None and sample <= self.last_vector: # Требуется общий интервал и новое время, без экстраполяции остановившейся фазы. return None # Несогласованные точки не смешиваются в один вектор. values = [] # Три значения будут относиться к одному и тому же моменту времени. for ch in range(3): # Порядок A/B/C определяет положительное направление вращения. left, right = self.points[ch] # Две соседние точки ограничивают достоверный интервал данной фазы. values.append(left[1] + (right[1] - left[1]) * (sample - left[0]) / (right[0] - left[0])) # Линейная интерполяция устраняет сдвиг, вызванный разными моментами фронтов PWM A/B/C. alpha = (2 * values[0] - values[1] - values[2]) / 3 # Полное преобразование Кларка удаляет синфазную составляющую, включая добавку SVPWM. beta = (values[1] - values[2]) / math.sqrt(3) # Ортогональная ось соответствует последовательности A, B, C. self.last_vector = sample # Данный момент времени считается обработанным даже при почти нулевом векторе. if math.hypot(alpha, beta) < 0.02: # Ниже 2% нормированного уровня направление вектора не считается определённым. return dict(sample=sample, theta=None) # Разрыв явно запрещает соединять графиком углы через нулевую амплитуду. return dict(sample=sample, theta=math.degrees(math.atan2(beta, alpha)) % 360) # θ выражен в электрических градусах 0…360 относительно положительной оси A. def differences(self, sample): # Положительный результат A→B означает запаздывание фазы B относительно A. angles = {} # Отсутствующая пара не получает выдуманный сдвиг 120 градусов. for first, second in ((0, 1), (1, 2), (2, 0)): # Измеряются все три запрошенные пары независимо. if first not in self.fields or second not in self.fields: # Каждой фазе пары нужен хотя бы один полный измеренный период. continue # Две подключённые фазы всё ещё могут дать одну доступную пару. a, b = self.fields[first], self.fields[second] # Последние одноимённые пересечения задают фазовые часы двух огибающих. if any(sample < item[0] or sample - item[0] > 2 * item[1] for item in (a, b)): # После двух периодов без нового измерения прежний сдвиг становится устаревшим. continue # Остановившийся или пропавший вход не сохраняет старое числовое значение. angle = 360 * ((sample - a[0]) / a[1] - (sample - b[0]) / b[1]) # Разность фазовых часов выражает запаздывание второй огибающей в градусах. angles['ABC'[first] + '→' + 'ABC'[second]] = (angle + 180) % 360 - 180 # Знаковый диапазон −180…180 различает запаздывание и опережение, включая обратную последовательность фаз. return angles # Результаты рассчитываются из записи и не предполагают заранее правильную трёхфазную систему.