slack to google home voice notifier (with VoiceText)

vt.py

# encoding: utf-8
# 
# Enjoy google home Japanese voice notifyer (via VoiceText https://cloud.voicetext.jp/webapi)
# Original script: https://code-life.hatenablog.com/entry/google-home-notifier-voicetextapi
# My env:
#   Raspberry Pi Zero WH
#   Google home mini.
#   Slack(make something like #googlehome channel, create Incoming webhook and Outgoing webhook to this channel)
#   python 3.7.0
#
# Start like this.
# global to home server via ngrok https://ngrok.com/ (free account url will change by reboot.)
# $ screen -dmS ngrok ngrok http 5000
# wait ngrok commandline interface. Raspberry Pi Zero will wait about 10-20 sec or more.
# and next, run this script gnu screen or nohup.
# $ screen -AdmS vt.py /usr/bin/env python vt.py
import os
import json
import time
import pychromecast
import requests
import subprocess
import configparser
from threading import Thread
from datetime import datetime
from flask import Flask, request, send_from_directory, jsonify
from voicetext import VoiceText

cfg = configparser.ConfigParser()
#vt.ini (utf-8)
#[settings]
#SLACK_WEBHOOK_URL = "....."
#......
#CACHE_WIPE_INTERVAL = 300
cfg.read("./vt.ini")

SLACK_WEBHOOK_URL   = cfg['settings']['SLACK_WEBHOOK_URL']
VT_APIKEY           = cfg['settings']['VT_APIKEY']
VT_DEFAULT_SPEAKER  = cfg['settings']['VT_DEFAULT_SPEAKER']
FRIENDLY_NAME       = cfg['settings']['FRIENDLY_NAME']
#IP_ADDRESS          = cfg['settings']['IP_ADDRESS']
CACHE_DIR           = cfg['settings']['CACHE_DIR']
CACHE_WIPE_INTERVAL = cfg.getint('settings', 'CACHE_WIPE_INTERVAL')

# get google home device
# if you can't get FRIENDLY_NAME then use IP_ADDRESS
#device = next(x for x in pychromecast.get_chromecasts() if x.host == IP_ADDRESS)
device = next(x for x in pychromecast.get_chromecasts() if x.device.friendly_name == FRIENDLY_NAME)
device.wait()

app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False  # Avoid garbled characters
app.config['port'] = 5000
url = None


def get_ngrok_address():
    global url
    if url is None:
        localhost_url = "http://localhost:4040/api/tunnels"  # Url with tunnel details
        url = requests.get(localhost_url).text  # Get the tunnel information
        j = json.loads(url)
        url = j['tunnels'][0]['public_url']  # Do the parsing of the get
        url = url.replace("http:", "https:")
        print(f"GET ngrok Address: {url}")
        requests.post(SLACK_WEBHOOK_URL, data=json.dumps({'text': f'Google-Home-Notifier Address: {url}'}))
    return url


def play_vt(url, text, speaker, emotion, speed, pitch, volume):
    filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}.wav"
    vt = VoiceText(VT_APIKEY).speaker(speaker)
    if speed:
        vt.speed(int(speed))
    if emotion:
        vt.emotion(emotion)
    if pitch:
        vt.pitch(int(pitch))
    if volume:
        vt.volume(int(volume))

    with open(f"cache/{filename}", 'wb') as f:
        f.write(vt.to_wave(text))
        cast(f"{url}/cache/{filename}", "audio/wav")

    res = subprocess.run("aplay cache/{0}".format(filename).split(), stdout = subprocess.PIPE)


def cast(url, mimetype):
    print(f"wav URL is {url}.")
    mc = device.media_controller
    mc.play_media(url, mimetype)
    mc.block_until_active()


@app.route("/cache/<path:path>")
def send_cache(path):
    return send_from_directory("cache", path)


@app.route("/notifier")
def notifier():
    text = request.args.get("text").replace(" ", "")
    if not text:
        audiourl = request.args.get("url")
        if not audiourl:
            return jsonify({"status": "ERROR", "message": "text is required."})
        cast(audiourl, "audio/mp3")
        return jsonify({"status": "OK", "cast": audiourl})

    speaker = request.args.get("speaker") or VT_DEFAULT_SPEAKER
    emotion = request.args.get("emotion")
    speed = request.args.get("speed")
    pitch = request.args.get("pitch")
    volume = request.args.get("volume")
    print(f"ChromeCast device will speak \"{text}\" in {speaker}. by Voice_Text")
    play_vt(get_ngrok_address(), text, speaker, emotion, speed, pitch, volume)
    return jsonify({"status": "OK",
                    "text": text,
                    "speaker": speaker,
                    "emotion": emotion,
                    "speed": speed,
                    "pitch": pitch,
                    "volume": volume
                    })


@app.route('/slack', methods=['POST'])
def slack():
    text = request.form['text']
    # ignore self information from voice.
    if text.find("Google-Home-Notifier Address:") != -1:
        return ''
    # supress slack reminder word from voice.
    text = text.replace("Reminder:","")
    text = text.replace(" ", "")[0:199]
    if not text:
        return jsonify({"status": "ERROR", "message": "text is required."})
    speaker = VT_DEFAULT_SPEAKER
    emotion = None
    speed = None
    pitch = None
    volume = None
    print(f"ChromeCast device will speak \"{text}\" in {speaker}. by Voice_Text")
    play_vt(get_ngrok_address(), text, speaker, emotion, speed, pitch, volume)
    return ''

#get room temperature via Nature Remo API
#@app.route('/remotp')
#def remotp():
#    #get Nature Remo Temperature and send answer to slack.
#    res = subprocess.run("python /home/pi/bin/RemoTempPostToSlack.py".split(), stdout = subprocess.PIPE)
#    return res.stdout

#get Weather from https://tenki.jp script.
#@app.route('/tenki')
#def tenki():
#    #get tenki.jp XXXXX(xxxxx city) and send answer to slack.
#    res = subprocess.run("python /home/pi/bin/getTenkijp.py".split(), stdout = subprocess.PIPE)
#    return res.stdout

@app.route("/")
def index():
    return "[example] " + get_ngrok_address() + "/notifier?text=テストメッセージ"


def wipe_cache_task():
    print(f"Cache wiping task started. Cache wiping interval is {CACHE_WIPE_INTERVAL} seconds.")
    while True:
        for path in os.listdir(CACHE_DIR):
            os.remove(f"{CACHE_DIR}/{path}")
        time.sleep(CACHE_WIPE_INTERVAL)


if __name__ == "__main__":
    if not os.path.isdir(CACHE_DIR):
        os.makedirs(CACHE_DIR)

    Thread(target=wipe_cache_task).start()
    Thread(target=get_ngrok_address).start()
    app.run()

Google HomeをPythonで自在に喋らせてみた（VoiceTextを使ってるので声も選べます） https://code-life.hatenablog.com/entry/google-home-notifier-voicetextapi
を自宅用に改造したもの