Création d'une station météo
#Station Météo (Partie Réception)
Le but de ce projet était de créer une station météo permettant de récupérer une trame émise, la décoder puis afficher la température qui était contenu.
Matériel utilisé :
Pour ce projet, j'ai utilisé une ADALM-PLUTO ainsi que le langage Python pour le programmer.
L'ADALM-PLUTO est une SDR (Software Defined Radio) ayant une grande gamme porteuse (de 325MHz à 3.8GHz) avec une résolution de 12 bits pour le convertisseur analogique-numérique. Cela signifie qu'il y a 212 soit 4096 valeurs possibles pour un échantillon.
Il permet aussi de faire de l'émission. L'avantage étant qu'il fonctionne en full-duplex, permettant l'émission et la réception en même temps.
Structure de la trame :
On a une trame de 42 bits découpée de la manière suivante :
- Synchro (2 bits) : ces 2 bits à 0 suivent 2 états de synchro et servent à annoncer le début d'une trame
- ID (8 bits) : c'est l'identifiants de l'émetteur, comme nous fabriquons une station météo, plusieurs capteurs de températures cohexistent. Il faut donc que le récepteur identifie le capteur avec lequel il a été appairé (intérieur/extérieur par exemple). Cette valeur n'a cependant pas beaucoup d'importance dans notre cas.
- Canal d'émission (4 bits) : l'émetteur transmet sur 7 canaux différents, ce demi-octet permet de savoir sur quel canal la température est reçu
- DATA (20 bits) : la partie data contient la température sur ces 12 premiers bits. Dans ces transmissions, les 8 bits suivant sont toujours à 0.
- CRC (8 bits) : le CRC permet de vérifier que la transmission s'est bien déroulé et que les données reçus sont bien celles qui ont été envoyés.
Paramétrage de l'ADALM PLUTO
# Importation des bibliothèques
import adi # communication avec Adalm Pluto (AP)
import numpy as np # gestion des tableaux en Python
import matplotlib.pyplot as plt # affichage de graphiques
sdr = adi.Pluto("ip:192.168.3.1") # adresse relevée dans le fichier config
print(sdr) # affichage des paramètres par défaut de réception (rx) et d'émission (tx) de l'A-P
# Choix des paramètres
Freq_osc_local = 433e6 # la fréquence porteuse théorique du signal à observer
Freq_PasseBas = 5e6 # la largeur du filtre de réception
Freq_ech = 10e6 # la fréquence d'échantillonnage du signal reçu (conséquence sur la taille mémoire utilisée)
print(" ")
Duree_Ech = 0.3 # Choix de la durée de temps pendant lequel on veut échantillonner, théoriquement en seconde
Tech = 1/Freq_ech # période d'échantillonnage
print("La période d'échantillonnage est de : ", round(Tech*1000000,3) , "µs")
Taille_Buffer = Duree_Ech * Freq_ech # Calcul de la taille mémoire nécessaire à cet échantillonnage
print("La taille de l'échantillon capté sera de : " , Taille_Buffer/1000, "k Octets")
print(" ")
# Affectation des paramètres à l'A-P
sdr.rx_lo = int(Freq_osc_local)
sdr.rx_rf_bandwidth = int(Freq_PasseBas)
sdr.sample_rate = int(Freq_ech)
sdr.rx_destroy_buffer() # (voir https://pysdr.org/fr/content-fr/pluto.html)
sdr.rx_buffer_size = int(Taille_Buffer)
sdr.gain_control_mode_chan0 = "slow_attack" #gain automatique
Ce code permet de paramétrer l'ADALM PLUTO à la réception que nous allons faire. Je met la fréquence de l'oscillateur local à 433MHz car c'est une fréquence réservée pour les radioamateurs. Cette fréquence correspond à la porteuse du signal que l'on veut recevoir.
Lors de la réception, l'ADALM-PLUTO va transposer la réception à une basse fréquence grâce à celle de l'oscillateur local. Je met donc la valeur du filtre passe-bas à 5MHz, ainsi je vais garder uniquement ce qui se trouve entre 0 et 5MHz lors de la réception. Cela permet de ne garder que la partie utile du signal.
Enfin, on met la fréquence d’échantillonnage à 10MHz afin d'avoir BEAUCOUP d'échantillons à traiter, ce qui nous permettra d'avoir une plus grande précision par la suite.
Acquisitions
Dans un premier temps, je vais faire une série d'acquisitions pour voir le temps entre les transmissions et la puissance de ces dernières.
import time
# la précision du temps d'exécution de la boucle :
Duree_Obs = 20 # choix en secondes
nbr_acq = Duree_Obs/Duree_Ech # combien faire d'itération à la boucle
print("ATTENDRE que le processus soit terminé.")
# La boucle et ses initialisations
debut=time.time()
PuissancesdBm = []
PrdBm = 0
for i in range (int(nbr_acq)) :
Donnees = sdr.rx() #acquisitions des données
Signal_Recu = abs(Donnees/(2**12)) #récupération de la valeur absolu
PrW = np.average((Signal_Recu**2)/1) #on met le signal sur une seule valeur
PrdBm = 10*np.log10(PrW/0.001) #on le passe en dBm
PuissancesdBm.append(PrdBm) #on l'ajoute au tableau des valeurs
print(".", end='') # met 1 point(.) par itération afin de voir la progression. "end=''" permet de mettre en ligne
fin = time.time()
Duree_Calc_reel = (fin - debut)/len(PuissancesdBm)
print("C'est fini !")
En fonction de la durée de l'observation que nous avons choisit, on va avoir un nombre d'acquisitions à faire.
Ce nombre d'acquisitions va être le nombre d'itérations de la boucle. Cette boucle fait une acquisition, traite le signal reçu (récupération de la valeur absolu et calcul de la moyenne sur l'échantillon) puis l'ajoute dans le tableau en dBm.
Le calcul de Duree_Calc_reel correspond au temps d'un échantillon et va être utile pour l'affichage des résultats.
Affichage des acquisitions
Une fois cela fait, on ajoute le code suivant pour faire un affichage des acquisitions :
ech_temps = np.linspace(0 , len(PuissancesdBm)*Duree_Calc_reel , len(PuissancesdBm)) #création d'une échelle de temps
deffig = plt.figure(figsize=(10,4)) # dimensionne le tracé
plt.plot( ech_temps, PuissancesdBm , color='red')
plt.title("Puissances observées sur une longue période")
plt.xlabel("temps en seconde ")
plt.ylabel("module en dBmW")
L'échelle de temps se faire de la manière suivante :
- on commence l'échelle à 0
- on la termine à la valeur correspondant au nombre d'échantillons multiplié par la durée d'un échantillon
- on a autant de point qu'il y a d'échantillons
Ce code nous donne donc le tracé suivant : 
Ce tracé montre qu'il y a 2 émissions rapproché avec 5 secondes d'intervalle. Ces émissions sont ensuite séparés par un temps mort de 15 secondes avant de répéter les émissions.
Exploitation des acquisitions
On peut exploiter ces différentes acquisitions pour avoir des valeurs utiles pour la suite. On va récupérer la puissance maximum que l'on a reçu et la puissance moyenne. Avec ces deux valeurs, on va calculer un seuil de détection qui nous servira à savoir si ce que l'on reçoit est du bruit ou une transmission.
Pmax = np.max(PuissancesdBm)
print("La puissance maximale observée est de :" , round(Pmax, 3) , "dBm")
Pmoyenne = np.average(PuissancesdBm)
print("La puissance moyenne (proche du bruit) est de :" , round(Pmoyenne, 3) , "dBm")
Pcomp = (Pmax + Pmoyenne)/2
print("Le niveau de détection peut etre choisi à :" , round(Pcomp, 3) , "dBm")
Avec ce que l'on a reçu, on obtient les valeurs suivantes :
La puissance maximale observée est de : 13.496 dBm La puissance moyenne (proche du bruit) est de : -0.128 dBm Le niveau de détection peut être choisi à : 6.684 dBm
Acquisition d'une trame de température
Maintenant que nous avons ces informations, nous pouvons faire en sorte de récupérer une unique trame contenant une température.
Pour cela, on reprend le programme précédent en le modifiant. On va tout d'abord ajouter différentes variables.
Tout d'abord on va utiliser les puissances qu'on vient de calculer :
niveau_comp = Pcomp # niveau intermédiaire entre le bruit et le signal de la figure précédente
PrdBm = Pmoyenne - 50 # initialisation avec une valeur que l'on est certain de ne jamais atteindre !
Elles permettront de savoir si l'acquisition de l'ADALM-PLUTO est une trame ou bien du bruit.
On va ensuite initialiser les tableaux et un compteur :
PuissancesdBm = [] #tableau avec uniquement les puissances
Sig_normalisé = [] #tableau avec le signal complexe normalisé
compteur = 0
print("ATTENDRE que le processus soit terminé.")
Ce compteur permettra le calcul de la durée d'une acquisitions et par conséquent la représentation temporelle de la trame.
On va maintenant faire la boucle qui vérifiera la puissance reçu et qui décidera de si c'est une trame ou du bruit.
for i in range (int(nbr_acq)) :
Donnees = sdr.rx()
Signal_Recu = abs(Donnees/(2**12)) #on garde uniquement la partie absolue
Signal_Recu_complexe = Donnees/(2**12) #on garde le signal complexe
PrW = np.mean(Signal_Recu**2)/1
PrdBm = 10*np.log10(PrW/0.001)
#print(PrdBm)
if PrdBm > niveau_comp :
PuissancesdBm.append(Signal_Recu) #on ajoute la puissance au tableau des puissances
Sig_normalisé.append(Signal_Recu_complexe) #on ajoute le signal complexe au tableau pour la représentation fréquentielle
print("!", end='')
compteur += 1
fin = time.time()
break
else:
print(".", end='')
compteur += 1
PuissancesdBm_list = [item for array in PuissancesdBm for item in array] #on ajoute toutes les valeurs contenu dans les np.array dans une unique liste python
Duree_Calc_reel = (fin - debut)/compteur
Cette boucle possède 3 blocs :
- l'acquisition de la SDR
- une vérification de la puissance reçu
- un arrangement de la liste et calcul de la durée d'une unique acquisition
L'acquisition de la SDR est la même que précédemment. La vérification de la puissance reçu se fait sur la moyenne de la partie réelle du signal. Si elle est supérieure au niveau de comparaison, cela signifie que nous avons reçu une trame. Nous ajoutons ensuite les échantillons reçus dans les tableaux pour les traiter plus tard.
Quand nous ajoutons ces échantillons dans les tableaux, ce sont des np.array intégré dans une liste python. Il faut donc extraire les données inclus dans ce np.array pour les mettre dans la liste python afin de bien faire la représentation temporelle. C'est à ça que sert la ligne d'arrangement de la liste.
Représentation temporelle
En reprenant le code de la partie précédente, on obtient cette représentation temporelle :
 Un peu avant 100 millisecondes, on peut voir 2 états bas long qui indique la synchro. On a donc une trame entière juste après et jusqu'au 2 autres longs états bas vers 175 ms.
Utilisation du signal reçu
Maintenant que nous avons une trame, que nous savons où elle commence et où elle termine, nous pouvons la traiter et récupérer une température.
Zoom sur la trame
Il faut tout d'abord zoomer sur cette trame dans ce que nous avons reçu. C'est à dire savoir à quel échantillon la trame commence et où elle finit. Cette étape se fait à taton en fonction de ce qui est reçu.
# réalisation d'un zoom en échantillons
t_debut = 1250000 # choix en fonction de la figure précédente
t_fin = 3000000 # choix en fonction de la figure précédente
visualisation = PuissancesdBm_list[int(t_debut) : int(t_fin)]
deffig = plt.figure(figsize=(6,2)) # dimensionne le tracé
plt.plot(visualisation)
plt.title("Signal reçu utile : zoom en échantillons")
 Afin d'être sur et certains de la fin de notre trame, on prend en plus quelques bits de la trame d'avant et les 2 bits de synchro de la trame suivante.
Mise en forme du signal
Notre signal comprends des valeurs entre 0 et 0,6 dbmW. Il nous fait maintenant le normaliser c'est à dire traduire ces valeurs en binaire donc leur donner une valeur qui est soit 0 soit 1.
Signal_Norma = []
for x in visualisation:
if x > 0.3:
Signal_Norma.append(1)
else:
Signal_Norma.append(0)
# affichange des données normalisé en vert
deffig = plt.figure(figsize=(6,2)) # dimensionne le tracé
plt.plot( Signal_Norma , color='green')
plt.title("Signal Normalisé en fonction des échantillons")
print("")
Ce code va donc vérifier si la puissance du signal est supérieur à un seuil, si oui on met sa valeur à 1, si non à 0. On obtient donc la représentation temporelle suivante pour notre trame : 
Transformation en trame binaire
Une fois cela fait, on peut compter les états haut et bas et avoir une trame binaire de 42 bits. Ici la valeur des bits ne correspond pas à l'états haut ou bas mais la durée de l'état bas.
On a 3 valeurs possible :
- état bas court (20 000 échantillons) -> 0
- état bas long (40 000 échantillons) -> 1
- état bas très long (80 000 échantillons) -> synchro
On va donc compter les différents états dans notre trame zoomé. Pour cela, on va vérifier si un état est à 1 ou à 0 et en fonction de l'état précédent, on va ajouter 1 à un compteur ou repartir à 0. Si l'état n-1 est le même que l'état n, on ajoute 1. Sinon on repart à 0.
Pour cela une simple et une boucle sont suffisants :
A_Compter = Signal_Norma_reduit
tableau_longueurs = []
compteur_0 = 0
compteur_1 = 0
bit_precedent = 2 # une valeur qui ne soit ni 0 ni 1
for i in A_Compter:
if i == 0 and bit_precedent == 0:
compteur_0 += 1
elif i == 0 and bit_precedent != 0:
compteur_0 += 1
tableau_longueurs.append(compteur_1)
bit_precedent = 0
compteur_0 = 0
elif i == 1 and bit_precedent == 1:
compteur_1 += 1
elif i == 1 and bit_precedent != 1:
compteur_1 += 1
tableau_longueurs.append(compteur_0)
bit_precedent = 1
compteur_1 = 0
On obtient alors un tableau avec la longueur des états :
[0, 215, 4717, 40329, 4708, 40340, 0, 0, 4672, 80333, 4740, 80352, 0, 0, 4742, 20326, 0, 0, 4684, 20324, 4696, 20306, 4713, 20300, 4717, 40352, 4686, 40320, 4717, 20299, 1, 0, 4449, 0, 19, 0, 75, 0, 6, 0, 9, 0, 148, 20323, 4696, 40318, 4716, 40331, 0, 0, 4707, 20319, 4699, 20310, 4709, 40338, 0, 0, 4702, 20317, 4696, 40317, 0, 3, 0, 0, 4585, 0, 127, 20293, 4726, 20307, 4709, 40353, 4, 0, 4681, 20308, 4089, 0, 107, 0, 5, 0, 9, 0, 14, 0, 18, 0, 1, 0, 23, 0, 4, 0, 67, 0, 11, 0, 95, 1, 55, 0, 47, 0, 78, 0, 29, 0, 9, 0, 14, 20303, 0, 0, 4719, 20304, 0, 0, 3942, 0, 84, 0, 54, 0, 11, 0, 122, 0, 42, 0, 6, 0, 238, 0, 16, 0, 183, 20286, 1, 2, 4722, 20298, 4351, 0, 175, 0, 78, 0, 108, 40332, 4706, 40315, 2, 0, 4724, 40370, 4660, 20310, 4716, 20305, 4708, 20304, 1, 0, 4711, 20326, 4694, 20299, 3690, 0, 63, 0, 98, 0, 27, 0, 13, 1, 132, 0, 21, 0, 72, 0, 128, 0, 60, 0, 109, 0, 6, 0, 24, 0, 42, 0, 42, 0, 33, 0, 127, 20294, 4733, 20285, 0, 2, 3678, 0, 39, 0, 6, 0, 127, 0, 73, 0, 160, 0, 9, 0, 33, 0, 24, 0, 12, 0, 16, 0, 15, 0, 15, 0, 492, 20326, 4691, 20304, 3607, 0, 590, 0, 513, 20338, 4683, 20299, 4719, 20315, 0, 0, 4705, 20312, 4609, 0, 96, 40310, 4722, 40318, 4718, 40348, 4669, 80366, 4707, 80342, 4753]
Mise sous forme binaire
Avec ces longueurs brutes, on peut obtenir un nouveau tableau contenant les valeurs binaires. Simplement en vérifiant la taille de l'état, on peut savoir si c'est un 1, un 0 ou une synchro.
Donnees_Binaires = [] # initialisation
for i in tableau_longueurs:
if i < 23000 and i > 19000: #état bas court
Donnees_Binaires.append(0)
elif i < 41000 and i > 38000: #état bas long
Donnees_Binaires.append(1)
elif i < 81000 and i > 79000: #état bas très long (synchro)
Donnees_Binaires.append("Synchro")
On obtient alors un tableau beaucoup plus lisible :
[1, 1, 'Synchro', 'Synchro', 0, 0, 0, 0, 1, 1, 0, 0, 1, 1, 0, 0, 1, 0, 1, 0, 0, 1, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1, 'Synchro', 'Synchro']
Nous voyons cependant des valeurs supplémentaires à notre trames de 42 bits. On va donc restreindre ce tableau pour n'avoir que notre trame.
Une_Trame_Bin = []
compteur_synchro = 0
for i in range (0 , len(Donnees_Binaires)) :
if Donnees_Binaires[i] == "Synchro" :
compteur_synchro = compteur_synchro + 1
if compteur_synchro == 2 :
Une_Trame_Bin.append(Donnees_Binaires[i])
print(Une_Trame_Bin)
On obtient alors notre trame dans un tableau :
La trame avec laquelle on va travailler est: [0, 0, 0, 0, 1, 1, 0, 0, 1, 1, 0, 0, 1, 0, 1, 0, 0, 1, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1]
On va aussi vérifier que la trame fait bien 42 bits. Cette vérification est nécessaire car lors d'une transmission, j'ai réussi à obtenir une trame de 34 bits.
# en supprimant l'identifiant de synchro
La_Trame_Bin = Une_Trame_Bin[1:len(Une_Trame_Bin)]
---
La longueur d'une trame est de : 42 Bits
Enfin, on va afficher les différentes écritures possibles de cette trame : en binaire, en hexadécimal et en décimal.
y = 0
for i in range (0 , len(La_Trame_Bin)) :
bit_i = La_Trame_Bin[i] # juste un élément d'écriture
bit_position = bit_i << (41-i) # place le premier bit extrait de la trame en position 41 (début de chiffre)
# et ainsi de suite
y = y | bit_position # réalisation d'un OU entre la précédente valeur de ma données et la nouvelle
# ceci permet de la rajouter !
LA_TRAME = y
print("La trame étudiée s'écrit : ", LA_TRAME, hex(LA_TRAME), bin(LA_TRAME))
---
La trame étudiée s'écrit : 219731656711 0x3329070007 0b11001100101001000001110000000000000111
CRC
Maintenant qu'on a notre trame écrite, on va pouvoir la vérifier grâce au CRC qu'elle contient.
Le CRC est un mécanisme qui grâce à un polynôme générateur et des divisions euclidienne de la trame avec ce polynôme avant l'envoi permet d'avoir une valeur sur quelques bits qui servent à vérifier l'intégrité de la trame.
Ici le polynôme associé à la suite binaire est constitué des 4 premiers octets de la trame (en ajoutant les 00 de début de trame) auxquels on rajoute les 4 bits de contrôle d'erreur calculés en émission. Ceux-ci sont constitués par les 4 bits de poids faible (à droite) du dernier octet. Soit un division à faire sur : 32 bits + 2 bits + 4 bits = 38 bits
# Récupération du deuxième 1/2 octet du dernier octet le code CRC
Dernier_OctB = LA_TRAME & 0b1111
print("Le code CRC est :", bin(Dernier_OctB))
CRC = Dernier_OctB
# Remplacement du dernier octet de la trame par ce code CRC.
# Remarque : à partir de ce point la trame utilisée fera 42-8+4=38 bits
print("Trame initiale : " , bin(LA_TRAME ))
TrameBinaire34 = LA_TRAME >> 8 # on supprime le dernier octet
print("Trame sans dernier octet : " , bin(TrameBinaire34))
TrameBinaire38 = TrameBinaire34 << 4 # on rajoute 4 bits pour préparer l'arrivée du code CRC
print("Trame 34 + 4 bits : ", bin(TrameBinaire38))
TrameBinaire38crc = TrameBinaire38 | CRC # on rajoute le CRC calculé
print("Trame avec CRC : " ,bin(TrameBinaire38crc))
---
Le code CRC est : 0b111
Trame initiale : 0b11001100101001000001110000000000000111
Trame sans dernier octet : 0b110011001010010000011100000000
Trame 34 + 4 bits : 0b1100110010100100000111000000000000
Trame avec CRC : 0b1100110010100100000111000000000111
Maintenant que la trame a 38 bits dont le CRC, on peut vérifier son intégrité.
dividende = TrameBinaire38crc # pour avoir une boucle généraliste
LongTrame = 38 # le nombre de bits de la séquence à diviser
LongDiv = LongTrame - 4 # pour les 4 derniers bits la division est différente (reste sur 4 bits)
polynome = 0b10011 # le nombre de bits du polynome générateur
LongPoly = 5
TestBPF = 0 # initialisation d'une variable qui permet de tester la valeur du bit de poids fort
TestBit = 0
for i in range (0 , LongDiv) :
TestBit = 1 << (LongDiv - i - 1) # place un "1" logique au meme niveau de que le bit de poids fort du dividende
TestBPF = dividende & TestBit # identification du bit de poid fort
if TestBPF != 0: # si le bit de poids fort est à 1
polynome_decale = polynome << (LongDiv - i - LongPoly) # on décale le polynome à gauche
dividende = dividende ^ polynome_decale # réalisation de la division/OUEX si bit de poid fort à 1
else : # réalisation de la division/OUEX si bit de poid fort à 0 = aucune action
dividende = dividende ^ 0b0 # inutile mais permet de structurer !
Reste = (dividende)
if Reste == 0 :
print("Le reste de cette division est = ", bin(Reste) , "Donc la transmission est bonne !")
else :
print("Le reste de cette division est = ", bin(Reste) , " ce qui n'est pas égal à 0. Donc la transmission n'est PAS bonne ...")
Le reste de cette division est = 0b0 Donc la transmission est bonne !
Notre transmission est réussi ! On peut donc passer à la dernière étape extraire une température de cette suite de 0 et de 1 !!!
Extraction de la température
Pour rappel, la trame a la forme suivante :
Les informations sont donc découpés dans les octets suivants :
- Octet 1 : ID
- Octet 2H : canal
- Octet 2B et 3 : DATA
De plus, la témpérature s'exprime en Fahrenheit de la façon suivante :
- TempF = (Octet3 bas x 256 + Octet3 haut x 16 + Octet2 - 900)/10
(Dans cette expression les octets sont exprimés en décimal.)
#octet 1 = identifiant
id = hex(LA_TRAME)[2:4]
print("L'identifiant de l'emetteur est : ", id)
#octet 2H = canal d'émission
canal = hex(LA_TRAME)[4]
print("L'acquisition est faites sur le canal : ", int(canal) + 1) #+1 pour avoir le vrai canal
#récupération des données en hexa (octets 2B à 3 inclus)
octet_2 = int(hex(LA_TRAME)[5], 16)
octet_3H = hex(LA_TRAME)[6]
octet_3B = hex(LA_TRAME)[7]
print("Octet 2 bas : ", octet_2, "\nOctet 3 haut :", octet_3H, "\nOctet 3 bas: ", octet_3B)
Temp_F = (int(octet_3B) * 256 + int(octet_3H) * 16 + int(octet_2) - 900) / 10
print("Température en °F : ", Temp_F)
#On transforme la température en °C
Temp_C = (Temp_F - 32) / 1.8
print("Température en °C : ", round(Temp_C, 1))
---
L'identifiant de l'emetteur est : 33
L'acquisition est faites sur le canal : 3
Octet 2 bas : 9
Octet 3 haut : 0
Octet 3 bas: 7
Température en °F : 90.1
Température en °C : 32.3
La température émise par la professeure était donc 32,3°C.