Conversion analogique-numérique et acquisition

1. Introduction

Cette page montre comment faire fonctionner le convertisseur analogique-numérique de l'Arduino Due, afin d'effectuer des acquisitions échantillonnées. L'échantillonnage sera fait par des interruptions déclenchées par un chronomètre. Les échantillons seront transmis en flot continu à l'ordinateur, afin d'être tracés dans une fenêtre graphique ou de subir différents traitements.

Le code arduino présenté peut servir de base à une application embarquée traitant des signaux audio.

2. Convertisseur analogique-numérique

2.a. Caractéristiques du convertisseur

Le microcontrôleur SAM3X8E comporte un convertisseur analogique-numérique 12 bits (ADC). La fréquence d'échantillonnage peut en principe atteindre 1 MHz. Un multiplexeur permet d'utiliser jusqu'à 12 entrées analogiques (bornes A0 à A11). Les tensions appliquées à ces entrées doivent être comprises entre 0 et 3,3 V. Lorsque le gain de l'amplificateur d'entrée est égal à 1, cette plage de tension donne des nombres compris entre 0 et 4095. Plus précisément, le convertisseur fonctionne avec une tension de référence VREF (très proche de 3,3 V) et une tension égale à VREF donne le nombre 4096. En pratique, on doit souvent faire la numérisation de signaux alternatifs (par exemple un signal audio). Voici un circuit analogique permettant de faire la conversion, que l'on appellera circuit décaleur :

Figure pleine page

Les entrées IN0 et IN1 sont branchées sur un amplificateur inverseur de gain unité. Le potentiomètre permet de régler le décalage pour qu'une tension nulle deviennent 1,65 V sur les entrées A0 et A1 de l'arduino (qui donnera la valeur médiane 2048 après conversion). L'amplificateur est alimenté directement par la borne 3,3 V de l'arduino. Il doit pouvoir fonctionner en rail to rail (sortie à pleine échelle); nous avons choisi pour cela l'amplificateur double TL2372. Les tensions appliquées en entrées doivent être comprises entre -1,65 V et 1,65 V, ce qui correspond à peu près à l'amplitude maximale d'un signal délivré par la sortie audio d'un ordinateur. En cas de dépassement, l'amplificateur sature à 0 ou 3,3 V, ce qui assure une protection des entrées A0 et A1 (qui sont directement reliées au microcontrôleur).

pmc_enable_periph_clk(ID_ADC);   
                   

adc_init(ADC, VARIANT_MCK, ADC_FREQ_MAX, 0);
                   

adc_configure_timing(ADC, 0, ADC_SETTLING_TIME_3, 1);
                   

adc_set_resolution(ADC,ADC_12_BITS);
                     

NVIC_EnableIRQ (ADC_IRQn) ;
                      

ADC->ADC_IDR=0xFFFFFFFF ;
                          

ADC->ADC_CHDR = 0xFFFF;
                          

ADC->ADC_CHSR = 0x0000;
                          

ADC->ADC_CHER = 0x0000;
                          

ADC->ADC_SEQR1 = 0x00000000;
for (k=0; k<nchannels; k++) {
      ADC->ADC_SEQR1 |= ((channels[k]&0xF)<<(4*k));
      ADC->ADC_CHER |= (1<<k);
    }
                               

ADC->ADC_CGR = 0x00000000;
ADC->ADC_COR = 0x00000000;
for (k=0; k<nchannels; k++) {
    if (gain[k]==2) ADC->ADC_CGR |= 2 << (2*k);
    if (gain[k]==4) ADC->ADC_CGR |= 3 << (2*k);
    if (offset[k]) ADC->ADC_COR |= 1 << k;
}
                                

#define ADC_MR_TRIG1 (1<<1)
ADC->ADC_MR = (ADC->ADC_MR & 0xFFFFFFF0) | ADC_MR_TRIG1 | ADC_MR_TRGEN | ADC_MR_USEQ;
                                

ADC->ADC_IDR=~(1<<channels[0]);
ADC->ADC_IER=1<<channels[0]; 
                                 

void ADC_Handler() {
    // opération à faire après une conversion
}
                                 

uint16_t x = *(ADC->ADC_CDR+channels[k]);
                                 

3. Programmation du chronomètre

Le convertisseur analogique-numérique (ADC) est programmé pour être déclenché périodiquement par un chronomètre. Il faut donc configurer un chronomètre (ou Timer) pour qu'il génère une onde carrée dont la période est égale à la période d'échantillonnage.

