Déplier des valeurs d'encodeur ou des angles qui rebouclent à un point numérique donné
Travailler avec des grandeurs wrappées (angles, compteurs, flux modulo) est courant en traitement du signal. Ce court article présente deux exemples simples et autonomes : la fonction statique unwrap() et le OnlineUnwrapper avec état (pour usage streaming/en ligne). Les deux attendent des entrées 1D et conservent le même nombre d’échantillons.
Utiliser unwrap() — dépliage statique (batch)
Idéal lorsque vous disposez de la série complète. Astuce : définissez le threshold si vos valeurs ou votre bruit nécessitent une sensibilité de détection de wrap différente.
# Exemple simple (autonome)
import numpy as np
from UliEngineering.SignalProcessing.WrappedValues import unwrap
# Angles en degrés, wrappés à 360
wrapped = np.array([350, 355, 1, 3]) # saut 355 -> 1 (wrap)
un = unwrap(wrapped, wrap_value=360)
print(un) # -> [350. 355. 361. 363.] (angle croissant continu)Utiliser OnlineUnwrapper — dépliage en ligne/streaming
À utiliser lorsque les données arrivent un échantillon à la fois ou par morceaux. Maintient un état interne.
Note : Les deux méthodes conservent le nombre d’échantillons et sont limitées à la 1D. Utilisez unwrap() pour le traitement batch et OnlineUnwrapper pour traiter les flux temps réel ou des entrées mixtes scalaire/morceau.
# Usage en ligne avec scalaires et morceaux
import numpy as np
from UliEngineering.SignalProcessing.WrappedValues import OnlineUnwrapper
u = OnlineUnwrapper(wrap_value=100) # compteurs wrappant à 100
# Alimenter avec des scalaires
print(u(10)) # -> 10
print(u(95)) # -> 95
print(u(2)) # -> 102 (wrap vers l'avant géré en ligne)
# Ou alimenter avec un morceau (tableau 1D)
chunk = np.array([98, 3, 5]) # continue à partir de l'état précédent
print(u(chunk)) # -> array([ 98., 103., 105.])Le code utilisé pour générer le tracé ci-dessus
#!/usr/bin/env python3
# SPDX-License-Identifier: CC0-1.0
"""Exemple : tracer des signaux de phase wrappés vs dépliés.
Licence : CC0 1.0 Universal — dedication au domaine public (voir examples/LICENSE-CC0-1.0.txt)
Exécuter comme script :
python examples/unwrap_plot.py # affiche le tracé
python examples/unwrap_plot.py --save out.png # sauvegarde dans un fichier
Ceci crée un signal de phase continu synthétique, le wrap à [0,2*pi),
puis le récupère à l'aide de la fonction batch `unwrap()` et du
`OnlineUnwrapper` avec état pour démontrer qu'ils concordent.
"""
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):
"""Crée un signal de phase continu qui *inverse sa direction* au milieu.
La première moitié a une vitesse angulaire positive et la seconde moitié
une vitesse angulaire négative afin que la phase s'inverse. La série
`wrapped` renvoyée est dans [0, wrap_value).
"""
np.random.seed(seed)
t = np.linspace(0.0, 10.0, n)
mid = n // 2
# vitesses angulaires (rad par unité de temps)
omega1 = 1.2 * 2 * np.pi
omega2 = -0.8 * 2 * np.pi
true_phase = np.empty(n, dtype=float)
# première moitié : pente positive + petite oscillation
true_phase[:mid] = omega1 * t[:mid] + 0.8 * np.sin(2 * np.pi * 0.3 * t[:mid])
# seconde moitié : démarrer de la dernière valeur de la première moitié pour la continuité
# puis intégrer la vitesse angulaire négative
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:])
)
# wrap à [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)
# Dépliage batch
static_unwrapped = unwrap(wrapped, wrap_value=wrap_value)
# Dépliage en ligne (style streaming)
u = OnlineUnwrapper(wrap_value=wrap_value)
online_unwrapped = np.array([u(x) for x in wrapped])
# Vérification de cohérence rapide
maxdiff = np.max(np.abs(static_unwrapped - online_unwrapped))
print(f"différence max entre unwrap() et OnlineUnwrapper : {maxdiff:.3e}")
fig, axs = plt.subplots(2, 1, figsize=(10, 6), sharex=True)
# marquer le changement de direction dans les tracés
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="changement de direction")
axs[0].plot(t, wrapped, color="#1f77b4", lw=1)
axs[0].set_title("Signal wrappé (0 .. 2π)")
axs[0].set_ylabel("phase (rad)")
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="phase réelle")
axs[1].set_title("Signal déplié (phase continue)")
axs[1].set_xlabel("temps")
axs[1].set_ylabel("phase (rad)")
axs[1].legend()
axs[1].grid(True)
plt.tight_layout()
if save_path:
fig.savefig(save_path, dpi=150)
print(f"Figure sauvegardée dans {save_path}")
else:
plt.show()
if __name__ == "__main__":
p = argparse.ArgumentParser(description="Tracer des signaux de phase wrappés et dépliés.")
p.add_argument("--save", dest="save", help="Sauvegarder la figure dans ce chemin au lieu de l'afficher")
args = p.parse_args()
plot_example(save_path=args.save)