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
|
from datetime import datetime, timedelta
import time
import RPi.GPIO as GPIO
import threading
import events
from config import config
BUTTON_GPIO = 25
REST_TIME = 0.3 # seconds
DIAL_RESET_TIME = 5 # seconds
class DialThread(threading.Thread):
def __init__(self, queue):
threading.Thread.__init__(self, args=(), kwargs=None)
self.queue = queue
self.daemon = True
GPIO.setmode(GPIO.BCM)
GPIO.setup(BUTTON_GPIO, GPIO.IN, pull_up_down=GPIO.PUD_UP)
def run(self):
pressed = True
rest_start = datetime.now()
count = 0
printed = True
while True:
# button is pressed when pin is LOW
if not GPIO.input(BUTTON_GPIO):
if not pressed:
count+=1
pressed = True
printed = False
# button not pressed (or released)
else:
if pressed:
rest_start = datetime.now()
pressed = False
if self.dial_over(rest_start) and not printed:
self.queue.put(count % 10) # wrap 10 to 0
count = 0
printed = True
time.sleep(0.01)
def dial_over(self, rest_start):
# After REST_TIME seconds, assume the rotary dial stopped spinning
return datetime.now() - rest_start > timedelta(seconds=REST_TIME)
class DialManager:
def __init__(self, phone_held, phone_hung_up):
self.sequence = []
self.update = datetime.now()
self._load_sequences(phone_held, phone_hung_up)
def _load_sequences(self, phone_held, phone_hung_up):
def recursive_add(classname, sequence_list, sequences):
key = int(sequence_list[0])
if not sequence_list[1:]:
sequences[key] = eval(classname)(self, phone_held, phone_hung_up)
else:
sequences[key] = {}
recursive_add(classname, sequence_list[1:], sequences[key])
self.sequences = {}
for sequence in config["sequences"]:
print(sequence, config["sequences"][sequence], sep="\t")
recursive_add(config["sequences"][sequence], list(sequence), self.sequences)
def dial(self, number):
if datetime.now() - self.update > timedelta(seconds=DIAL_RESET_TIME):
self.sequence = []
self.sequence.append(number)
match = self.match_sequence()
print(self.sequence, match, sep="\t")
if match is not None:
self.sequence = []
self.update = datetime.now()
return match
def match_sequence(self):
def recursive_find(sequence, sequences):
if not sequence:
return None
digit = sequence[0]
if not digit in sequences:
return None
if isinstance(sequences[digit], dict):
return recursive_find(sequence[1:], sequences[digit])
return sequences[digit]
return recursive_find(self.sequence, self.sequences)
def clear_sequence(self):
self.sequence = []
|