Le microcontrôleur SAM3X8E possède 9 chronomètre-compteurs 32 bits (Timer-Counter ou TC). Chacun possède trois voies, chaque voie ayant son propre compteur. Nous allons utiliser la voie 0 du chronomètre TC0. On commence par activer l'horloge pour ce chronomètre :

uint32_t channel = 0;
pmc_enable_periph_clk (TC_INTERFACE_ID + 0*3+channel) ;
             

Pour configuer le chronomètre, on peut utiliser les fonctions de l'API Atmel définies dans hardware/arduino/sam/system/libsam/source/tc.c, ou bien accéder directement aux registres. Pour notre application, les fonctions de l'API Atmel sont suffisantes, mais il faut tout de même consulter la documentation du microcontrôleur SAM3X pour connaître les configurations des registres.

On commence par choisir la fréquence d'horloge du chronomètre :

uint8_t clock = TC_CMR_TCCLKS_TIMER_CLOCK1; // horloge 84MHz/2=42 MHz
              

Pour ce choix, la fréquence est la moitié de celle de l'horloge principale, soit 84/2=42 MHz. Les autres choix possibles sont TC_CMR_TCCLKS_TIMER_CLOCK2 (division par 8), TC_CMR_TCCLKS_TIMER_CLOCK3 (division par 32) et TC_CMR_TCCLKS_TIMER_CLOCK4 (division par 128). Sachant que le nombre de tops d'horloge pour une période d'échantillonnage est codé sur 32 bits, une fréquence d'horloge de 42 MHz permettra de descendre au centième de Hertz.

Le chronomètre peut fonctionner en mode catpure ou en mode waveform. Ce dernier mode permet de générer un signal pour déclencher l'ADC (le mode capture sert à faire des mesures de temps ou de fréquence).

Le compteur proprement dit d'une voie est constitué d'un registre 32 bits nommé CV (Counter Value), qui est incrémenté d'une unité à chaque top d'horloge. Pour notre choix d'horloge, il y a un top d'horloge tous les 1/42 microsecondes. Il y a aussi 3 registres 32 bits nommés RA, RB et RC. Le registre CV peut être comparé aux valeurs stockées dans ces registres pour déclencher différents évènements, selon le principe du déclenchement par seuil. On choisit ici le mode UP_RC (automatic trigger on RC compare), qui remet CV à zéro lorsque la valeur de RC est atteinte. Chaque voie d'un chronomètre comporte deux entrée-sorties TIOA et TIOB (à deux états 0 et 1). Nous allons utiliser TIOA en tant que sortie. La figure suivante montre l'évolution de CV au cours du temps et la sortie TIOA :

Le registre RC contient la période d'échantillonnage, exprimée en nombre de tops de l'horloge (qui est à 42 MHz). Le compteur (CV) revient donc à zéro à cette période. Le registre RA contient la moitié de RC. Pour obtenir la sortie TIOA donnée sur la figure, il faut choisir TC_CMR_ACPC_SET (RC compare effect on TIOA = set) pour que TIOA bascule sur le niveau haut lorsque CV atteint RC, et TC_CMR_ACPA_CLEAR (RA compare effect on TIOA = clear) pour que TIOA bascule sur le niveau bas lorsque CV atteint RA.

La sortie périodique TIOA ainsi générée peut servir à piloter une sortie numérique TTL ou bien à déclencher une interruption. Dans le cas présent, la sortie TIOA déclenche la conversion analogique-numérique (car l'ADC a été configuré pour être piloté par TIOA, avec l'option de déclenchement ADC_MR_TRIG1).

Voici comment se fait la configuration du chronomètre :

TC_Configure(TC0,channel, TC_CMR_WAVE | TC_CMR_WAVSEL_UP_RC | TC_CMR_ACPA_CLEAR | TC_CMR_ACPC_SET |clock); 
                     

Supposons que ticks contienne la période d'échantillonnage en unité de tops d'horloge, sous forme d'un entier non signé de 32 bits. Les valeurs de RC et RA sont attribuées par :

TC_SetRC(TC0,channel,ticks); 
TC_SetRA(TC0,channel,ticks >> 1);
                      

Le déclenchement du chronomètre se fait par :

TC_Start(TC0, channel);
                      

Dès que le chronomètre fonctionne, le signal TIOA qu'il génère déclenche les conversions analogique-numériques. Après chaque conversion, la fonction ADC_Handler est appelée par interruption.

Pour stopper le chronomètre :

