SmartThings용 모션 센서로서의 컴퓨터 비전

Raspberry Pi 3 및 PiCam을 사용하여 이 컴퓨터 비전 구동 센서는 얼굴을 감지하고 LAN을 통해 존재 데이터를 전송합니다(UPNP).

먼저 작동하는 카메라와 Open CV가 설치된 Raspberry Pi 3가 있다고 가정합니다. 그렇지 않다면 이 가이드를 추천합니다. 🙂

맞춤 SmartThings 기기 핸들러 만들기

SmartThings IDE에서 Computer Vision 모션 센서용 새 기기 핸들러를 만듭니다.

'내 기기 핸들러' 섹션으로 이동하여 오른쪽 상단에 있는 '+ 새 기기 핸들러 만들기'를 클릭합니다.

이 경우 코드에서 생성합니다. 두 번째 탭인 '코드에서'를 클릭하고 첨부된 코드를 붙여넣습니다. '기기 핸들러' "만들기"를 클릭합니다.

다음 페이지에서 "게시"를 클릭하여 장치를 사용할 수 있도록 합니다.

SmartThings 앱 작성

기기 핸들러와 유사하게 '내 SmartApps' 섹션으로 이동하여 오른쪽 상단에서 '+ 새 SmartApp 만들기'를 클릭합니다.

우리는 또한 그것을 코드에서 만들 것입니다. 두 번째 탭인 "코드에서"를 클릭하고 첨부된 코드 "SmartApp"을 붙여넣습니다. "만들기"를 클릭합니다.

다음 페이지에서 "게시"를 클릭하십시오.

라즈베리 파이 준비하기

이제 카메라에서 이미지를 가져와서 얼굴을 감지하고 SmartThings에 보고하는 Python 스크립트를 추가해야 합니다.

첫 번째 d imutils 및 twisted 다운로드 및 설치

imutils 패키지가 아직 설치되어 있지 않다면 GitHub에서 가져오거나 다음을 통해 설치하는 것이 좋습니다.

pip install imutils 

꼬인 경우:

sudo apt-get 설치 python-twisted-web 

이제 모든 것이 준비되었으므로 /home/pi로 이동합니다. 스크립트를 저장할 디렉토리를 만듭니다.

mkdir 카메라CD 카메라 

스크립트 파일 생성:

sudo nano ssdpcamera.py 

첨부된 코드를 붙여넣습니다. “ 카메라 Python 스크립트' "control + x"를 누른 다음 "y"를 눌러 저장하고 Enter 키를 누릅니다.

python ssdpcamera.py를 입력하여 스크립트를 테스트하면 다음과 같이 표시되어야 합니다.

라즈베리 파이 검색 및 페어링

SmartThings 모바일 앱에서 오른쪽 하단의 Marketplace로 이동하여 'SmartApps' 탭을 클릭하고 마지막으로 '+ My Apps'에서 'Computer Vision(Connect)'을 찾습니다.

Raspberry Pi가 켜져 있고 python 스크립트가 실행 중인지 확인하십시오.

SmartApp이 검색 프로세스를 시작하고 발견되면 선택 대화 상자를 클릭하고 장치를 선택하고 "완료"를 클릭합니다.

이렇게 하면 계정에 장치가 생성되고 업데이트 수신이 시작됩니다.

자동 시작

마지막으로 Raspberry Pi를 켤 때 Python 스크립트를 자동으로 실행하려면 /etc/rc.local을 편집하고 다음 줄을 추가하면 됩니다.

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

() 두 명령을 모두 백그라운드에서 실행합니다.

코드

