Computer Vision comme capteur de mouvement pour SmartThings

À l'aide d'un Raspberry Pi 3 et d'une PiCam, ce capteur alimenté par vision par ordinateur détecte les visages et envoie des données de présence sur le réseau local - UPNP à SmartThings.

Je vais commencer par supposer que vous avez un Raspberry Pi 3 avec une caméra fonctionnelle et Open CV installé dessus. Si ce n'est pas le cas, je vous recommande ce tutoriel 🙂

Création d'un gestionnaire d'appareils SmartThings personnalisé

Dans l'IDE SmartThings, nous créons un nouveau gestionnaire d'appareils pour notre capteur de mouvement Computer Vision.

Accédez à la section "Mon gestionnaire d'appareils", cliquez sur "+ Créer un nouveau gestionnaire d'appareils" dans le coin supérieur droit.

Dans ce cas, nous allons le créer à partir du code. Cliquez sur le deuxième onglet « À partir du code », collez le code ci-joint « Device Handler » et cliquez sur « Créer ».

Sur la page suivante, cliquez sur « Publier » pour mettre l'appareil à votre disposition.

Écrire l'application SmartThings

Semblable au gestionnaire de périphérique, nous irons dans la section "Mes SmartApps", cliquez sur "+ Créer de nouvelles SmartApps" dans le coin supérieur droit.

Nous allons également le créer à partir du code. Cliquez sur le deuxième onglet « À partir du code », collez le code joint « SmartApp » et cliquez sur « Créer ».

Dans la page suivante, cliquez sur « Publier ».

Préparer le Raspberry Pi 

Nous devons maintenant ajouter le script python qui obtiendra des images de la caméra, détectera les visages et les signalera à SmartThings.

Première d télécharger et installer imutils et twisted

Si vous n'avez pas déjà installé le package imutils, vous voudrez le récupérer sur GitHub ou l'installer via :

pip install imutils 

Pour tordu :

sudo apt-get install python-twisted-web 

Maintenant que tout est prêt, accédez à  /home/pi et créez un répertoire pour stocker le script

caméra mkdir cameracd 

Créez le fichier de script :

sudo nano ssdpcamera.py 

Collez le code ci-joint " Script Python de la caméra" et enregistrez en appuyant sur « contrôle + x » puis « y » et entrez.

Testez le script en tapant python ssdpcamera.py, vous devriez voir quelque chose comme ceci :

Découverte et appairage du Raspberry Pi

À partir de l'application mobile SmartThings, nous devons accéder à Marketplace dans le coin inférieur droit, cliquer sur l'onglet "SmartApps" et enfin rechercher dans "+ Mes applications" pour "Computer Vision (Connect)"

Assurez-vous que le Raspberry Pi est allumé et que le script python est en cours d'exécution.

La SmartApp lancera le processus de découverte et une fois trouvé, cliquez sur la boîte de dialogue de sélection et sélectionnez l'appareil et cliquez sur "Terminé".

Cela créera l'appareil dans votre compte et commencera à recevoir des mises à jour.

Démarrage automatique

Enfin, si vous souhaitez exécuter le script python automatiquement lorsque vous allumez le Raspberry Pi, vous pouvez éditer /etc/rc.local et ajouter la ligne suivante.

(sleep 10;python /home/pi/camera/ssdpcamera.py)& 

Le () fait que les deux commandes s'exécutent en arrière-plan.

Code

