Computer Vision als Bewegungssensor für SmartThings

Mit einem Raspberry Pi 3 und einer PiCam erkennt dieser computergestützte Sensor Gesichter und sendet Anwesenheitsdaten über LAN – UPNP an SmartThings.

Ich gehe zunächst davon aus, dass Sie einen Raspberry Pi 3 mit einer funktionierenden Kamera und darauf installiertem Open CV haben. Wenn nicht, empfehle ich diese Anleitung 🙂

Benutzerdefinierten SmartThings-Gerätehandler erstellen

In der SmartThings-IDE erstellen wir einen neuen Geräte-Handler für unseren Computer Vision-Bewegungssensor.

Gehen Sie zum Abschnitt "Mein Geräte-Handler" und klicken Sie oben rechts auf "+ Neuen Geräte-Handler erstellen".

In diesem Fall erstellen wir es aus Code. Klicken Sie auf den zweiten Tab "Vom Code" und fügen Sie den angehängten Code ein "Geräte-Handler" und klicken Sie auf „Erstellen“.

Klicken Sie auf der nächsten Seite auf „Veröffentlichen“, um Ihnen das Gerät zur Verfügung zu stellen.

Schreiben der SmartThings-App

Ähnlich wie beim Geräte-Handler gehen wir zum Abschnitt "Meine SmartApps", klicken Sie oben rechts auf "+ Neue SmartApps erstellen".

Wir werden es auch aus Code erstellen. Klicken Sie auf den zweiten Tab "Vom Code", fügen Sie den angehängten Code "SmartApp" ein und klicken Sie auf „Erstellen“.

Klicken Sie auf der nächsten Seite auf „Veröffentlichen“.

Vorbereitung des Raspberry Pi 

Jetzt müssen wir das Python-Skript hinzufügen, das Bilder von der Kamera erhält, Gesichter erkennt und an SmartThings berichtet.

Zuerst d Imutils und verdreht herunterladen und installieren

Wenn Sie das imutils-Paket noch nicht installiert haben, sollten Sie es von GitHub herunterladen oder installieren über:

pip install imutils 

Für verdreht:

sudo apt-get install python-twisted-web 

Jetzt, wo alles fertig ist, gehe zu  /home/pi und erstellen Sie ein Verzeichnis zum Speichern des Skripts

mkdir cameracd camera 

Erstellen Sie die Skriptdatei:

sudo nano ssdpcamera.py 

Fügen Sie den angehängten Code ein " Kamera-Python-Skript” und speichern Sie, indem Sie "Strg + x" und dann "y" drücken und eingeben.

Testen Sie das Skript, indem Sie python ssdpcamera.py eingeben. Sie sollten etwa Folgendes sehen:

Entdecken und Koppeln des Raspberry Pi

In der mobilen SmartThings-App sollten wir unten rechts zum Marktplatz gehen, auf den Tab "SmartApps" klicken und schließlich in "+ Meine Apps" nach "Computer Vision (Connect)" suchen

Stellen Sie sicher, dass der Raspberry Pi eingeschaltet ist und das Python-Skript ausgeführt wird.

Die SmartApp startet den Erkennungsprozess und sobald sie gefunden wurde, klicken Sie auf den Auswahldialog und wählen Sie das Gerät aus und klicken Sie auf „Fertig“.

Dadurch wird das Gerät in Ihrem Konto erstellt und der Empfang von Updates beginnt.

Automatischer Start

Wenn Sie schließlich das Python-Skript beim Einschalten des Raspberry Pi automatisch ausführen möchten, können Sie /etc/rc.local bearbeiten und die folgende Zeile hinzufügen.

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

Das () lässt beide Befehle im Hintergrund laufen.

Code

