""" Устойчивый контроллер режима радиоканала SDR Rover Link. Контроллер устраняет частое переключение режимов из-за шума измерения Eb/N0. Реализованы: 1. Экспоненциальное усреднение Eb/N0. 2. Гистерезис порогов. 3. Подтверждение повышения режима. 4. Минимальное время удержания режима. 5. Быстрое аварийное понижение режима. Доступные режимы: 0 — изображения отключены; 128 — фрагменты по 128 байт; 512 — фрагменты по 512 байт; 1024 — фрагменты по 1024 байта. """ from dataclasses import dataclass # ============================================================ # Допустимые режимы # ============================================================ IMAGE_OFF = 0 FRAGMENT_MODE_128 = 128 FRAGMENT_MODE_512 = 512 FRAGMENT_MODE_1024 = 1024 VALID_MODES = ( IMAGE_OFF, FRAGMENT_MODE_128, FRAGMENT_MODE_512, FRAGMENT_MODE_1024, ) # ============================================================ # Результат одного шага контроллера # ============================================================ @dataclass(frozen=True) class LinkControllerResult: """ Состояние контроллера после нового измерения Eb/N0. """ step: int measured_eb_n0_db: float filtered_eb_n0_db: float previous_mode: int selected_mode: int mode_changed: bool pending_upgrade_mode: int | None pending_upgrade_count: int reason: str # ============================================================ # Контроллер # ============================================================ class LinkModeController: """ Устойчивый адаптивный контроллер режима изображения. Пороговые значения выбраны по результатам Lab015 и Lab016. Гистерезис означает, что повышение и понижение режима происходят при разных значениях Eb/N0. Пример для режима 512 байт: повышение 128 → 512: Eb/N0 >= 8,7 дБ понижение 512 → 128: Eb/N0 < 8,1 дБ Благодаря разнице 0,6 дБ небольшие колебания около порога не вызывают постоянного переключения. """ def __init__( self, initial_mode: int = IMAGE_OFF, filter_alpha: float = 0.35, upgrade_confirmation_count: int = 3, minimum_hold_steps: int = 2, ) -> None: """ Создать контроллер. Параметры ---------- initial_mode: Начальный режим. filter_alpha: Коэффициент экспоненциального фильтра. Большое значение: быстрее реакция; слабее фильтрация шума. Малое значение: медленнее реакция; сильнее фильтрация. upgrade_confirmation_count: Сколько последовательных измерений требуется для повышения режима. minimum_hold_steps: Минимальное число шагов удержания режима перед очередным повышением. """ if initial_mode not in VALID_MODES: raise ValueError( f"Недопустимый начальный режим: {initial_mode}" ) if not 0.0 < filter_alpha <= 1.0: raise ValueError( "filter_alpha должен находиться в диапазоне 0...1" ) if upgrade_confirmation_count <= 0: raise ValueError( "upgrade_confirmation_count должен быть положительным" ) if minimum_hold_steps < 0: raise ValueError( "minimum_hold_steps не может быть отрицательным" ) self.current_mode = initial_mode self.filter_alpha = filter_alpha self.upgrade_confirmation_count = ( upgrade_confirmation_count ) self.minimum_hold_steps = ( minimum_hold_steps ) self.filtered_eb_n0_db: float | None = None self.step = 0 self.steps_in_mode = 0 self.pending_upgrade_mode: int | None = None self.pending_upgrade_count = 0 # ======================================================== # Внутренние функции # ======================================================== def _update_filter( self, measured_eb_n0_db: float, ) -> float: """ Обновить экспоненциальное среднее. Формула: filtered = alpha * measured + (1 - alpha) * previous_filtered """ if self.filtered_eb_n0_db is None: self.filtered_eb_n0_db = measured_eb_n0_db else: self.filtered_eb_n0_db = ( self.filter_alpha * measured_eb_n0_db + ( 1.0 - self.filter_alpha ) * self.filtered_eb_n0_db ) return self.filtered_eb_n0_db def _emergency_safe_mode( self, measured_eb_n0_db: float, ) -> int: """ Определить максимально безопасный режим по мгновенному измерению. Используется для быстрого понижения режима, не дожидаясь усреднения. """ if measured_eb_n0_db < 5.8: return IMAGE_OFF if measured_eb_n0_db < 7.7: return FRAGMENT_MODE_128 if measured_eb_n0_db < 8.7: return FRAGMENT_MODE_512 return FRAGMENT_MODE_1024 def _desired_mode_from_filtered( self, filtered_eb_n0_db: float, ) -> int: """ Определить желаемый режим с учётом гистерезиса. Повышение выполняется только на один уровень за шаг. """ if self.current_mode == IMAGE_OFF: if filtered_eb_n0_db >= 6.8: return FRAGMENT_MODE_128 return IMAGE_OFF if self.current_mode == FRAGMENT_MODE_128: if filtered_eb_n0_db < 6.2: return IMAGE_OFF if filtered_eb_n0_db >= 8.7: return FRAGMENT_MODE_512 return FRAGMENT_MODE_128 if self.current_mode == FRAGMENT_MODE_512: if filtered_eb_n0_db < 8.1: return FRAGMENT_MODE_128 if filtered_eb_n0_db >= 9.8: return FRAGMENT_MODE_1024 return FRAGMENT_MODE_512 if self.current_mode == FRAGMENT_MODE_1024: if filtered_eb_n0_db < 9.2: return FRAGMENT_MODE_512 return FRAGMENT_MODE_1024 raise RuntimeError( f"Неизвестный текущий режим: {self.current_mode}" ) def _reset_pending_upgrade( self, ) -> None: """ Сбросить накопленное подтверждение повышения. """ self.pending_upgrade_mode = None self.pending_upgrade_count = 0 def _change_mode( self, new_mode: int, ) -> None: """ Немедленно установить новый режим. """ if new_mode not in VALID_MODES: raise ValueError( f"Недопустимый режим: {new_mode}" ) self.current_mode = new_mode self.steps_in_mode = 0 self._reset_pending_upgrade() # ======================================================== # Основной шаг # ======================================================== def update( self, measured_eb_n0_db: float, ) -> LinkControllerResult: """ Обработать новое измерение Eb/N0. Логика: 1. Обновить фильтр. 2. Проверить аварийное понижение. 3. Проверить обычное понижение по фильтру. 4. Для повышения потребовать несколько подтверждений. 5. Вернуть результат. """ measured_eb_n0_db = float( measured_eb_n0_db ) previous_mode = self.current_mode filtered_eb_n0_db = self._update_filter( measured_eb_n0_db ) reason = "Режим удерживается" # ---------------------------------------------------- # Аварийное быстрое понижение # ---------------------------------------------------- emergency_mode = self._emergency_safe_mode( measured_eb_n0_db ) if emergency_mode < self.current_mode: self._change_mode( emergency_mode ) reason = ( "Аварийное понижение по мгновенному Eb/N0" ) else: desired_mode = self._desired_mode_from_filtered( filtered_eb_n0_db ) # ------------------------------------------------ # Обычное понижение # ------------------------------------------------ if desired_mode < self.current_mode: self._change_mode( desired_mode ) reason = ( "Понижение по фильтрованному Eb/N0" ) # ------------------------------------------------ # Возможное повышение # ------------------------------------------------ elif desired_mode > self.current_mode: if ( self.steps_in_mode < self.minimum_hold_steps ): self._reset_pending_upgrade() reason = ( "Повышение запрещено: " "не выдержано минимальное время режима" ) else: if ( self.pending_upgrade_mode == desired_mode ): self.pending_upgrade_count += 1 else: self.pending_upgrade_mode = ( desired_mode ) self.pending_upgrade_count = 1 reason = ( "Накопление подтверждений " "для повышения режима" ) if ( self.pending_upgrade_count >= self.upgrade_confirmation_count ): self._change_mode( desired_mode ) reason = ( "Режим повышен после " "нескольких подтверждений" ) else: self._reset_pending_upgrade() mode_changed = ( self.current_mode != previous_mode ) result = LinkControllerResult( step=self.step, measured_eb_n0_db=measured_eb_n0_db, filtered_eb_n0_db=filtered_eb_n0_db, previous_mode=previous_mode, selected_mode=self.current_mode, mode_changed=mode_changed, pending_upgrade_mode=( self.pending_upgrade_mode ), pending_upgrade_count=( self.pending_upgrade_count ), reason=reason, ) self.step += 1 self.steps_in_mode += 1 return result