TC_Stop(TC0,channel);
pmc_disable_periph_clk (TC_INTERFACE_ID + 0*3+channel);
                      

4. Mémorisation des échantillons

Les nombres obtenus par la conversion analogique-numérique doivent être mémorisés afin d'être transmis par le port USB à l'ordinateur. Pour obtenir un flot continu de données vers l'ordinateur, il faut utiliser un tampon de mémoire (Buffer) circulaire. Plus précisément, on utilise, pour chaque voie à acquérir, NBUF tampons, chacun comportant BUF_SIZE échantillons. Voici la déclaration du tableau contenant ces tampons :

#define MAX_NCHAN 2
#define NBUF 0x10
#define NBUF_MASK 0xF // (NBUF-1)
#define BUF_SIZE 0x100
#define BUF_MASK 0xFF // (BUF_SIZE-1)

uint16_t buf[MAX_NCHAN][NBUF][BUF_SIZE];
              

Le nombre de voies maximal est choisi ici égal à 2. Le nombre de tampons et la taille des tampons doivent être une puissance de 2. Les échantillons sont stockés dans un tampon dans la fonction ADC_Handler. Lorsque le tampon est plein, on écrit sur le suivant. Lorsque le dernier tampon est plein, on écrit à nouveau sur le premier (c'est pourquoi on parle de tampon circulaire). Le programme chargé de transférer les données par le port série sera défini dans la fonction loop. Il effectuera le transfert d'un tampon dès qu'il est plein. Le fait d'avoir plusieurs tampons (ici 16) permet de faire face aux variations de temps d'opération qui peuvent survenir lors du transfert des données par le port série. Si le transfert est légèrement ralenti (à cause par exemple d'un ralentissement de la lecture par l'ordinateur), la mémorisation des échantillons peut continuer. Dans l'exemple ci-dessus, le tableau contient 2*16*256=8192 mots de 16 bits, soit 16 ko (sur un total disponible de 96 ko).

L'accès à un élément d'un tampon se fait par une variable globale 16 bits indice_buf. L'accès à un tampon se fait par la variable globale 8 bits compteur_buf. Voici le contenu de la fonction ADC_Handler appelée par interruption après chaque numérisation :

void ADC_Handler() {
  int k;
  for (k=0; k<nchannels; k++) {
    buf[k][compteur_buf][indice_buf] = *(ADC->ADC_CDR+channels[k]);
  }
  indice_buf++;
  if (indice_buf == BUF_SIZE) { // uint16_t indice_buffer
    indice_buf = 0;
    compteur_buf = (compteur_buf+1)&NBUF_MASK;
  }
}
                  

5. Transmission des données

Pour le transfert des échantillons, on utilise une variable globale 16 bits compteur_buf_transmis qui indique le prochain tampon à transmettre. Lorsqu'il y a des tampons remplis par ADC_Handler mais pas encore transmis, ce compteur est différent de compteur_buf. Voici donc le transfert des données :

int k;
if (compteur_buf_transmis!=compteur_buf) {
    for (k=0; k<nchannels; k++) SerialUSB.write((uint8_t *)buf[k][compteur_buf_transmis],BUF_SIZE*2);
    compteur_buf_transmis=(compteur_buf_transmis+1)&NBUF_MASK;
}
              

Ce bloc sera exécuté dans la fonction loop.

6. Programme Arduino

On présente ici un programme complet pour l'Arduino Due, qui communique avec un ordinateur par le port USB natif (port USB géré par le microcontrôleur).

Voici tout d'abord l'entête avec la déclaration des constantes et des variables globales :

#include "Arduino.h"

#define ACQUISITION 104
#define STOP_ACQUISITION 105
#define ADC_MR_TRIG1 (1<<1)
uint8_t nchannels;
#define MAX_NCHAN 2
uint8_t channels[MAX_NCHAN];
uint8_t gain[MAX_NCHAN];
uint8_t offset[MAX_NCHAN];
uint8_t channels_pins[8] ={7,6,5,4,3,2,1,0};
#define NBUF 0x10
#define NBUF_MASK 0xF // (NBUF-1)
#define BUF_SIZE 0x100
#define BUF_MASK 0xFF // (BUF_SIZE-1)
uint16_t buf[MAX_NCHAN][NBUF][BUF_SIZE];
volatile uint8_t compteur_buf,compteur_buf_transmis;
volatile uint16_t indice_buf;
uint32_t nblocs;
uint8_t sans_fin;
            

MAX_NCHAN est le nombre maximal de voies qui pourront être utilisées. On peut augmenter ce nombre à condition que le tampon tienne dans la mémoire disponible. Le tableau channels_pins contient la correspondance entre le numéro de la borne sur la platine (0=A0, 1=A1, etc) et le numéro de la voie utilisée par l'ADC. Par exemple, la borne A0 correspond à la voie 7.

La fonction suivante configure l'ADC :

void config_adc() {
    int k;
    pmc_enable_periph_clk(ID_ADC);
    adc_init(ADC, VARIANT_MCK, ADC_FREQ_MAX, 0);
    adc_configure_timing(ADC, 0, ADC_SETTLING_TIME_3, 1);
    adc_set_resolution(ADC,ADC_12_BITS);
    NVIC_EnableIRQ (ADC_IRQn) ;   
    ADC->ADC_IDR=0xFFFFFFFF ;  
    ADC->ADC_CHDR = 0xFFFF; 
    ADC->ADC_CHSR = 0x0000; 
    ADC->ADC_WPMR &= ~0x1;
    ADC->ADC_CHER = 0x0000;
    ADC->ADC_SEQR1 = 0x00000000;
    ADC->ADC_CGR = 0x00000000;
    ADC->ADC_COR = 0x00000000;
    for (k=0; k<nchannels; k++) {
      ADC->ADC_SEQR1 |= ((channels[k]&0xF)<<(4*k));
      ADC->ADC_CHER |= (1<<k);
      if (gain[k]==2) ADC->ADC_CGR |= 2 << (2*k);
      if (gain[k]==4) ADC->ADC_CGR |= 3 << (2*k);
      if (offset[k]) ADC->ADC_COR |= 1 << k;
    }
    ADC->ADC_MR = (ADC->ADC_MR & 0xFFFFFFF0) | ADC_MR_TRIG1 | ADC_MR_TRGEN | ADC_MR_USEQ;
    ADC->ADC_IDR=~(1<<channels[0]); 
    ADC->ADC_IER=1<<channels[0];
}
            

La fonction suivante configure et démarre le chronomètre :

void config_timer(uint32_t ticks) {
  uint32_t channel = 0;
  uint8_t clock = TC_CMR_TCCLKS_TIMER_CLOCK1; // horloge 84MHz/2=42 MHz
  pmc_enable_periph_clk (TC_INTERFACE_ID + 0*3+channel) ;
  TC_Configure(TC0,channel, TC_CMR_WAVE | TC_CMR_WAVSEL_UP_RC | TC_CMR_ACPA_CLEAR | TC_CMR_ACPC_SET |clock); 
  TC_SetRC(TC0,channel,ticks); 
  TC_SetRA(TC0,channel,ticks >> 1);
  TC_Start(TC0, channel);
}
            

La fonction suivante stoppe le chronomètre :

void stop_timer() {
    uint32_t channel = 0;
    TC_Stop(TC0,channel);
    pmc_disable_periph_clk (TC_INTERFACE_ID + 0*3+channel);
}
            

Voici la fonction appelée lors des interruptions, c'est-à-dire juste après les conversions :

void ADC_Handler() {
  int k;
  for (k=0; k<nchannels; k++) {
    buf[k][compteur_buf][indice_buf] = *(ADC->ADC_CDR+channels[k]);
  }
  indice_buf++;
  if (indice_buf == BUF_SIZE) { // uint16_t indice_buffer
    indice_buf = 0;
    compteur_buf = (compteur_buf+1)&NBUF_MASK;
  }
}
             

La fonction setup ouvre le port USB et initialise les tampons :

void setup() {
   SerialUSB.begin(115200);
   int i,j,k;
   for (i=0; i<NBUF; i++)
     for (j=0; j<BUF_SIZE; j++)
        for (k=0; k<MAX_NCHAN; k++) 
            buf[k][i][j] = 2048;
   compteur_buf = compteur_buf_transmis = 0;
}
             

La commande ACQUISITION envoyé par l'ordinateur doit être suivi des informations suivantes :

• Le nombre de voies (entier 8 bits).
• Les numéros des bornes de ces voies (entiers 8 bits).
• Les gains pour ces voies (entiers 8 bits).
• Le nombre de tops d'horloges pour la période d'échantillonnage (entier 32 bits).
• Le nombre de blocs d'échantillons à transmettre (entier 32 bits). Chaque bloc comporte BUF_SIZE échantillons pour chaque voie. Si ce nombre est nul, l'acquisition se fait sans fin.

La fonction suivante effectue la lecture de ces données et déclenche l'acquisition :

void lecture_acquisition() {
    uint32_t c1,c2,c3,c4;
    uint32_t ticks;
    int k;
    while (SerialUSB.available()<1) {};
    nchannels = SerialUSB.read();
    while (SerialUSB.available()<nchannels) {};
    for (k=0; k<nchannels; k++) {
      if (k < MAX_NCHAN) channels[k] = channels_pins[SerialUSB.read()]; 
    }
    while (SerialUSB.available()<nchannels) {};
    for (k=0; k<nchannels; k++) {
      if (k < MAX_NCHAN) gain[k] = SerialUSB.read(); 
    }
    while (SerialUSB.available()<nchannels) {};
    for (k=0; k<nchannels; k++) {
      if (k < MAX_NCHAN) offset[k] = SerialUSB.read(); 
    }
    if (nchannels > MAX_NCHAN) nchannels = MAX_NCHAN;
    while (SerialUSB.available()<4) {};
    c1 = SerialUSB.read();
    c2 = SerialUSB.read();
    c3 = SerialUSB.read();
    c4 = SerialUSB.read();
    ticks = ((c1 << 24) | (c2 << 16) | (c3 << 8) | c4);
    while (SerialUSB.available()<4) {};
    c1 = SerialUSB.read();
    c2 = SerialUSB.read();
    c3 = SerialUSB.read();
    c4 = SerialUSB.read();
    nblocs = ((c1 << 24) | (c2 << 16) | (c3 << 8) | c4);
    if (nblocs==0) {
        sans_fin = 1;
        nblocs = 1;
    }
    else sans_fin = 0;
    compteur_buf=compteur_buf_transmis=0;
    indice_buf = 0;
    config_adc();
    config_timer(ticks);
}

             

La fonction suivante stoppe l'acquisition :

void stop_acquisition() {
    ADC->ADC_MR &= ~(0x1);
    stop_timer();
    compteur_buf=compteur_buf_transmis=0;
}
             

La fonction suivante lit le port USB et agit si une commande est présente :

void lecture_serie() {
   char com;
   if (SerialUSB.available()>0) {
        com = SerialUSB.read();
        if (com==ACQUISITION) lecture_acquisition();
        if (com==STOP_ACQUISITION) stop_acquisition();
   }
}
             

Voici la fonction loop. Elle détecte la présence d'un tampon plein non encore transmis. Le cas échéant, elle transmet ce tampon puis décrémente le compteur de blocs dans le cas d'une acquisition avec un nombre fini de blocs (un bloc est le contenu d'un tampon, soit BUF_SIZE échantillons).

void loop() {
    int k;
    if ((compteur_buf_transmis!=compteur_buf)&&(nblocs>0)) {
        for (k=0; k<nchannels; k++) SerialUSB.write((uint8_t *)buf[k][compteur_buf_transmis],BUF_SIZE*2);
        compteur_buf_transmis=(compteur_buf_transmis+1)&NBUF_MASK;
        if (sans_fin==0) {
            nblocs--;
            if (nblocs==0) stop_acquisition();
        }
    }
    else lecture_serie();
}
             

7. Programme python

import serial
import numpy
import time
import threading
            

Le programme python (pour python 3.x) comporte une classe dont les fonctions permettent d'envoyer les données de configuration à l'arduino.

Le constructeur ouvre la communication avec l'arduino et définit des constantes, qui doivent bien sûr être identiques à celles définies dans le programme arduino.

class Arduino:
    def __init__(self,port):
        self.ser = serial.Serial(port,115200)
        time.sleep(2)
        self.ACQUISITION = 104
        self.STOP_ACQUISITION = 105
        self.clockFreq = 42.0e6 # frequence d'horloge
        self.TAILLE_BLOC = 256
        self.TAILLE_BLOC_INT8 = self.TAILLE_BLOC*2
        
    def close(self):
        self.ser.close()
             

Les fonctions suivantes permettent d'envoyer un entier 8 bits, 16 bits, et 32 bits. Les octets de poids fort sont envoyés en premier, conformément à la convention utilisée dans le code arduino présenté plus haut (convention big endian).

    def write_int8(self,v):
        self.ser.write((v&0xFF).to_bytes(1,byteorder='big'))

    def write_int16(self,v):
        v = numpy.int16(v)
        char1 = int((v & 0xFF00) >> 8)
        char2 = int((v & 0x00FF))
        self.ser.write((char1).to_bytes(1,byteorder='big'))
        self.ser.write((char2).to_bytes(1,byteorder='big'))
        
    def write_int32(self,v):
        v = numpy.int32(v)
        char1 = int((v & 0xFF000000) >> 24)
        char2 = int((v & 0x00FF0000) >> 16)
        char3 = int((v & 0x0000FF00) >> 8)
        char4 = int((v & 0x000000FF))
        self.ser.write((char1).to_bytes(1,byteorder='big'))
        self.ser.write((char2).to_bytes(1,byteorder='big'))
        self.ser.write((char3).to_bytes(1,byteorder='big'))
        self.ser.write((char4).to_bytes(1,byteorder='big'))       
              

La fonction suivante lance une acquisition. voie est la liste des voies utilisées. gains est la liste des gains correspondants (1,2 ou 4). offset est un tableau configurant l'offset pour chaque voie : 0 pour ne pas appliquer d'offset avant la conversion, 1 pour retrancher VREF/2 avant la conversion. nb est le nombre de blocs à acquérir, qui doit être nul pour une acquisition sans fin.

    def lancer_acquisition(self,voies,gains,offsets,fechant,nb):
        ticks = int(self.clockFreq/fechant)
        self.write_int8(self.ACQUISITION)
        self.nvoies = nv = len(voies)
        self.write_int8(nv)
        for k in range(nv):
            self.write_int8(voies[k])
        for k in range(nv):
            self.write_int8(gains[k])
        for k in range(nv):
            self.write_int8(offsets[k])
        self.write_int32(ticks)
        self.write_int32(nb)
              

La fonction suivante stoppe l'acquisition en cours :

    def stopper_acquisition(self):
        self.write_int8(self.STOP_ACQUISITION)
              

La fonction suivante effectue la lecture d'un bloc (pour une voie). Elle renvoie un tableau de dtype=numpy.float32, qui contient en fait des nombres entiers compris entre 0 et 4095 (converion 12 bits).

    def lecture(self):
        buf = self.ser.read(self.TAILLE_BLOC_INT8)
        data = numpy.zeros(self.TAILLE_BLOC,dtype=numpy.float32)
        j = 0
        for i in range(self.TAILLE_BLOC):
        	data[i] = buf[j]+0x100*buf[j+1]
        	j += 2
        return data
              

Dans certains cas, on ne cherche pas à convertir les nombres entiers en valeurs de tension car la tension n'est pas l'information recherchée. Par exemple pour un microphone, il suffira de retrancher la valeur moyenne correspondant au niveau zéro (en principe 2048). Si l'on souhaite calculer la tension, il faut tenir compte de l'existence d'un décalage (offset) et du gain (1,2 ou 4). Si un décalage a été appliqué à la tension en entrée (par exemple avec le circuit présenté plus haut), ce décalage doit être égal à VREF/2, qui correspond à la valeur 2048. Si le gain est différent de 1, le circuit qui applique le décalage doit être réglé afin que ce décalage soit effectivement VREF/2. Pour régler ce circuit, il faut appliquer une tension nulle en entrée du circuit puis ajuster le décalage afin que la valeur après conversion soit égale à 2048 (au bit près); la tension du décalage (lue au voltmètre) donne la valeur de VREF/2. Si le gain est 1, on peut utiliser le décalage que l'on veut et il est alors plus simple de relever la valeur obtenue lorsque la tension en entrée est nulle. Si x est le nombre entier entre 0 et 4095 et G le gain du convertisseur (1,2 ou 4) alors la tension est donnée par :

Après que l'acquisition a été lancée, il faut faire une lecture asynchrone des données envoyées par l'arduino. On se fixe comme objectif de traiter ou d'afficher les signaux par paquets, chaque paquet étant constitué d'un certain nombre de blocs. Les blocs sont les unités transmises par l'arduino, qui contiennent ici 256 échantillons. Nous allons faire la lecture des données envoyées par l'arduino sur un fil d'exécution. On doit pour cela définir une classe qui hérite de threading.Thread. Voici le constructeur de cette classe :

class AcquisitionThread(threading.Thread): 
    def __init__(self,arduino,voies,gains,offsets,fechant,nblocs,npaquets):
        threading.Thread.__init__(self)
        self.arduino = arduino
        self.nvoies = len(voies)
        self.voies = voies
        self.gains = gains
        self.offsets = offsets
        self.fechant = fechant
        self.running = False
        self.nblocs = nblocs # nombre de blocs dans un paquet
        self.npaquets = npaquets
        self.taille_bloc = arduino.TAILLE_BLOC
        self.taille_paquet = self.taille_bloc * self.nblocs
        self.data = numpy.zeros((self.nvoies,self.nblocs*arduino.TAILLE_BLOC*self.npaquets))
        self.compteur_paquets = 0
        self.compteur_paquets_lus = 0
        self.nechant = 0
              

nblocs est le nombre de blocs dans un paquet (chaque bloc contient TAILLE_BLOC=256 échantillons) et npaquets et le nombre de paquets à acquérir. Les échantillons seront stockés dans le tableau self.data, dont chaque ligne correspond à une voie.

La fonction run est exécutée lorsqu'on lance le Thread. Elle consiste en une boucle qui lit les données provenant de l'arduino et les stocke dans self.data. Cette boucle incrémente self.compteur_paquets à chaque fois qu'un paquet a été transmis par l'arduino.

    def run(self):
        self.arduino.lancer_acquisition(self.voies,self.gains,self.offsets,self.fechant,0) # acquisition sans fin
        self.running = True
        indice_bloc = 0
        while self.running:
            i = self.compteur_paquets*self.taille_paquet +indice_bloc*self.taille_bloc
            j = i+self.taille_bloc
            for v in range(self.nvoies):
                self.data[v,i:j] = self.arduino.lecture()
            self.nechant = j
            indice_bloc += 1
            if indice_bloc==self.nblocs:
                indice_bloc = 0
                self.compteur_paquets += 1
                if self.compteur_paquets==self.npaquets:
                    self.stop()
               

La fonction suivante permet de stopper l'acquisition :

    def stop(self):
        self.running = False
        time.sleep(1)
        #self.join()
        self.arduino.stopper_acquisition()
               

L'appel self.join() permet d'attendre la sortie de la fonction run.

Voici la fonction qui permet de récupérer le dernier paquet. S'il n'y a pas de paquet encore disponible, elle renvoie -1. Deux appels consécutifs de cette fonctions renvoient deux paquets consécutifs (ou -1).

    def paquet(self): # obtention du dernier paquet
        if self.compteur_paquets==self.compteur_paquets_lus:
            return -1
        i = self.compteur_paquets_lus*self.taille_paquet
        j = i+self.taille_paquet                    
        self.compteur_paquets_lus += 1
        return self.data[:,i:j]
               

Dans certains cas, on ne veut pas attendre qu'un paquet soit disponible pour obtenir les derniers échantillons. La fonction suivante permet de récupérer les nombre derniers échantillons. La fonction renvoie les instants, les échantillons et la durée correspondante. Si le nombre d'échantillons demandé n'est pas encore disponible, elle renvoie les valeurs -1.

    def echantillons(self,nombre):
        j = self.nechant
        i = j-nombre
        if i<0: i=0
        if j-i<=0: return (-1,-1,-1)
        temps = numpy.arange(j-i)/self.fechant
        tmax = nombre/self.fechant
        return (temps,self.data[:,i:j],tmax)
                

8. Test avec tracé du signal

Le programme de test suivant effectue une acquisition sur deux voies avec un tracé des deux signaux dans une fenêtre matplotlib gérée par une animation. Avec la fonction animate1, les données sont récupérées par paquets à un intervalle de temps égal au temps qu'il faut pour numériser un paquet. Avec la fonction animate2, on récupère les nombre_echant derniers échantillons toutes les 50 ms. Pour les signaux rapides, il est préférable d'utiliser animate1. Pour les signaux lents (dont les paquets sont plus longs qu'une seconde), il est préférable d'utiliser animate2 car celle-ci permet de rafraichir le tracé au fûr et à mesure de l'arrivée des échantillons.

import numpy
from matplotlib.pyplot import *
import matplotlib.animation as animation
from arduinoDueAcquisitionSignal import *

ard = Arduino("COM9") 
fechant = 40000
nblocs = 20 # nombre de blocs dans un paquet (un bloc = 256 échantillons)
duree_paquet = nblocs*ard.TAILLE_BLOC/fechant
print("durée d'un paquet = %f"%(duree_paquet))
npaquets = 200 # nombre de paquets
print("durée totale = %f"%(duree_paquet*npaquets))
nechant = ard.TAILLE_BLOC*nblocs # nombre d'échantillons dans un paquet
delai = duree_paquet
nombre_echant = 1000 # nombre d'échantillons lus à chaque rafraichissement (animate2)
t = numpy.arange(nechant)*1.0/fechant
fig,ax = subplots()
line0, = ax.plot(t,numpy.zeros(nechant))
line1, = ax.plot(t,numpy.zeros(nechant))
ax.axis([0,t.max(),-100,5000]) # fenêtre pour un paquet
ax.grid()

voies = [0,1]
G=1
gains = [G,G]
offsets = [1,1]
acquisition = AcquisitionThread(ard,voies,gains,offsets,fechant,nblocs,npaquets)
acquisition.start()

reglage = False
if reglage:
    ax.axis([0,t.max(),2000,2096])
    ax.plot([0,t.max()],[2048,2048])
conversion = True
if conversion:
    VREF = 1.646*2 # à lire au voltmètre
    conv = VREF/(2*G)/2048
    ax.axis([0,t.max(),-VREF/2,VREF/2])

def animate1(i):
    global line0,line1,acquisition,signal,conversion,conv
    data = acquisition.paquet()
    if isinstance(data,int)==False:
        if conversion:
            data[0] = (data[0]-2048)*conv
            data[1] = (data[1]-2048)*conv
        line0.set_ydata(data[0])
        line1.set_ydata(data[1])
        
            

def animate2(i):
    global line0,line1,acquisition,conversion,VREF,conv
    (temps,data,tmax) = acquisition.echantillons(nombre_echant)
    if isinstance(data,int)==False:
        if conversion:
            data[0] = (data[0]-2048)*conv
            data[1] = (data[1]-2048)*conv
            ax.axis([0,tmax,-VREF/2,VREF/2])
        else:
            ax.axis([0,tmax,-100,5000])
        line0.set_ydata(data[0])
        line1.set_ydata(data[1])
        line0.set_xdata(temps)
        line1.set_xdata(temps)

ani = animation.FuncAnimation(fig,animate1,100,interval=delai*1000)
#ani = animation.FuncAnimation(fig,animate2,100,interval=50)
show()
acquisition.stop()
ard.close()
numpy.savetxt("signaux-1.txt",acquisition.data.T,delimiter="\t",fmt="%0.4e")
show()          
             

Voyons comment effectuer le réglage et l'étallonnage. On doit brancher un voltmètre entre la masse et l'entrée A0 de l'arduino. L'entrée IN0 du circuit décaleur étant à la masse, on tourne le potentiomètre pour que la tension sur l'entrée A0 soit à 1,65 V. On pose reglage=True dans le script ci-dessus, ce qui permet de visualiser précisément l'écart entre les valeurs des échantillons et la valeur 2048 (le bruit de quantification est bien visible). Il faut tourner le potentiomètre du circuit décaleur pour que la valeur moyenne soit à 2048. La tension lue au voltmètre sur l'entrée A0 est alors VREF/2. La valeur de VREF doit être reportée dans le script ci-dessus.

Si conversion=True les valeurs converties en tension sont tracées. Sinon, les nombres entiers (entre 0 et 4095) sont tracés. Le choix d'un offset 1 pour une voie permet d'amplifier (d'un gain 2 ou 4) le signal alternatif appliqué sur l'entrée IN0 ou IN1 du circuit décaleur. Si l'on souhaite traiter un signal à valeurs positives (par exemple celui provenant d'une photodiode), on n'utilise pas de circuit décaleur et le gain est appliqué sans offset.

Pour tester le bon fonctionnement de l'échantillonnage, on numérise une sinusoïde d'amplitude 1 V et de fréquence 200,0 Hz (fréquence d'échantillonnage 40 kHz, durée 12,8 s) puis on fait l'analyse spectrale de la manière suivante :

import numpy
from matplotlib.pyplot import *
import scipy.signal

def spectre(t,u):
    N=len(u)
    te=t[1]-t[0]
    zeros=numpy.zeros(6*N)
    U = numpy.concatenate((u*scipy.signal.blackman(N),zeros))
    NN=len(U)
    spectre = numpy.absolute(numpy.fft.fft(U))*2.0/N/0.42
    freq = numpy.arange(NN)*1.0/(NN*te)
    return (freq,spectre)

fe=40000
te=1/fe
[u0,u1] = numpy.loadtxt("signaux-1.txt",unpack=True)
N=len(u0)
t=numpy.arange(N)*te
(f,A)=spectre(t,u0)


figure()
plot(f,A)
xlabel("f (Hz)")
ylabel("A (V)")
xlim(195,205)
grid()

show()               
               

Pour faire un test très précis, il est bon d'utiliser un générateur numérique, qui permet d'ajuster la fréquence très précisément.