Electric Train V3

Feel free to have a look on the V2 first:

The biggest change is in the camera able to follow the phone orientation to update its angle. I also replace the front/rear sensors.


Camera tracking

I used the html5 API to detect the phone orientation:

if (window.DeviceOrientationEvent) {
  // Our browser supports DeviceOrientation
  window.addEventListener("deviceorientation", deviceOrientationListener);
} else {
  console.log("Sorry, your browser doesn't support Device Orientation");
function deviceOrientationListener(event) {
  var c = document.getElementById("myCanvas");
  var ctx = c.getContext("2d");

  ctx.clearRect(0, 0, c.width, c.height);
  ctx.fillStyle = "#FF7777";
  ctx.font = "14px Verdana";
  ctx.fillText("Alpha: " + Math.round(event.alpha), 10, 20);
  ctx.moveTo(180, 75);
  ctx.lineTo(210, 75);
  ctx.arc(180, 75, 60, 0, event.alpha * Math.PI / 180);

  ctx.fillStyle = "#FF6600";
  ctx.fillText("Beta: " + Math.round(event.beta), 10, 140);
  ctx.fillRect(180, 150, event.beta, 90);

  ctx.fillStyle = "#FF0000";
  ctx.fillText("Gamma: " + Math.round(event.gamma), 10, 270);
  ctx.fillRect(90, 340, 180, event.gamma);
  var aMsg = event.alpha.toString()+"_"+event.beta.toString()+"_"+event.gamma.toString();
  console.log("aMsg" + aMsg);

Which send the 3 orientation information to the Tornado python server running on the Raspberry pi of the train. First I was doing JSON REST call to send the string containing the information but it was too slow to have the camera moving in real time. This was the perfect opportunity to use websocket for more real time communication.

function onOpen(evt) { 
        doSend("Hi there!");
    function onClose(evt) { 
    function onMessage(evt) { 
        console.log('message: ' + evt.data);
    function onError(evt) { 
        writeToScreen('error' + evt.data);
    function doSend(message) { 
function testWebSocket() {
        websocket.onopen = function(evt) { onOpen(evt) };
        websocket.onclose = function(evt) { onClose(evt) };
        websocket.onmessage = function(evt) { onMessage(evt) };
        websocket.onerror = function(evt) { onError(evt) };

if (!'WebSocket' in window){
    console.log("Sorry, your browser doesn't support Websockets");
} else {
var wsUri = "ws://";
var websocket = new WebSocket(wsUri);

Which is received on the server side and put in a variable (see the class Handler_WS) :

    def on_message(self, iMessage):
        """Methode call when the server receive a message"""
        logging.info('Receive incoming message:'+str(iMessage))

This variable is then read every 125ms by the “foo” function:

tornado.ioloop.PeriodicCallback(lambda: foo(aTrain), 125).start()

At the end the real method called is in charge of updating the turret position. The whole stuff is based on an existing framework called servoBlaster which will take care of driving the Servo.

def updateTurretFromScreenAngle(self):
        if (self._cellAngles!=""):
            #Update Gamma
            aGamma = self._cellAngles.split("_")[2]
            aGammaF = float(aGamma)
            aGammaI = int(aGammaF)
            aGammaisNegative = False
            if (aGammaI<0): aGammaI=(aGammaI*-1)-40 aGammaisNegative = True else: aGammaI=140-aGammaI if ((aGammaI>0)and(aGammaI<100)):
                self._turretHeight = 100 - aGammaI
            #Update Alpha

Servo Blaster is library able to drive Servo on the Raspberry pi using software PWM. It is pretty hard to do since the Pi is not running a real-time OS. It relies on very low level interruption to ensure the timing needed to have a proper PWM are respected. You can have more info on it here:


It basically start a daemon (which I added in the crontab to be launch at boot time) on which you can interact with writing the desired position of each servo in /dev/servoblaster like:

echo 3=120 > /dev/servoblaster 

I also used servo blaster to send PWM info to the motor driver to change the train speed (since this functionality was broken when I moved from Arduino to Rapsberry Pi).

Contact sensors

I replace the old contact sensor by some new sensor able to detect an incoming obstacle before impact.


They are still binary sensors that will turn high if they detect an obstacle but they have a wider range between 2 and 10 centimeters. This allows the train to detect incoming obstacle and stop before hitting it. The sensor is available on ADAfruit:


I made some videos on this new version on YouTube:


As always the code is available here:

Remplacement serveur web lighttpd par twisted

Dans mon système domotique fait maison j’utilise twisted pour la partie “logique” du système et lighttpd pour la UI. Pour des raisons de simplicité j’ai décidé d’utiliser twisted pour gérer la partie Web.

ChangeSet : https://code.google.com/p/domotique-1/source/detail?r=281ebe02e19eef0f07098430de31681c8a836713

Il y a juste un petit piège pour la gestion du php dans twisted. Il faut definir une classe en charge de la gestion des pages php :

class PhpScript(twcgi.FilteredScript):
 Twisted wrapping class to execute PHP code.
 filter = '/usr/bin/php-cgi' # Points to the perl parser
def runProcess(self, env, request, qargs=[]):
 env['REDIRECT_STATUS'] = ''
 return FilteredScript.runProcess(self, env, request, qargs)

En cas de probleme il faudra verifier que le binaire “/usr/bin/php-cgi” est present.
Ensuite il faut declarer la classe dans le reacteur avec :

root.processors = {".php": PhpScript}

Test nouvelle API freebox OS pour arret automatique Wifi

Avec le déploiement du Freebox OS depuis Juin 2013 l’API pour dialoguer avec la Freebox évolue. Heureusement Free offre une très bonne documentation pour utiliser cette nouvelle API :


La documentation est plutôt claire et comporte beaucoup d’exemple.

J’ai décidé de tester cette nouvelle API pour 2 raisons :

  1. Mon système de domotique actuel utilise la Freebox pour le control de la Télé. Pour le moment il utilise la vieille API :
    def sendMsgToFreebox(self,key):
    url = "http://hd1.freebox.fr/pub/remote_control?" + "key=" + key + "&code=59999459"
    logging.error("sending : " + url)
    reponse = urllib.urlopen(url)
    logging.error("reponse : " + str(reponse))

    Cette API sera peut-être remplace par la nouvelle version et je serais donc obliger de migrer

  2. Je veux ajouter le control du wifi sur le système domotique pour pouvoir allumer ou éteindre le wifi. Le but final étant un poil plus complexe.

Je veux créer une application android qui allume le wifi de l’appartement en même temps que le wifi de la tablette. Plus exactement l’application android devra envoyer un message au système domotique pour allumer le wifi. Le système allumera le wifi s’il n’est pas déjà en route.

Ensuite le système domotique monitor le nombre de client utilisant le wifi en temps réel. Il peut également prendre la décision de couper le wifi si aucun client est connecte depuis un certain temps. Le but de cette application est de couper le wifi quand il n’est pas utilisé pour réduire les ondes présentes dans l’appartement.

Dans un second temps je souhaite ajouter un bouton prêt de la Freebox pour pourvoir effectuer une demande d’allumage du wifi (comme le fait l’application android) pour les tablettes qui n’ont pas de connexion 3G et ne peuvent donc pas envoyer la demande d’allumage wifi par la 3G.

L’image ci-dessous résume le flow :


1 – L’utilisateur active le wifi sur sa tablette Wifi/3G. L’application envoie une demande d’activation Wifi au Raspberry Pi en utilisant le réseau 3G


1bis – L’utilisateur appuie sur le bouton d’activation du wifi sur le Raspberry Pi

2 – Le système domotique héberge sur le Raspberry Pi envoie une demande à la Freebox d’activation du wifi si le wifi n’est pas déjà actif

3 – L’utilisateur se connecte automatiquement au wifi et peut maintenant utilise le réseau wifi pour aller sur internet

4 – Le système domotique monitor régulièrement le nombre d’utilisateur wifi. Si le nombre de device est de 0 depuis plus de X temps (1 heure par défaut) le système éteint le wifi de la Freebox en utilisant l’API.

Le système complet ne fonctionne pas encore. Pour le moment j’arrive uniquement à me logger sur la Freebox (système de challenge avec token). Je continuerai les étapes suivantes les WE prochains et j’updaterai l’article. Voila le code actuel :

# -*- coding: utf-8 -*-

import requests
import logging
import time
import itertools
import json
import os
import sys

from hashlib import sha1
import hmac

class FreeboxApplication:
'''Represents an application which interact with freebox server
API doc : http://dev.freebox.fr/sdk/os/'''

def __init__(self):
#I kept the same parameter name than the one use in freebox API for more readability
#To know if the APP is register on freeboxOS side
#Registration parameters

def __repr__(self):
aRetString = ""
aRetString = aRetString + "self.app_id: " + str(self.app_id)
aRetString = aRetString + "self.app_name: " + str(self.app_name)
aRetString = aRetString + "self.app_version: " + str(self.app_version)
aRetString = aRetString + "self.device_name: " + str(self.device_name)
aRetString = aRetString + "self.registerIntoFreeboxServer: " + str(self.registerIntoFreeboxServer)
aRetString = aRetString + "self.app_token: " + str(self.app_token)
aRetString = aRetString + "self.track_id: " + str(self.track_id)
aRetString = aRetString + "self.challenge: " + str(self.challenge)
return aRetString

def getataForRequests(self):
return json.dumps({"app_id": self.app_id,"app_name": self.app_name,"app_version": self.app_version,"device_name": self.device_name})

def loadAppTokenFromFile(self):
if (os.path.isfile("AppToken.txt")):
aAppTokenBackupFile = open("AppToken.txt", "r")
self.app_token = aAppTokenBackupFile.read()
logging.info("APP token read from file. New APP token is : " + str(self.app_token))
logging.info("No file for APP token - request a new one")
#Fin Degeu

def initialLogging(self):
#only once. Register the APP on freebox side
logging.info("Starting initial registration")
aRequestUrl = "http://mafreebox.freebox.fr/api/v1/login/authorize/"
aHeaders = {'Content-type': 'application/json', 'Accept': 'application/json'}

logging.debug("URL used : " + aRequestUrl)
logging.debug("Datas used : " + str(self.getataForRequests()))

aRequestResult = requests.post(aRequestUrl, data=self.getataForRequests(), headers=aHeaders)
logging.debug("Request result : " + str(aRequestResult))
logging.debug("Request result : " + str(aRequestResult.json()))
logging.debug("Registration result : " + str(aRequestResult.json()['success']))

#if (aRequestResult.status_code != "200") or (aRequestResult.json()['success'] != True):
if (aRequestResult.status_code != requests.codes.ok) or (aRequestResult.json()['success'] != True):
logging.critical("Error during intial registration into Freebox Server")
logging.debug("Please go to your Freebox. There should be a message saying that an application request access to freebox API. Please validate the request using the front display")
self.app_token = aRequestResult.json()['result']['app_token']
self.track_id = aRequestResult.json()['result']['track_id']
logging.debug("app_token : " + str(self.app_token))
logging.debug("track_id : " + str(self.track_id))
logging.info("Ending initial registration")

aLoopInd = 0
while ((self.registerIntoFreeboxServer != True) and (aLoopInd < 10)):
time.sleep(15) # Delay for 1 minute (60 seconds)
aLoopInd = aLoopInd + 1
if (self.registerIntoFreeboxServer != True):
logging.critical("Initial registration fails - Exiting with error")
aAppTokenBackupFile = open("AppToken.txt", "w")
#Fin Degeu

def trackRegristration(self):
logging.info("Starting trackRegristration")
aRequestUrl = "http://mafreebox.freebox.fr/api/v1/login/authorize/" + str(self.track_id)
aHeaders = {'Content-type': 'application/json', 'Accept': 'application/json'}

logging.debug("URL used : " + aRequestUrl)

aRequestResult = requests.get(aRequestUrl, headers=aHeaders)
logging.debug("Request result : " + str(aRequestResult))
logging.debug("Request result : " + str(aRequestResult.json()))
if (aRequestResult.status_code != requests.codes.ok):
logging.critical("Error during trackRegristration")
if (aRequestResult.json()['result']['status'] == "granted"):
logging.debug("OK during trackRegristration")
logging.info("APP is correclty registered")
logging.info("Ending trackRegristration")

def logWithPassword(self, iPassword):
#only once. Register the APP on freebox side
logging.info("Starting logWithPassword")
aRequestUrl = "http://mafreebox.freebox.fr/api/v1/login/session/"
aHeaders = {'Content-type': 'application/json', 'Accept': 'application/json'}

logging.debug("URL used : " + aRequestUrl)

aDataToLog = json.dumps({"app_id": self.app_id,"password": iPassword})

logging.debug("Datas used : " + str(aDataToLog))

aRequestResult = requests.post(aRequestUrl, data=aDataToLog, headers=aHeaders)
logging.debug("Request result : " + str(aRequestResult))
logging.debug("Request result : " + str(aRequestResult.json()))
logging.debug("Registration result : " + str(aRequestResult.json()['success']))

#if (aRequestResult.status_code != "200") or (aRequestResult.json()['success'] != True):
if (aRequestResult.status_code != requests.codes.ok) or (aRequestResult.json()['success'] != True):
logging.critical("Error during intial registration into Freebox Server")
logging.debug("You re log")
logging.info("Ending logWithPassword")

def loginProcedure(self):
logging.info("Starting loginProcedure")
aRequestUrl = "http://mafreebox.freebox.fr/api/v1/login/"
aHeaders = {'Content-type': 'application/json', 'Accept': 'application/json'}

logging.debug("URL used : " + aRequestUrl)

aRequestResult = requests.get(aRequestUrl, headers=aHeaders)
logging.debug("Request result : " + str(aRequestResult))
logging.debug("Request result : " + str(aRequestResult.json()))
if (aRequestResult.status_code != requests.codes.ok):
logging.critical("Error during loginProcedure")
if (aRequestResult.json()['success'] == True):
logging.debug("OK during loginProcedure")
logging.info("We have the challenge : " + str(achallenge))
return achallenge
logging.critical("Error during loginProcedure")
logging.info("Ending loginProcedure")

def computePassword(self, iChallenge):
hashed = hmac.new(self.app_token, iChallenge, sha1)
logging.info("Password computed : " + str(hashed.digest().encode('hex')))
return hashed.digest().encode('hex')

def loginfull(self):
aNewChallenge = self.loginProcedure()
#password = hmac-sha1(app_token, challenge)
#voir http://stackoverflow.com/questions/8338661/implementaion-hmac-sha1-in-python
aPassword = self.computePassword(aNewChallenge)

print ("Starting")


#Clean previous log file
with open(aLogFileToUse, 'w'):

logging.basicConfig(filename=aLogFileToUse,level=logging.DEBUG,format='%(asctime)s - %(levelname)s - %(message)s')

aMyApp = FreeboxApplication()


print ("Ending")

L’ensemble du code est dispo sur mon dépôt de domotique.

Update 3 octobre :

Une fois logger on peut utiliser l’API mais il n’y a pas bcp de permission setter a ON par défaut. Il faut mieux aller dans l OS freebox et tout autoriser avant de poursuivre les tests :

FreeboxOs permissions

Ensuite on verifie que les permissions sont bien prise en compte. Pour cela on refait une demande de login et on regarde la reponse :

2013-10-03 19:40:08,070 - DEBUG - Request result : {u'result': 
{u'challenge': u'7IxxxxxxxxxxxwP+cb', u'password_salt': u'uALF+MxxxxxxxxqYi',
 u'permissions': {u'contacts': True, u'settings': True, u'explorer': True,
 u'calls': True, u'downloader': True}, u'session_token': u'K6jjKxxxxxxxxxxxxxxxxxxxxxWCq'}, u'success': True}

Maintenant les permissions sont toutes a True….. On peut continuer. J’ai changer le script pour récupérer la liste des appareils connecte au wifi. J’ai beaucoup de chance car il y a également des infos sur les dernières connexion et je pourrai donc faire un système un peu plus intelligent pour couper le wifi même si il y a des appareil connecte du moment qu ils n’utilisent pas réellement le wifi.

La dernière version est dispo sur mon dépôt de domotique et liste les appareils wifi connecte (et leur nombre). Elle ne coupe/active pas encore le wifi….a suivre

Detection visage en Python avec OpenCV et camera IP

J’ai ajoute une webcam dans le salon qui stream en direct sur internet mais le but est de détecter les personnes pressentes dans le salon et de pouvoir créer des actions plus intelligentes que celle que j’ai aujourd’hui.

Actuellement il y a une platine Arduino avec un détecteur de présence, température, humidité dans le salon. Le système est donc capable de détecter les personnes mais il ne peut qu’ouvrir les volets pou allumer une lumière.

Ce que je veux faire maintenant est de détecter les personnes présentes dans le salon et en fonction de leur attitude le système effectuera des actions plus intelligentes. Par exemple si quelqu’un se met sur le canapé face a la télévision…..alors on allume la télévision automatiquement 😉

La première étape a été l achat d’une camera IP low cost. J’ai choisit ce modèle car d’après les forums que j’ai pu voir leur framework est plutôt ouvert et on peut accéder au flux de la camera facilement depuis openCV :


Foscam FI8910W

Ensuite… on trouve pas mal de tuto sur google pour utiliser openCV et les cameraIP (il suffit de mixer les 2 codes 😉 ). OpenCV s’installe très facilement sur Ubuntu et s’interface plutôt facilement avec python grâce a la libraire CV2 (ne pas utiliser CV).

Voila le code fonctionnel :

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import cv2, math
import numpy as np

#vc = cv2.VideoCapture("")
vc = cv2.VideoCapture(0)

if vc.isOpened(): # try to get the first frame
    rval, frame = vc.read()
    rval = False

while rval:
    cv2.imshow("preview", frame)
    rval, frame = vc.read()

    hc = cv2.CascadeClassifier("/usr/share/opencv/haarcascades/haarcascade_frontalface_alt2.xml")
    faces = hc.detectMultiScale(frame)
    for face in faces:
        cv2.rectangle(frame, (face[0], face[1]), (face[0] + face[2], face[0] + face[3]), (255, 0, 0), 3)

    key = cv2.waitKey(20)
    if key == 27: # exit on ESC

Il faut juste faire attention que le fichier xml de filtre existe bien (sinon le telecharger sur internet et changer la path). Le résultat avec mon colloc 😉


Étape suivante…..porter tout ça sur Raspbian….et je sens que ça va être pénible. Ensuite il faudra être capable de dire ce que font les personnes présentes dans le salon….



Migration du server/client domotique de “Select” vers “twisted”

Le serveur domotique hébergé sur mon RaspberryPi déclenche certaines actions en fonction d’évènements. Par exemple lorsqu’ un détecteur de présence est déclenché on allume la lumière de la même pièce. Dans cet exemple l’évènement sera reçût par le port USB du RaspberryPi (car il proviendra de l’Arduino Leonardo). Il existe d’autre évènements qui arriveront sur le RaspberryPi par d’autre moyen (TCP pour le site web de commande par exemple).

Jusqu’ a présent le serveur été code en python et attendait les évènements grâce au module python « select » qui me permettait de monitorer a la fois la liaison USB et TCP. Avec ce modèle je ne pouvais cependant pas faire du processing en arrière-plan (déclencher un évènement a une certaine heure par exemple) puisque le « select » est bloquant. J’ai donc choisit de migrer mon code vers Twisted qui correspond exactement a ce que je veux faire « Twisted is an event-driven networking engine written in Python and licensed under the open source »

La documentation est très bien faite est disponible sur le site officiel : http://twistedmatrix.com/trac/

Dans mon cas je veux que twisted monitor 2 « récepteurs » :

  • La liaison USB qui communique avec l’Arduino Leonardo est reçoit donc les messages des capteurs (capteur présence, détection incendie) qui peuvent être trigger à tout moment. Cette liaison est aussi utilisée pour envoyer des ordres aux capteurs/actionneurs du réseau (par exemple pour allumer la lumière)
  • La liaison TCP qui communique avec un client python fortement couple a un script PHP qui reçoit les ordres de la GUI HTML. Cette liaison est utilisée dans les 2 sens (le serveur répond au client qui affiche la réponse sur la page web).

La création de ses 2 « handlers » se fait assez facilement à condition de trouver le bon protocole duquel il faut hériter. Pour la liaison TCP il faut hériter de « twisted.internet.protocol. Protocol » (il existe déjà pas mal d’exemple pour ce cas de figure). Pour la liaison USB avec L’Arduino Leonardo on doit hériter de « twisted.protocols.basic. LineReceiver ». Ce protocole permet de recevoir les messages du port USB à condition qu’il soit bien envoyé comme une ligne complète. Il ne faut pas se tromper avec un autre protocole sinon les données seront tronquées.

Chacun de ses « handlers » a une méthode qui sera appelé lors de la réception de message : « lineReceived » pour l USB et « dataReceived » pour le TCP. Voilà à quoi ressemblent les 2 handlers

class UsbHandler(LineReceiver):
    """protocol handling class for USB """

    def __init__(self,iBrain,iRegisteredDevice):
        self.brain = iBrain
        self.registeredDevices = iRegisteredDevice

    def lineReceived(self, line):
        logging.info("USB Handler created to process : " + str(line))

class TcpHandler(Protocol):
    """protocol handling class for TCP """

    def __init__(self,iBrain,iRegisteredDevice):
        self.brain = iBrain
        self.registeredDevices = iRegisteredDevice

    def dataReceived(self, data):
        logging.info("Tcp Handler created to process : " + str(data))
        if "READ" in str(data):
            logging.info("READ command")
            aRest = aBrain.ReadDeviceStatus2(data,aRegisterDevices)
            logging.info("READ command res " + str(aRest))
        elif str(data) == "STOP": 
            logging.info("STOP command")
            logging.info("Write command")


Ensuite on attache ses 2 protocoles au “reacteur” qui est le Cœur de twisted. Pour le TCP on lui ajoute une surcouche au travers une factory (je ne sais pas pkoi il ne faut pas faire la même chose pour l USB…)

reactor.listenTCP(50007, TcpHandlerFactory(aBrain,aRegisterDevices))
SerialPort(UsbHandler(aBrain,aRegisterDevices), '/dev/ttyACM0', reactor, 9600)


Avec ce design je me retrouve dans la même configuration que celle de mon “framework – Select” fait maison. On va donc ajouter une boucle qui sera capable de gérer la logique background (celle qui ne dépend pas d’un évènement particulier).

Pour cela on créer une méthode (dans notre cas elle se contente d’appeler la vrai méthode de processing) :

def tired_task(iBrain):
    #logging.info("I want to run slowly" + str (datetime.datetime.now()))

et on l ajoute au “reactor” avec une frequence :

lc2 = LoopingCall(tired_task, aBrain)

Dans mon cas cette méthode va mettre a jour les capteurs automatiquement quand nécessaire. Par exemple si le dernier relevé de température date de plus de 5 minutes dans le salon et que la config de ce capteur autorise un relevé toutes les 5 minutes….on le met à jour. Cette méthode va également vérifier l’ensemble des capteurs stable de l’appartement et les updater si nécessaire. Par exemple on va éteindre la lampe de l’entrée si personne n’a été détecté depuis 10 minutes.

    def smartProcessing2(self,iListOfDevice):
        '''Une boucle qui a lieu regulierement pour prendre des decision. Elle va verifier les detecteur et en fonction triger certains evenements.
        Par ex si qq un est detecte dans l entree on decide d allumer la lumiere.
        Cette classe va aussi verifier si certaines autres actions (non lie a la detection de personne) peuvent etre prise.
        Par ex : si on a pas eut de detection de personne depuis un moment ds l entree et que la lumiere est allume....on etient
        Enfin elle va aussi mettre a jour tous les capteurs en fonction de leur refresh rate
        Par exemple si ca fait trop longtemps qu on a pas updater la T alors on la met a jour'''
        logging.info("Begining of a smart loop")

        #Step 1 : Verifier tous les detecteurs (interupteurs stables) pour voir si ils ont ete actives et prendre les actions correspondantes avant de les reset
        #Par exemple si le detecteur de fumee a ete active alors on va envoyer un mail 
        logging.info("Checking all possible event")
        for aOneDevice in iListOfDevice.registeredDevices:
            logging.debug("checking event : " + str(aOneDevice.id))
            if ((aOneDevice.id == 2) and (aOneDevice.currentStatus=="unstable")):
            elif ((aOneDevice.id == 10) and (aOneDevice.currentStatus=="unstable")):
            elif ((aOneDevice.id == 9) and (aOneDevice.currentStatus=="unstable")):

        #Setp 2 : On reset les actions resultantes des detections passe
        #Par exemple si la lumiere de l entree ete ON car qq un avait ete detecte depuis 10 minutes mais qu il y a plus eu de detection depuis 10 min....on eteind
        logging.info("Reseting all previous automatic actions")
        for aOneDevice in iListOfDevice.registeredDevices:
            logging.debug("checking states : " + str(aOneDevice.id))
            if ((aOneDevice.id == 9) and ((iListOfDevice.getDevice(3)).currentStatus=="on") and (datetime.datetime.now() - aOneDevice.LastTMeaureDate > datetime.timedelta (seconds = 600))):
            elif ((aOneDevice.id == 10) and ((iListOfDevice.getDevice(8)).currentStatus=="on") and (datetime.datetime.now() - aOneDevice.LastTMeaureDate > datetime.timedelta (seconds = 180))):

        #Setp 3 : On force un refresh des capteurs periodiques
        logging.info("Force the auto refresh of capteur")
        for aOneDevice in iListOfDevice.registeredDevices:
            logging.debug("checking autoupdate : " + str(aOneDevice.id))
            if ( (aOneDevice.stateCanBeRefresh == True) and (aOneDevice.refreshOngoing == False)and (datetime.datetime.now() - aOneDevice.LastTMeaureDate > datetime.timedelta (minutes = aOneDevice.refreshRatemin) ) ):
                logging.debug("We can refresh : " + str(aOneDevice.id))

Grace à cette nouvelle boucle intelligente la mise à jour des capteurs ne se fait plus dans la crontab avec :

# toutes les heures + 7 min on refresh la emperature entree
 7 * * * * /home/pi/Usb_Arduino_Leonardo/PythonWrapperWebArduinoUsbS.py -o CRONTAB -s 30

La mise à jour des capteurs fait maintenant partie intégrante de l’application.

J’ai également supprimé la base de données qui été interrogée par le site web et remplit par le serveur. Maintenant le site web interroge le serveur pour récupérer l’état T de n’importe quel capteur (en utilisant le format JSON).

case "CMD_READ" :
 $aCommandToExecute = WRAPPER2 . "-o " . getenv(REMOTE_ADDR) . " -s " . $_REQUEST["iCmdToExecute"] . " -t READ";
 $output = array();
 exec($aCommandToExecute, $output);

renvoit :

[“‘{\”py\/object\”: \”Deipara_Objects.CapteurMesure\”, \”InPossibleCmd\”: {\”15\”: \”recoit Nouvelle T\”}, \”physicalLocation\”: \”\”, \”ActionsCommands\”: {\”15\”: \”self.currentStatus=aData\\\\nself.LastTMeaureDate=datetime.datetime.now()\\\\nself.refreshOngoing = False\”}, \”OutPossibleCmd\”: {\”15\”: \”recoit Nouvelle T\”}, \”porteuse\”: \”GATEWAY\”, \”stateCanBeRefresh\”: true, \”type\”: \”CapteurMesure\”, \”LastRefreshDate\”: {\”py\/repr\”: \”datetime\/datetime.datetime(2013, 4, 7, 22, 13, 48, 269822)\”}, \”refreshRatemin\”: 3, \”refreshOngoing\”: false, \”id\”: 15, \”currentStatus\”: 209.0, \”Reset\”: \”\”, \”LastTMeaureDate\”: {\”py\/repr\”: \”datetime\/datetime.datetime(2013, 4, 8, 15, 31, 38, 381700)\”}, \”PossibleStates\”: {}, \”description\”: \”\”}'”]

Au site Web et les informations nécessaires pour la page seront utilisés. Toute la logique est en train de migrer vers le serveur python pour qu’il coordonne l’ensemble des actions. Je clarifie également toutes les interfaces pour utiliser des formats standard (JSON plutôt que des select de colonnes dans une base).

L’ensemble du code est toujours dispo ICI

Création d’un daemon python capable de communiquer avec le port USB et un socket

Actuellement quand je clique sur un lien qui nécessite l’envoie d’un message a un capteur dans l’appartement comme par exemple « actualiser la température » :


Le site web envoie une requête AJAX qui appel un script python pour communiquer par USB depuis le RaspberryPi vers l’Arduino Leonardo. Le script monitor ensuite le port USB pour la réponse de l’Arduino Leonardo pendant 10s pour récupérer la réponse du capteur et la transmettre au site web et à l’utilisateur.


Ce design fonctionne plutôt bien mais bug de temps en temps quand plusieurs personne tentent de rafraichir la température et l’humidité en même temps car une réponse « humidité » peut arriver avant une réponse « température » et les réponses seront inverse. De plus cela ne me permet pas de mettre des capteurs  qui ne nécessite pas d’être appelé pour envoyer une information tel qu’un détecteur de fumée 😉
J’ai donc décidé de revoir le script python pour lui permettre de recevoir des infos de capteur a tout moment sans interaction avec l’utilisateur. Ce daemon doit cependant toujours être capable de recevoir les demandes des utilisateurs du site web…
Le daemon est donc en constante attente soit de message provenant de l’Arduino Leonardo au travers du port USB, soit de message de l’utilisateur (site web) provenant d’un client léger python au travers d’un socket réseau.


Le monitoring simultané du port USB et d’un socket se fait grâce au module python « select » qui est un appel direct à la fonction « select » disponible sous linux. C’est le moyen le plus simple que j’ai trouvé pour faire du multiplexing sur mes 2 entrées (port USB et socket). Le code est donc plutôt simple puisque l’on attend juste que « select » nous donne la main pour tester quelle « entrée » a reçue quelque chose. En fonction de l’entrée on procéder au traitement. Voilà le code avec quelque commentaire pour mieux comprendre 😉

Un gros changement a également eut lieu dans la chaine de réponse puisque le process est maintenant asychrone. Il faut donc inclure plus d’information dans la réponse tel que l’ID du capteur pour savoir a quoi correspond la valeur reçue sur le port USB. Voilà un nouvel exemple de réponse de l’Arduino  Leonardo
La réponse comporte plusieurs champ/valeur tel que l’ID du capteur. Cet ID sera également inséré dans la DB.

Bonus :

Comme il m’arrive de temps en temps de redémarrer mon Raspberry Pi et que le daemon n’est pas lance au démarrage le system pourrai ne plus fonctionner….
J’ai donc légèrement modifie mon client qui est appelé par le serveur web pour détecter si le daemon est bien démarrer sur le RaspberryPi. Si ce n’est pas le cas le Server se charge de démarrer le daemon. On ne pourra donc pas répondre a la requête utilisateur mais au moins les suivantes fonctionneront bien.

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((HOST, PORT))
    print("no server available")
    print("starting server")
    subprocess.call("python PythonWrapperWebArduinoUsbD.py", shell=True)
    #on va rester la a l infini

Exemple :

On lance le script python serveur avec les arguments tel qui l serait lancer par le site web a la réception d’une requête client (les arguments utilisés simulant un Ping avec réponse hardcode ‘456’ d’un capteur.

pi@raspberrypi ~/Usb_Arduino_Leonardo $ python PythonWrapperWebArduinoUsbS.py -s J -i 20 -o TOTO
no server available
starting server

Le client détecte bien qu’il n y a pas de serveur et en lance donc un. Dans un autre terminal on relance le client

Client :

pi@raspberrypi ~/Usb_Arduino_Leonardo $ python PythonWrapperWebArduinoUsbS.py -s J -i 20 -o TOTO
Received 'MSG:J_ID:20_ORIGIN:TOTO'
pi@raspberrypi ~/Usb_Arduino_Leonardo $

Server :

pi@raspberrypi ~/Usb_Arduino_Leonardo $ python PythonWrapperWebArduinoUsbS.py -s J -i 20 -o TOTO
no server available
starting server
handle the server socket
handle all other sockets
input is : MSG:J_ID:20_ORIGIN:TOTO
Writting input to USB port and sending back to sender
Log line : DATE: 2013-03-11 21:02:08.518208 ORIGIN: TOTO CMD: J ID: 20
handle all other sockets
input is :
Closing socket

Le serveur à bien reçu la demande du client et à forwarder la demande a l’Arduino Leonardo par le port USB. Ensuite il a repris sa boucle et quelque secondes plus tard il a reçu la réponse ‘456’ sur le port USB. La réponse a été décodé et stocke dans la DataBase pour ensuite être utilise sur le site Web.