Как развернуть значения энкодера или углы, которые переносятся в определённой числовой точке

Работа с обёрнутыми величинами (углами, счётчиками, потоками по модулю) — распространённая задача в обработке сигналов. В этом коротком посте показаны два простых самодостаточных примера: статическая функция unwrap() и объект OnlineUnwrapper с сохранением состояния (для потоковой/онлайн-обработки). Оба варианта принимают на вход одномерные данные и сохраняют то же количество отсчётов.

Unwrap

Использование unwrap() — статическое (пакетное) развёртывание

Лучше всего подходит, когда у вас есть весь ряд данных целиком. Совет: задайте порог, если ваши значения или шум требуют другой чувствительности обнаружения переноса.

unwrap_example.py
# Простой пример (самодостаточный)
import numpy as np
from UliEngineering.SignalProcessing.WrappedValues import unwrap

# Углы в градусах, обёрнутые на 360
wrapped = np.array([350, 355,   1,   3])   # скачок 355 -> 1 (перенос)
un = unwrap(wrapped, wrap_value=360)

print(un)  # -> [350. 355. 361. 363.]  (непрерывно возрастающий угол)

Использование OnlineUnwrapper — онлайн/потоковое развёртывание

Используйте, когда данные поступают по одному отсчёту за раз или блоками. Поддерживает внутреннее состояние.

Примечание: Оба метода сохраняют количество отсчётов и работают только с одномерными данными. Используйте unwrap() для пакетной обработки, а OnlineUnwrapper — для обработки потоков в реальном времени или смешанного ввода скаляров/блоков.

online_unwrapper_example.py
# Онлайн-использование со скалярами и блоками
import numpy as np
from UliEngineering.SignalProcessing.WrappedValues import OnlineUnwrapper

u = OnlineUnwrapper(wrap_value=100)   # счётчики с переносом на 100

# Передаём скаляры
print(u(10))   # -> 10
print(u(95))   # -> 95
print(u(2))    # -> 102  (перенос вперёд обрабатывается онлайн)

# Или передаём блок (одномерный массив)
chunk = np.array([98, 3, 5])  # продолжается с предыдущего состояния
print(u(chunk))  # -> array([ 98., 103., 105.])

Код, использованный для построения графика выше

Unwrap

plot_wrapped_unwrapped.py
#!/usr/bin/env python3
# SPDX-License-Identifier: CC0-1.0
"""Пример: построение графиков обёрнутых и развёрнутых фазовых сигналов.

Лицензия: CC0 1.0 Universal — передача в общественное достояние (см. examples/LICENSE-CC0-1.0.txt)

Запуск как скрипт:
    python examples/unwrap_plot.py        # показывает график
    python examples/unwrap_plot.py --save out.png  # сохраняет в файл

Создаётся синтетический непрерывный фазовый сигнал, он обёртывается в [0,2*pi),
затем восстанавливается с помощью пакетной функции `unwrap()` и объекта
`OnlineUnwrapper` с сохранением состояния, чтобы показать их согласованность.
"""
import argparse

import numpy as np
import matplotlib.pyplot as plt
plt.style.use("ggplot")

from UliEngineering.SignalProcessing.WrappedValues import unwrap, OnlineUnwrapper

def make_data(n=1000, wrap_value=2 * np.pi, seed=0):
    """Создаёт непрерывный фазовый сигнал, который *меняет направление* в середине.

    Первая половина имеет положительную угловую скорость, а вторая —
    отрицательную угловую скорость, так что фаза меняет направление.
    Возвращаемый ряд `wrapped` находится в диапазоне [0, wrap_value).
    """
    np.random.seed(seed)
    t = np.linspace(0.0, 10.0, n)
    mid = n // 2
    # угловые скорости (рад на единицу времени)
    omega1 = 1.2 * 2 * np.pi
    omega2 = -0.8 * 2 * np.pi

    true_phase = np.empty(n, dtype=float)
    # первая половина: положительный наклон + небольшое колебание
    true_phase[:mid] = omega1 * t[:mid] + 0.8 * np.sin(2 * np.pi * 0.3 * t[:mid])
    # вторая половина: начинаем от последнего значения первой половины для непрерывности
    # и затем интегрируем отрицательную угловую скорость
    start_phase = true_phase[mid - 1]
    true_phase[mid:] = (
        start_phase
        + omega2 * (t[mid:] - t[mid - 1])
        + 0.8 * np.sin(2 * np.pi * 0.3 * t[mid:])
    )

    # обёртывание в [0, wrap_value)
    wrapped = np.mod(true_phase, wrap_value)
    return t, true_phase, wrapped


def plot_example(save_path=None):
    wrap_value = 2 * np.pi
    t, true_phase, wrapped = make_data(n=1200, wrap_value=wrap_value)

    # Пакетное развёртывание
    static_unwrapped = unwrap(wrapped, wrap_value=wrap_value)

    # Онлайн-развёртывание (в стиле потока)
    u = OnlineUnwrapper(wrap_value=wrap_value)
    online_unwrapped = np.array([u(x) for x in wrapped])

    # Быстрая проверка
    maxdiff = np.max(np.abs(static_unwrapped - online_unwrapped))
    print(f"максимальная разница между unwrap() и OnlineUnwrapper: {maxdiff:.3e}")

    fig, axs = plt.subplots(2, 1, figsize=(10, 6), sharex=True)

    # отмечаем изменение направления на графиках
    mid_idx = len(t) // 2
    axs[0].axvline(t[mid_idx], color="#444444", ls=":", lw=0.8)
    axs[1].axvline(t[mid_idx], color="#444444", ls=":", lw=0.8, label="изменение направления")

    axs[0].plot(t, wrapped, color="#1f77b4", lw=1)
    axs[0].set_title("Обёрнутый сигнал (0 .. 2π)")
    axs[0].set_ylabel("фаза (рад)")
    axs[0].grid(True)

    axs[1].plot(t, static_unwrapped, color="#2ca02c", lw=1, label="unwrap()")
    axs[1].plot(t, online_unwrapped, color="#ff7f0e", lw=1, ls="--", label="OnlineUnwrapper")
    axs[1].plot(t, true_phase, color="#7f7f7f", lw=0.8, ls=":", label="истинная фаза")
    axs[1].set_title("Развёрнутый сигнал (непрерывная фаза)")
    axs[1].set_xlabel("время")
    axs[1].set_ylabel("фаза (рад)")
    axs[1].legend()
    axs[1].grid(True)

    plt.tight_layout()
    if save_path:
        fig.savefig(save_path, dpi=150)
        print(f"График сохранён в {save_path}")
    else:
        plt.show()


if __name__ == "__main__":
    p = argparse.ArgumentParser(description="Построение графиков обёрнутых и развёрнутых фазовых сигналов.")
    p.add_argument("--save", dest="save", help="Сохранить график по этому пути вместо показа на экране")
    args = p.parse_args()
    plot_example(save_path=args.save)

Check out similar posts by category: UliEngineering, Python