SDR-Rover/experiments/lab017_link_hysteresis.py

"""
Lab017. Фильтрация Eb/N0 и гистерезис режимов.

Эксперимент сравнивает два адаптера:

1. Наивный адаптер:
   переключает режим при каждом пересечении порога.

2. Устойчивый адаптер:
   использует:
   - экспоненциальное усреднение;
   - гистерезис;
   - подтверждение повышения;
   - аварийное быстрое понижение.

Профиль Eb/N0 специально содержит колебания
около порогов переключения.
"""

from csv import DictWriter
from pathlib import Path

import matplotlib.pyplot as plt

from protocol.link_controller import (
    FRAGMENT_MODE_1024,
    IMAGE_OFF,
    LinkModeController,
)


# ============================================================
# Настройки
# ============================================================

OUTPUT_DIRECTORY = Path(
    "data/processed/lab017"
)

OUTPUT_DIRECTORY.mkdir(
    parents=True,
    exist_ok=True,
)

CSV_PATH = (
    OUTPUT_DIRECTORY
    / "lab017_hysteresis_results.csv"
)

GRAPH_PATH = (
    OUTPUT_DIRECTORY
    / "lab017_hysteresis_comparison.png"
)


# ============================================================
# Профиль качества канала
# ============================================================

EB_N0_PROFILE_DB = [
    10.2,
    9.7,
    9.4,
    9.6,
    9.3,
    9.7,
    9.2,
    9.6,
    9.0,
    8.7,
    8.4,
    8.6,
    8.3,
    8.7,
    8.2,
    8.6,
    8.0,
    7.8,
    7.5,
    7.7,
    7.4,
    7.8,
    7.2,
    6.8,
    6.5,
    6.3,
    6.6,
    6.2,
    6.7,
    6.1,
    5.5,
    6.0,
    6.4,
    6.7,
    6.9,
    7.1,
    7.3,
    8.0,
    8.8,
    9.0,
    9.4,
    9.8,
    10.0,
    10.2,
    10.1,
]


# ============================================================
# Наивное решение без гистерезиса
# ============================================================

def choose_naive_mode(
    eb_n0_db: float,
) -> int:
    """
    Немедленно выбрать режим по одному измерению.

    Гистерезиса и фильтрации нет.
    """

    if eb_n0_db < 6.5:
        return 0

    if eb_n0_db < 8.5:
        return 128

    if eb_n0_db < 9.5:
        return 512

    return 1024


# ============================================================
# Запуск устойчивого контроллера
# ============================================================

controller = LinkModeController(
    initial_mode=FRAGMENT_MODE_1024,
    filter_alpha=0.35,
    upgrade_confirmation_count=3,
    minimum_hold_steps=2,
)

results = []

for measured_eb_n0_db in EB_N0_PROFILE_DB:

    stable_result = controller.update(
        measured_eb_n0_db
    )

    naive_mode = choose_naive_mode(
        measured_eb_n0_db
    )

    results.append(
        {
            "step": stable_result.step,
            "measured_eb_n0_db": (
                stable_result.measured_eb_n0_db
            ),
            "filtered_eb_n0_db": (
                stable_result.filtered_eb_n0_db
            ),
            "naive_mode": naive_mode,
            "stable_mode": (
                stable_result.selected_mode
            ),
            "mode_changed": (
                stable_result.mode_changed
            ),
            "pending_upgrade_mode": (
                stable_result.pending_upgrade_mode
            ),
            "pending_upgrade_count": (
                stable_result.pending_upgrade_count
            ),
            "reason": stable_result.reason,
        }
    )


# ============================================================
# Число переключений
# ============================================================

def count_mode_switches(
    modes: list[int],
) -> int:
    """
    Посчитать число изменений режима.
    """

    return sum(
        previous_mode != current_mode
        for previous_mode, current_mode in zip(
            modes,
            modes[1:],
        )
    )


naive_modes = [
    result["naive_mode"]
    for result in results
]

stable_modes = [
    result["stable_mode"]
    for result in results
]

naive_switch_count = count_mode_switches(
    naive_modes
)