#!/usr/bin/python2.7""" Computer Vision Camera for SmartThingsCopyright 2016 Juan Pablo Risso <[email protected]>Abhängigkeiten:python-twisted, cv2, pyimagesearchLizenziert unter die Apache-Lizenz, Version 2.0 (die "Lizenz"); Sie dürfen diese Datei nur in Übereinstimmung mit der Lizenz verwenden. Eine Kopie der Lizenz erhalten Sie unter:http://www.apache.org/licenses/LICENSE-2.0Unless gesetzlich vorgeschrieben oder schriftlich vereinbart wird, wird Software, die unter der Lizenz vertrieben wird, "WIE BESEHEN" OHNE GEWÄHRLEISTUNGEN ODER BEDINGUNGEN JEGLICHER ART, weder ausdrücklich noch stillschweigend, vertrieben. Siehe die Lizenz für die spezifische Sprache, die die Genehmigungen und Beschränkungen im Rahmen der Lizenz regelt. """import argparseimport loggingimport cv2import urllib2import imutilsfrom time import timefrom picamera.array import PiRGBArrayfrom picamera import PiCamerafrom twisted.web import server, resourcefrom twisted.internet import reactorfrom 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.25.250' -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'# initialisiert die Kamera und holt eine Referenz auf die Raw-Kamera# capturecamera =PiCamera()camera.resolution =(640 , 480)camera.framerate =32rawCapture =PiRGBArray(camera, size=(640, 480))auxcount =0# Konstruieren Sie den Gesichtsdetektor und lassen Sie die Kamera aufwärmenfd =FaceDetector("cascades/haarcascade_frontalface_default.xml")time.sleep (0.1)def define_ip_for_host(host):"""Lokale IP-Adresse bestimmen, die verwendet wird, um mit einem bestimmten Host zu kommunizieren"""test_sock =DatagramProtocol()test_sock_listener =Reaktor.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):"""Schreibt an In-Memory-String an eine Twisted-Anfrage"""implements(IBodyProducer)def __init__(self, body):self.body =bodyself.length =len(body)def startProducing(self, Consumer):# pylint:disable=invalid- name"""Starte die Produktion des übergebenen Strings für den angegebenen Consumer"""consumer.write(self.body)return success(None)def pauseProducing(self):# pylint:disable=ungültiger-name"""Produktion unterbrechen - keine Op """passdef stopProducing(self):# pylint:disable=invalid-name""" Stoppen der Produktion - keine Operation"""passclass SSDPServer(DatagramProtocol):"""Empfang und Antwort auf M-SEARCH-Erkennungsanfragen vom SmartThings-Hub" ""def __init__(self, interface='', status_port=0, device_target=''):self.interface =interfaceself.device_target =device_targetself.status_port =status_portself.port =Reaktor.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]außer ValueError:returnlines =header.split ('\r\n')cmd =lines.pop(0).split(' ')lines =[x.replace(':', ':', 1) für x in Zeilen]lines =[x für x in Zeilen if len(x)> 0]headers =[x.split(':', 1) für x in Zeilen]headers =dict([(x[0].lower(), x[1]) für x in headers])logging.debug('SSDP-Befehl %s %s - from %s:%d with headers %s', cmd[0], cmd[1], host, port, headers)search_target =''if ' st' in headers:search_target =headers['st']if cmd[0] =='M-SEARCH' und cmd[1] =='*' und search_target in self.device_target:logging.info('Received %s %s für %s von %s:%d', cmd[0], cmd[1], search_target, host, port)url ='http://%s:%d/status' % (determine_ip_for_host(host) , self.status_port)Antwort =SEARCH_R ESPONSE % (url, search_target, UUID, self.device_target)self.port.write(response, (host, port))else:logging.debug('Ignorierter SSDP-Befehl %s %s', cmd[0], cmd[ 1])def stop(self):"""Multicast-Gruppe verlassen und aufhören"""self.port.leaveGroup(SSDP_ADDR, interface=self.interface)self.port.stopListening()class StatusServer(resource.Resource):"""HTTP-Server, der den Status der Kamera an den SmartThings-Hub übermittelt"""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_statusresource.Resource.Resource. self)def render_SUBSCRIBE(self, request):# pylint:disable=invalid-name"""Subscribe-Anfragen vom ST-Hub bearbeiten - Hub möchte über Statusaktualisierungen des Garagentors benachrichtigt werden"""headers =request.getAllHeaders()logging.debug ("ABONNIEREN:%s", headers)if 'callback' in headers:cb_url =headers['callback'][1:-1]if nicht cb_url in self.subscription_list:self.subs cription_list[cb_url] ={}#reactor.stop()logging.info('Abo hinzugefügt %s', cb_url)self.subscription_list[cb_url]['expiration'] =time() + 24 * 3600return ""def render_GET( self, request):# pylint:disable=invalid-name"""Abrufanfragen vom ST-Hub verarbeiten"""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("Abfrageanfrage von %s für %s - %s zurückgegeben",request.getClientIP(),request.path,cmd)return msgelse:logging.info( "Gefälschte Anfrage von %s für %s erhalten",request.getClientIP(),request.path)return ""class MonitorCamera(object):"""Überwacht den Kamerastatus und generiert Benachrichtigungen bei jeder Statusänderung"""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)# nimm das rohe NumPy-Array, das den Bildrahmen repräsentiert =rawCapture.array# ändere die Größe des Rahmens und wandele ihn in Grayscaleframe um =imutils.resize(frame, width =640)gray =cv2.cvtColor (frame, cv2.COLOR_BGR2GRAY)# Gesichter im Bild erkennen und dann den Frame klonen# damit wir darauf zeichnen könnenfaceRects =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 geändert von %s zu %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('State change from %s to %s', self.camera_status['last_state'], current_state)self.camera_status['last_state'] =current_stateself.notify_hubs() # loop over die Face Bounding Boxes und zeichnen sie für (fX, fY, fW, fH) in faceRects:cv2.rectangle(frameClone, (fX, fY), (fX + fW, fY + fH), (0, 255, 0), 2 )# Video-Feed in einem neuen GUI-Fenster anzeigen#cv2.imshow("Face", videorotate)rawCapture.truncate(0) # Nächstes planen checkreactor.callLater(0, self.check_garage_state, current_state, auxcount) # pylint:disable=no-memberdef notification_hubs(self):"""Benachrichtigen Sie die abonnierten SmartThings-Hubs, dass eine Statusänderung aufgetreten ist"""if self.camera_status['last_state'] =='inactive':cmd ='status-inactive'else:cmd ='status-active'für Abonnement in self.subscription_list:if self.subscription_list[subscription]['expiration']> time():logging.info("Hub %s benachrichtigen", Abonnement)msg ='%suuid:%s::%s' % (cmd, UUID, self.device_target)body =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"""Handhabe den SmartThings-Hub, der einen Statuscode an den POST zurückgibt. Dies ist eigentlich unerwartet - es wird normalerweise geschlossen die Verbindung für POST/PUT ohne Angabe eines Antwortcodes."""if response.code ==202:logging.info("Statusupdate akzeptiert")else:logging.error("Unerwarteter Antwortcode:%s", response.code )def handle_error(self, response):# pylint:disable=no-self-use"""Behandeln Sie Fehler, die bei der Ausführung von NOTIFY generiert werden. Es scheint keine Möglichkeit zu geben, ResponseFailed zu vermeiden - der SmartThings Hub generiert keinen richtigen Antwortcode für POST oder PUT, und wenn NOTIFY verwendet wird, ignoriert es den Text."""if isinstance(response.value, ResponseFailed):Logging.debug("Antwort fehlgeschlagen (erwartet)")else:logging.error("Unerwartete Antwort:%s", Antwort)def main():"""Hauptfunktion für die Verwendung über die Befehlszeile"""arg_proc =argparse .ArgumentParser(description='Stellt einem SmartThings-Hub den Kamerastatus aktiv/inaktiv zur Verfügung')arg_proc.add_argument('--httpport', dest='http_port', help='HTTP-Portnummer', default=8080, type=int) arg_proc.add_argument('--deviceindex', dest='device_index', help='Geräteindex', default=1, type=int)arg_proc.add_argument('--pollingfreq', dest='polling_freq', help=' Anzahl der Sekunden zwischen dem Abfragen des Kamerastatus', default=5, type=int)arg_proc.add_argument('--debug', dest='debug', help='Debug-Nachrichten aktivieren', 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')# SSDP-Server zur Verarbeitung von DiscoverySSDPServer(status_port=options.http_port, device_target =device_target)# HTTP-Site zur Verarbeitung von Subscriptions/pollingstatus_site =server.Site(StatusServer(device_target,subscription_list, camera_status))reactor.listenTCP(options.http_port, status_site) # pylint:disable=no-memberlogging.info('Initialisierung abgeschlossen' )# Kamerastatus überwachen und Benachrichtigungen bei Statusänderung sendenMonitorCamera(device_target=device_target,subscription_list=subscription_list,camera_status=camera_status)reaktor.run() # pylint:disable=no-memberif __name__ =="__main__":main()Source:Computer Vision als Bewegungssensor für SmartThings