#!/usr/bin/python2.7""" Computer Vision Camera for SmartThingsCopyright 2016 Juan Pablo Risso <[email protected]>종속성:python-twisted, cv2, pyimagesearchLicensed under Apache 라이선스, 버전 2.0("라이선스"), 라이선스를 준수하지 않는 한 이 파일을 사용할 수 없습니다. http://www.apache.org/licenses/LICENSE-2.0 관련 법률에서 요구하거나 서면으로 동의한 경우, 라이선스에 따라 배포된 소프트웨어는 명시적이든 묵시적이든 어떠한 종류의 보증이나 조건도 없이 "있는 그대로" 배포됩니다. 라이선스에 따른 권한 및 제한 사항을 규율하는 특정 언어는 라이선스를 참조하십시오. """import argparseimport loggingimport cv2import urllib2imutils에서 시간 가져오기 picamera.array에서 가져오기 PiRGBArray에서 picamera에서 PiCamera 가져오기 twisted.web 가져오기 서버, twisted.internet에서 원자로 가져오기 twisted.internet에서 가져오기 성공 지연 twisted.inter에서 가져오기 net.protocol 가져오기 DatagramProtocolfrom twisted.web.client 가져오기 Agentfrom twisted.web.http_headers 가져오기 Headersfrom twisted.web.iweb 가져오기 IbodyProducerfrom twisted.web._newclient 가져오기 ResponseFailedfrom zope.interface 가져오기 구현SSDP_PORT =1900SSDP_ADDR ='239.5.2250'239.55.255.2 -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'# 카메라를 초기화하고 원시 카메라에 대한 참조를 가져옵니다# capturecamera =PiCamera()camera.resolution =(640 , 480)camera.framerate =32rawCapture =PiRGBArray(camera, size=(640, 480))auxcount =0# 얼굴 감지기를 구성하고 카메라가 워밍업하도록 합니다.fd =FaceDetector("cascades/haarcascade_frontalface_default.xml")time.sleep (0.1)def 결정_ip_for_host(호스트):"""특정 호스트와 통신하는 데 사용되는 로컬 IP 주소 확인"""test_sock =DatagramProtocol()test_sock_listener =reactor.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):"""쓰기 꼬인 요청에 대한 메모리 내 문자열"""implements(IBodyProducer)def __init__(self, body):self.body =bodyself.length =len(body)def startProducing(self, consumer):# pylint:disable=invalid- name"""지정된 소비자에게 제공된 문자열 생성 시작"""consumer.write(self.body)return success(None)def pauseProducing(self):# pylint:disable=invalid-name"""생산 일시 중지 - 작업 없음 """passdef stopProducing(self):# pylint:disable=invalid-name""" 생산 중단 - 작업 없음"""passclass SSDPServer(DatagramProtocol):"""SmartThings 허브에서 M-SEARCH 검색 요청 수신 및 응답" ""def __init__(self, interface='', status_port=0, device_target=''):self.interface =interfaceself.device_target =device_targetself.status_port =status_portself.port =reactor.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 in lines if len(x)> 0]headers =[x.split(':', 1) for x in lines]headers =dict([(x[0].lower(), x[1]) for x in headers])logging.debug('SSDP 명령 %s %s - 헤더 %s가 있는 %s:%d, cmd[0], cmd[1], 호스트, 포트, 헤더)search_target =''if' st' in headers:search_target =headers['st']if cmd[0] =='M-SEARCH' and cmd[1] =='*' and search_target in self.device_target:logging.info('수신 %s %s의 %s:%d', cmd[0], cmd[1], search_target, host, port)url ='http://%s:%d/status' % (determine_ip_for_host(host) , self.status_port) 응답 =SEARCH_R ESPONSE % (url, search_target, UUID, self.device_target)self.port.write(response, (host, port))else:logging.debug('무시된 SSDP 명령 %s %s', cmd[0], cmd[ 1])def stop(self):"""멀티캐스트 그룹을 나가서 수신 중지"""self.port.leaveGroup(SSDP_ADDR, interface=self.interface)self.port.stopListening()class StatusServer(resource.Resource):"""카메라 상태를 SmartThings 허브에 제공하는 HTTP 서버"""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.__in self)def render_SUBSCRIBE(self, request):# pylint:disable=invalid-name"""ST 허브의 구독 요청 처리 - 허브는 차고 문 상태 업데이트 알림을 받기를 원합니다."""headers =request.getAllHeaders()logging.debug ("SUBSCRIBE:%s", headers) headers:cb_url =headers['callback'][1:-1]self.subscription_list:self.subs에서 cb_url이 아닌 경우 '콜백' cription_list[cb_url] ={}#reactor.stop()logging.info('추가된 구독 %s', cb_url)self.subscription_list[cb_url]['만료'] =time() + 24 * 3600return ""def render_GET( self, request):# pylint:disable=invalid-name"""ST 허브의 폴링 요청 처리"""if request.path =='/status':if self.garage_door_status['last_state'] =='inactive' :cmd ='상태-비활성'else:cmd ='상태-활성'msg ='%suuid:%s::%s' % (cmd, UUID, self.device_target)logging.info("%s에서 %s에 대한 폴링 요청 - %s 반환",request.getClientIP(),request.path,cmd)return msgelse:logging.info( "%s에서 %s에 대한 가짜 요청을 받았습니다.",request.getClientIP(),request.path)return ""class MonitorCamera(object):"""카메라 상태를 모니터링하고 상태가 변경될 때마다 알림을 생성합니다."""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)# 이미지를 나타내는 원시 NumPy 배열을 가져옵니다. =rawCapture.array# 프레임 크기를 조정하고 이를 grayscaleframe =imutils.resize(frame, width =640)gray =cv2.cvtColor로 변환합니다. (frame, cv2.COLOR_BGR2GRAY)# 이미지에서 얼굴을 감지한 다음 프레임을 복제하여 # 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('상태가 %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 ='비활성' logging.info('상태가 %s에서 %s(으)로 변경됨', self.camera_status['last_state'], current_state)self.camera_status['last_state'] =current_stateself.notify_hubs() # 루프 오버 면 경계 상자를 그리고 faceRects:cv2.rectangle(frameClone, (fX, fY), (fX + fW, fY + fH), (0, 255, 0), 2의 (fX, fY, fW, fH)에 대해 그립니다. )# 새 GUI 창에 비디오 피드 표시#cv2.imshow("Face", videorotate)rawCapture.truncate(0) # 다음 예약 checkreactor.callLater(0, self.check_garage_state, current_state, auxcount) # pylint:disable=no-memberdef notify_hubs(self):"""상태 변경이 발생했음을 구독 SmartThings 허브에 알립니다.""" if self.camera_status['last_state'] =='inactive':cmd ='status-inactive'else:cmd =self.subscription_list:if self.subscription_list[subscription]['expiration']> time():logging.info("알림 허브 %s", 구독)msg ='%suuid:%s::%s' %(cmd, UUID, self.device_target) 본문 =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"""상태 코드를 POST로 반환하는 SmartThings 허브를 처리합니다. 이것은 실제로 예상치 못한 일입니다. 일반적으로 닫힙니다. 응답 코드를 제공하지 않고 POST/PUT에 대한 연결입니다."""if response.code ==202:logging.info("상태 업데이트가 허용됨")else:logging.error("예기치 않은 응답 코드:%s", response.code )def handle_error(self, response):# pylint:disable=no-self-use"""NOTIFY를 수행하는 동안 발생하는 오류를 처리합니다. ResponseFailed를 피하는 방법은 없는 것 같습니다. SmartThings Hub는 POST 또는 PUT에 대한 적절한 응답 코드를 생성하지 않으며,NOTIFY를 사용하면 본문을 무시합니다."""if isinstance(response.value, ResponseFailed):logging.debug("응답 실패(예상)")else:logging.error("예기치 않은 응답:%s", 응답)def main():"""명령줄에서 사용을 처리하는 주요 기능"""arg_proc =argparse .ArgumentParser(description='SmartThings 허브에 카메라 활성/비활성 상태 제공')arg_proc.add_argument('--httpport', dest='http_port', help='HTTP 포트 번호', default=8080, type=int) arg_proc.add_argument('--deviceindex', dest='device_index', help='디바이스 인덱스', default=1, type=int)arg_proc.add_argument('--pollingfreq', dest='polling_freq', help=' 카메라 상태 폴링 사이의 시간(초)', default=5, type=int)arg_proc.add_argument('--debug', dest='debug', help='디버그 메시지 활성화', default=False, action='store_true' ) 옵션 =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('카메라 초기화 중')# 검색을 처리할 SSDP 서버SSDPServer(status_port=options.http_port, device_target =device_target)# 구독을 처리할 HTTP 사이트/pollingstatus_site =server.Site(StatusServer(device_target, subscription_list, camera_status))reactor.listenTCP(options.http_port, status_site) # pylint:disable=no-memberlogging.info('초기화 완료' )# 카메라 상태를 모니터링하고 상태에 대한 알림을 보냅니다. changeMonitorCamera(device_target=device_target,subscription_list=subscription_list,camera_status=camera_status) reactor.run() # pylint:disable=no-memberif __name__ =="__main__":main()Source:Computer SmartThings용 모션 센서로서의 비전