#!/usr/bin/python2.7""" Caméra de vision par ordinateur pour SmartThingsCopyright 2016 Juan Pablo Risso <[email protected]>Dépendances :python-twisted, cv2, pyimagesearchLicensed under la Licence Apache, Version 2.0 (la "Licence") ; vous ne pouvez pas utiliser ce fichier sauf en conformité avec la Licence. Vous pouvez obtenir une copie de la Licence à l'adresse :http://www.apache.org/licenses/LICENSE-2.0Sauf requis par la loi applicable ou convenu par écrit, les logiciels distribués sous la licence sont distribués « EN L'ÉTAT », SANS GARANTIE NI CONDITIONS D'AUCUNE SORTE, expresses ou implicites. Voir la licence pour le langage spécifique régissant les autorisations et les limitations en vertu de la licence. """import argparseimport loggingimport cv2import urllib2import imutilsfrom time import timefrom picamera.array import PiRGBArrayfrom picamera import PiCamerafrom twisted.web import server, resourcefrom twisted.internet import réacteurfrom twisted.internet.defer import successfrom twisted.inter net.protocol import DatagramProtocolfrom twisted.web.client import Agentfrom twisted.web.http_headers import Headersfrom twisted.web.iweb import IBodyProducerfrom twisted.web._newclient import ResponseFailedfrom zope.interface import implementsSSDP_PORT =1900SSDP_ADDR ='239.255.255.250'UUID =eb4 -9220-11e4-96fa-123b93f75cba'SEARCH_RESPONSE ='HTTP/1.1 200 OK\r\nCACHE-CONTROL:max-age=30\r\nEXT:\r\nLOCATION:%s\r\nSERVER:Linux, UPnP/ 1.0, Pi_Garage/1.0\r\nST:%s\r\nUSN:uuid:%s::%s'# initialise la caméra et récupère une référence à la caméra brute# capturecamera =PiCamera()camera.resolution =(640 , 480)camera.framerate =32rawCapture =PiRGBArray(camera, size=(640, 480))auxcount =0# construire le détecteur de visage et permettre à la caméra de chauffer upfd =FaceDetector("cascades/haarcascade_frontalface_default.xml")time.sleep (0.1)def determine_ip_for_host(host):"""Déterminer l'adresse IP locale utilisée pour communiquer avec un hôte particulier"""test_sock =DatagramProtocol()test_sock_listener =réacteur.lis tenUDP(0, test_sock) # pylint:disable=no-membertest_sock.transport.connect(host, 1900)my_ip =test_sock.transport.getHost().hosttest_sock_listener.stopListening()return my_ipclass StringProducer(object):"""Écrit un chaîne en mémoire à une demande Twisted """implements(IBodyProducer)def __init__(self, body):self.body =bodyself.length =len(body)def startProducing(self, consumer):# pylint:disable=invalid- name"""Commencer à produire la chaîne fournie au consommateur spécifié"""consumer.write(self.body)return success(None)def pauseProducing(self):# pylint:disable=invalid-name"""Pause de production - pas d'opération """passdef stopProducing(self):# pylint:disable=invalid-name""" Arrêter de produire - pas d'op"""passclass SSDPServer(DatagramProtocol):"""Recevoir et répondre aux demandes de découverte M-SEARCH du hub SmartThings" ""def __init__(self, interface='', status_port=0, device_target=''):self.interface =interfaceself.device_target =device_targetself.status_port =status_portself.port =réacteur.listenMulticas t(SSDP_PORT, self, listenMultiple=True) # pylint:disable=no-memberself.port.joinGroup(SSDP_ADDR, interface=interface)reactor.addSystemEventTrigger('before', 'shutdown', self.stop) # pylint:disable=no-memberdef datagramReceived(self, data, (host, port)):try:header, _ =data.split(b'\r\n\r\n')[:2]except ValueError:returnlines =header.split ('\r\n')cmd =lines.pop(0).split(' ')lines =[x.replace(':', ':', 1) for x in lines]lines =[x for x en lignes si len(x)> 0]headers =[x.split(':', 1) pour x en lignes]headers =dict([(x[0].lower(), x[1]) pour x in headers])logging.debug('SSDP command %s %s - from %s:%d with headers %s', cmd[0], cmd[1], host, port, headers)search_target =''if ' st' dans les en-têtes:search_target =headers['st']if cmd[0] =='M-SEARCH' et cmd[1] =='*' et search_target dans self.device_target:logging.info('Received %s %s pour %s de %s:%d', cmd[0], cmd[1], search_target, host, port)url ='http://%s:%d/status' % (determine_ip_for_host(host) , self.status_port)réponse =SEARCH_R ESPONSE % (url, search_target, UUID, self.device_target)self.port.write(response, (host, port))else:logging.debug('Commande SSDP ignorée %s %s', cmd[0], cmd[ 1])def stop(self):"""Quitter le groupe de multidiffusion et arrêter d'écouter"""self.port.leaveGroup(SSDP_ADDR, interface=self.interface)self.port.stopListening()class StatusServer(resource.Resource) :"""Serveur HTTP qui sert le statut de la caméra au hub SmartThings"""isLeaf =Truedef __init__(self, device_target, subscription_list, garage_door_status):self.device_target =device_targetself.subscription_list =subscription_listself.garage_door_status =garage_door_status__inresource.Resource( self)def render_SUBSCRIBE(self, request):# pylint:disable=invalid-name"""Gérer les demandes d'abonnement du hub ST - le hub veut être informé des mises à jour de l'état de la porte de garage"""headers =request.getAllHeaders()logging.debug ("SOUSCRIRE : %s", en-têtes) si 'callback' dans les en-têtes :cb_url =headers['callback'][1 :-1] si ce n'est pas cb_url dans self.subscription_list:self.subs cription_list[cb_url] ={}#reactor.stop()logging.info('Abonnement ajouté %s', cb_url)self.subscription_list[cb_url]['expiration'] =time() + 24 * 3600return ""def render_GET( self, request):# pylint:disable=invalid-name"""Gérer les demandes d'interrogation du hub ST"""if request.path =='/status':if self.garage_door_status['last_state'] =='inactive' :cmd ='status-inactive'else:cmd ='status-active'msg ='%suuid:%s::%s' % (cmd, UUID, self.device_target)logging.info("Demande d'interrogation de %s pour %s - renvoyé %s",request.getClientIP(),request.path,cmd)return msgelse:logging.info( "Reçu une fausse demande de %s pour %s",request.getClientIP(),request.path)return ""class MonitorCamera(object):"""Surveille l'état de la caméra, générant des notifications chaque fois que son état change"""def __init__( self, device_target, subscription_list, camera_status):# pylint:disable=too-many-argumentsself.device_target =device_targetself.subscription_list =subscription_listself.camera_stat us =camera_statuscurrent_state ='inactive'reactor.callLater(0, self.check_garage_state, current_state, auxcount) # pylint:disable=no-memberdef check_garage_state(self, current_state, auxcount):self.current_state =current_stateself.auxcount =auxcountcamera.capture( rawCapture, format="bgr", use_video_port=True)# récupère le tableau NumPy brut représentant l'imageframe =rawCapture.array# redimensionne le cadre et convertit-le en grayscaleframe =imutils.resize(frame, width =640)gray =cv2.cvtColor (frame, cv2.COLOR_BGR2GRAY)# détecter les visages dans l'image puis cloner le frame# afin que nous puissions dessiner sur itfaceRects =fd.detect(gray, scaleFactor =1.1, minNeighbors =10,minSize =(30, 30))frameClone =frame.copy()if faceRects !=():auxcount =0 if current_state =='inactive':current_state ='active' logging.info('State changé de %s à %s', self.camera_status['last_state '], current_state)self.camera_status['last_state'] =current_stateself.notify_hubs()else :auxcount =auxcount + 1if auxcount ==60:current_state ='inactive' logging.info('État changé de %s à %s', self.camera_status['last_state'], current_state)self.camera_status['last_state'] =current_stateself.notify_hubs() # boucle sur les cadres de délimitation du visage et dessinez-les pour (fX, fY, fW, fH) dans faceRects:cv2.rectangle(frameClone, (fX, fY), (fX + fW, fY + fH), (0, 255, 0), 2 )# afficher le flux vidéo sur une nouvelle fenêtre graphique#cv2.imshow("Face", videorotate)rawCapture.truncate(0) # Planifier le prochain checkreactor.callLater(0, self.check_garage_state, current_state, auxcount) # pylint:disable=no-memberdef notify_hubs(self):"""Notifier les hubs SmartThings abonnés qu'un changement d'état s'est produit"""if self.camera_status['last_state'] =='inactive':cmd ='status-inactive'else:cmd ='status-active'for subscription in self.subscription_list:if self.subscription_list[subscription]['expiration']> time():logging.info("Notifying hub %s", subscription)msg ='%suuid:%s::%s' % (cmd, UUID, self.device_target)b ody =StringProducer(msg)agent =Agent(reactor)req =agent.request('POST',subscription,Headers({'CONTENT-LENGTH':[len(msg)]}),body)req.addCallback(self. handle_response)req.addErrback(self.handle_error)def handle_response(self, response):# pylint:disable=no-self-use"""Gérer le hub SmartThings en renvoyant un code d'état au POST.C'est en fait inattendu - il ferme la connexion pour POST/PUT sans donner de code de réponse."""if response.code ==202:logging.info("Mise à jour d'état acceptée")else:logging.error("Code de réponse inattendu :%s", réponse. code)def handle_error(self, response):# pylint:disable=no-self-use"""Gérer les erreurs générant l'exécution de NOTIFY. Il ne semble pas y avoir de moyen d'éviter ResponseFailed - le SmartThings Hub ne génère pas de code de réponse approprié pour POST ou PUT, et si NOTIFY est utilisé, il ignore le corps."""if isinstance(response.value, ResponseFailed):logging.debug("La réponse a échoué (attendue)")else:logging.error("Réponse inattendue :%s", réponse)def main():"""Fonction principale pour gérer l'utilisation à partir de la ligne de commande"""arg_proc =argparse .ArgumentParser(description='Fournit l'état actif/inactif de la caméra à un hub SmartThings')arg_proc.add_argument('--httpport', dest='http_port', help='HTTP port number', default=8080, type=int) arg_proc.add_argument('--deviceindex', dest='device_index', help='Device index', default=1, type=int)arg_proc.add_argument('--pollingfreq', dest='polling_freq', help=' Nombre de secondes entre l'interrogation de l'état de la caméra', default=5, type=int)arg_proc.add_argument('--debug', dest='debug', help='Enable debug messages', default=False, action='store_true' )options =arg_proc.parse_args()device_target ='urn:schema s-upnp-org:device:RPi_Computer_Vision:%d' % (options.device_index)log_level =logging.INFOif options.debug:log_level =logging.DEBUGlogging.basicConfig(format='%(asctime)-15s %(levelname)- 8s %(message)s', level=log_level)subscription_list ={}camera_status ={'last_state':'unknown'}logging.info('Initializing camera')# Serveur SSDP pour gérer la découverteSSDPServer(status_port=options.http_port, device_target =device_target)# Site HTTP pour gérer les abonnements/pollingstatus_site =server.Site(StatusServer(device_target, subscription_list, camera_status))reactor.listenTCP(options.http_port, status_site) # pylint:disable=no-memberlogging.info('Initialisation terminée' )# Surveiller l'état de la caméra et envoyer des notifications sur le changement d'étatMonitorCamera(device_target=device_target,subscription_list=subscription_list,camera_status=camera_status) réacteur.run() # pylint:disable=no-memberif __name__ =="__main__":main()Source :ordinateur Vision comme capteur de mouvement pour SmartThings