Raspberry Pi にGoogle Assistant SDKを搭載して「OK Google」してみる

app.py

# -*- coding:utf-8 -*-
from bottle import route, run
import irm
import os

@route('/light_on')
def light_on():
    irm.playIR("room_light_on.json")
    irm.playIR("stand_light_on.json")
    print("Light On!")
    return "Light On!"

@route('/light_off')
def light_off():
    irm.playIR("room_light_off.json")
    irm.playIR("stand_light_off.json")
    print("Light Off!")
    return "Light Off!"

run(host='localhost', port=8080, debug=True)

irm.py

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

import sys
import serial
import time
import json
import argparse
import os

here = os.path.abspath(os.path.dirname(__file__))
ir_serial = serial.Serial("/dev/ttyACM0", 9600, timeout = 1)
#ir_serial = serial.Serial("/dev/tty.usbmodem01231", 9600, timeout = 1)

def captureIR(path):
  print ("Capturing IR...")
  ir_serial.write(b"c\r\n")
  time.sleep(3.0)
  msg = ir_serial.readline()
  print (msg)
  if path and not 'Time Out' in msg:
    saveIR(path)

def playIR(path):
  if path and os.path.isfile(path):
    print ("Playing IR with %s ..." % path)
    f = open(path)
    data = json.load(f)
    f.close()
    recNumber = len(data['data'])
    rawX = data['data']

    reply_c = "n,%d\r\n" % recNumber
    ir_serial.write(bytes(reply_c,"UTF-8"))
    ir_serial.readline()

    postScale = data['postscale']
    reply_d = "k,%d\r\n" % postScale
    ir_serial.write(bytes(reply_d,"UTF-8"))
    #time.sleep(1.0)
    msg = ir_serial.readline()
    #print (msg)

    for n in range(recNumber):
        bank = n / 64
        pos = n % 64
        if (pos == 0):
          reply_e = "b,%d\r\n" % bank
          ir_serial.write(bytes(reply_e,"UTF-8"))

        reply_f = "w,%d,%d\n\r" % (pos, rawX[n])
        ir_serial.write(bytes(reply_f,"UTF-8"))

    ir_serial.write(b"p\r\n")
    msg = ir_serial.readline()
    print (msg)
    #ir_serial.close()
  else:
    print ("Playing IR...")
    ir_serial.write(b"p\r\n")
    time.sleep(1.0)
    msg = ir_serial.readline()
    print (msg)

def saveIR(path):
  print ("Saving IR data to %s ..." % path)
  rawX = []
  ir_serial.write(b"I,1\r\n")
  time.sleep(1.0)
  recNumberStr = ir_serial.readline()
  recNumber = int(recNumberStr, 16)

  ir_serial.write(b"I,6\r\n")
  time.sleep(1.0)
  postScaleStr = ir_serial.readline()
  postScale = int(postScaleStr, 10)

  #for n in range(640):
  for n in range(recNumber):
      bank = n / 64
      pos = n % 64
      if (pos == 0):
          reply_a = "b,%d\r\n" % bank
          ir_serial.write(bytes(reply_a,"UTF-8"))

      reply_b = "d,%d\n\r" % pos
      ir_serial.write(bytes(reply_b,"UTF-8"))
      xStr = ir_serial.read(3)
      xData = int(xStr, 16)
      rawX.append(xData)

  data = {'format':'raw', 'freq':38, 'data':rawX, 'postscale':postScale}

  f = open(path, 'w')
  json.dump(data, f)
  f.close()
  print ("Done !")

if __name__ == "__main__":
  # parse options
  parser = argparse.ArgumentParser(description='irMagician CLI utility.')
  parser.add_argument('-c', '--capture', action="store_true", dest="cap", help="capture IR data", default=False)
  parser.add_argument('-p', '--play', action="store_true", dest="play", help="play IR data", default=False)
  parser.add_argument('-s', '--save', action="store_true", dest="save", help="save IR data", default=False)
  parser.add_argument('-f', '--file', action="store", dest="file", help="IR data file (json)", default=False)

  args = parser.parse_args()

  if args.play:
    playIR(args.file)

  if args.save and args.file:
    saveIR(args.file)

  if args.cap:
    captureIR(args.file)

  # release resources
  ir_serial.close()