stable_switch_count = count_mode_switches(
    stable_modes
)


# ============================================================
# Вывод результатов
# ============================================================

print(
    "=== Lab017. Гистерезис режима канала ==="
)

print("\nРезультаты по шагам:")

print(
    f"{'Шаг':>5}"
    f"{'Eb/N0':>10}"
    f"{'Фильтр':>11}"
    f"{'Наивный':>11}"
    f"{'Устойчивый':>13}"
    f"{'Изменён':>10}"
    f"{'Причина':>45}"
)

print("-" * 105)

for result in results:

    print(
        f"{result['step']:>5}"
        f"{result['measured_eb_n0_db']:>7.1f} дБ"
        f"{result['filtered_eb_n0_db']:>8.2f} дБ"
        f"{result['naive_mode']:>11}"
        f"{result['stable_mode']:>13}"
        f"{str(result['mode_changed']):>10}"
        f"  {result['reason']}"
    )


print("\nЧисло переключений:")

print(
    "Наивный адаптер:",
    naive_switch_count,
)

print(
    "Устойчивый адаптер:",
    stable_switch_count,
)

print(
    "Предотвращено лишних переключений:",
    naive_switch_count - stable_switch_count,
)


# ============================================================
# Сохранение CSV
# ============================================================

with CSV_PATH.open(
    "w",
    newline="",
    encoding="utf-8-sig",
) as csv_file:

    writer = DictWriter(
        csv_file,
        fieldnames=list(
            results[0].keys()
        ),
    )

    writer.writeheader()
    writer.writerows(results)


# ============================================================
# График
# ============================================================

steps = [
    result["step"]
    for result in results
]

measured_values = [
    result["measured_eb_n0_db"]
    for result in results
]

filtered_values = [
    result["filtered_eb_n0_db"]
    for result in results
]


figure, axes = plt.subplots(
    2,
    1,
    figsize=(12, 9),
    sharex=True,
)


# ------------------------------------------------------------
# Верхний график: Eb/N0
# ------------------------------------------------------------

axes[0].plot(
    steps,
    measured_values,
    marker=".",
    label="Измеренное Eb/N0",
)

axes[0].plot(
    steps,
    filtered_values,
    linewidth=2,
    label="Фильтрованное Eb/N0",
)

axes[0].axhline(
    6.5,
    linestyle=":",
    label="Порог наивного адаптера",
)

axes[0].axhline(
    8.5,
    linestyle=":",
)

axes[0].axhline(
    9.5,
    linestyle=":",
)

axes[0].set_ylabel(
    "Eb/N0, дБ"
)

axes[0].set_title(
    "Фильтрация измерения качества канала"
)

axes[0].grid(
    True
)

axes[0].legend()


# ------------------------------------------------------------
# Нижний график: режимы
# ------------------------------------------------------------

axes[1].step(
    steps,
    naive_modes,
    where="mid",
    marker=".",
    label="Наивный адаптер",
)

axes[1].step(
    steps,
    stable_modes,
    where="mid",
    marker="o",
    label="Гистерезис и фильтрация",
)

axes[1].set_yticks(
    [
        0,
        128,
        512,
        1024,
    ],
    [
        "IMAGE OFF",
        "128 B",
        "512 B",
        "1024 B",
    ],
)

axes[1].set_xlabel(
    "Шаг времени"
)

axes[1].set_ylabel(
    "Режим"
)

axes[1].set_title(
    "Сравнение переключений режимов"
)

axes[1].grid(
    True
)

axes[1].legend()


figure.tight_layout()

figure.savefig(
    GRAPH_PATH,
    dpi=160,
)

plt.close(
    figure
)


# ============================================================
# Проверки
# ============================================================

assert results

assert IMAGE_OFF in stable_modes

assert FRAGMENT_MODE_1024 in stable_modes

assert stable_switch_count < naive_switch_count

assert CSV_PATH.exists()

assert GRAPH_PATH.exists()


print("\nCSV:")

print(CSV_PATH)

print("\nГрафик:")

print(GRAPH_PATH)

print(
    "\nПроверка пройдена: "
    "гистерезис уменьшил количество переключений."
)