1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
|
# This is part of Kaylee
# -- this code is licensed GPLv3
# Copyright 2015-2016 Clayton G. Hobbs
# Portions Copyright 2013 Jezra
import re
import json
import hashlib
import os
from argparse import ArgumentParser, Namespace
import requests
from gi.repository import GLib
class Config:
"""Keep track of the configuration of Kaylee"""
# Name of the program, for later use
program_name = "kaylee"
# Directories
conf_dir = os.path.join(GLib.get_user_config_dir(), program_name)
cache_dir = os.path.join(GLib.get_user_cache_dir(), program_name)
data_dir = os.path.join(GLib.get_user_data_dir(), program_name)
# Configuration files
opt_file = os.path.join(conf_dir, "options.json")
# Cache files
history_file = os.path.join(cache_dir, program_name + "history")
hash_file = os.path.join(cache_dir, "hash.json")
# Data files
strings_file = os.path.join(data_dir, "sentences.corpus")
lang_file = os.path.join(data_dir, 'lm')
dic_file = os.path.join(data_dir, 'dic')
def __init__(self):
# Ensure necessary directories exist
self._make_dir(self.conf_dir)
self._make_dir(self.cache_dir)
self._make_dir(self.data_dir)
# Set up the argument parser
self.parser = ArgumentParser()
self.parser.add_argument("-i", "--interface", type=str,
dest="interface", action='store',
help="Interface to use (if any). 'g' for GTK or 'gt' for GTK" +
" system tray icon")
self.parser.add_argument("-c", "--continuous",
action="store_true", dest="continuous", default=False,
help="Start interface with 'continuous' listen enabled")
self.parser.add_argument("-p", "--pass-words",
action="store_true", dest="pass_words", default=False,
help="Pass the recognized words as arguments to the shell" +
" command")
self.parser.add_argument("-H", "--history", type=int,
action="store", dest="history",
help="Number of commands to store in history file")
self.parser.add_argument("-m", "--microphone", type=int,
action="store", dest="microphone", default=None,
help="Audio input card to use (if other than system default)")
self.parser.add_argument("--valid-sentence-command", type=str,
dest="valid_sentence_command", action='store',
help="Command to run when a valid sentence is detected")
self.parser.add_argument("--invalid-sentence-command", type=str,
dest="invalid_sentence_command", action='store',
help="Command to run when an invalid sentence is detected")
# Read the configuration file
self._read_options_file()
# Parse command-line arguments, overriding config file as appropriate
self.parser.parse_args(namespace=self.options)
def _make_dir(self, directory):
if not os.path.exists(directory):
os.makedirs(directory)
def _read_options_file(self):
try:
with open(self.opt_file, 'r') as f:
self.options = json.load(f)
self.options = Namespace(**self.options)
except FileNotFoundError:
# Make an empty options namespace
self.options = Namespace()
class Hasher:
"""Keep track of hashes for Kaylee"""
def __init__(self, config):
self.config = config
try:
with open(self.config.hash_file, 'r') as f:
self.hashes = json.load(f)
except IOError:
# No stored hash
self.hashes = {}
def __getitem__(self, hashname):
try:
return self.hashes[hashname]
except (KeyError, TypeError):
return None
def __setitem__(self, hashname, value):
self.hashes[hashname] = value
def get_hash_object(self):
"""Returns an object to compute a new hash"""
return hashlib.sha256()
def store(self):
"""Store the current hashes into a the hash file"""
with open(self.config.hash_file, 'w') as f:
json.dump(self.hashes, f)
class LanguageUpdater:
"""
Handles updating the language using the online lmtool.
This class provides methods to check if the corpus has changed, and to
update the language to match the new corpus using the lmtool. This allows
us to automatically update the language if the corpus has changed, saving
the user from having to do this manually.
"""
def __init__(self, config):
self.config = config
self.hasher = Hasher(config)
def update_language_if_changed(self):
"""Test if the language has changed, and if it has, update it"""
if self.language_has_changed():
self.update_language()
self.save_language_hash()
def language_has_changed(self):
"""Use hashes to test if the language has changed"""
self.stored_hash = self.hasher['language']
# Calculate the hash the language file has right now
hasher = self.hasher.get_hash_object()
with open(self.config.strings_file, 'rb') as sfile:
buf = sfile.read()
hasher.update(buf)
self.new_hash = hasher.hexdigest()
return self.new_hash != self.stored_hash
def update_language(self):
"""Update the language using the online lmtool"""
print('Updating language using online lmtool')
host = 'http://www.speech.cs.cmu.edu'
url = host + '/cgi-bin/tools/lmtool/run'
# Submit the corpus to the lmtool
response_text = ""
with open(self.config.strings_file, 'rb') as corpus:
files = {'corpus': corpus}
values = {'formtype': 'simple'}
r = requests.post(url, files=files, data=values)
response_text = r.text
# Parse response to get URLs of the files we need
path_re = r'.*<title>Index of (.*?)</title>.*'
number_re = r'.*TAR([0-9]*?)\.tgz.*'
for line in response_text.split('\n'):
# If we found the directory, keep it and don't break
if re.search(path_re, line):
path = host + re.sub(path_re, r'\1', line)
# If we found the number, keep it and break
elif re.search(number_re, line):
number = re.sub(number_re, r'\1', line)
break
lm_url = path + '/' + number + '.lm'
dic_url = path + '/' + number + '.dic'
self._download_file(lm_url, self.config.lang_file)
self._download_file(dic_url, self.config.dic_file)
def save_language_hash(self):
self.hasher['language'] = self.new_hash
self.hasher.store()
def _download_file(self, url, path):
r = requests.get(url, stream=True)
if r.status_code == 200:
with open(path, 'wb') as f:
for chunk in r:
f.write(chunk)
|