-
Notifications
You must be signed in to change notification settings - Fork 7
/
fst_threads.py
229 lines (191 loc) · 10.2 KB
/
fst_threads.py
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
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
'''
Free-Snap-Tap V1.1.5
last updated: 241105-2004
'''
from threading import Thread, Event # to play aliases without interfering with keyboard listener
from time import sleep # sleep(0.005) = 5 ms
import pygetwindow as gw # to get name of actual window for focusapp function
alias_thread_logging = []
class Macro_Thread(Thread):
'''
execute macros/alias in its own threads so the delay is not interfering with key evaluation
'''
def __init__(self, key_group, stop_event, alias_name, fst_keyboard):
Thread.__init__(self)
self.daemon = True
self.key_group = key_group
self.stop_event = stop_event
self.alias_name = alias_name
self._fst = fst_keyboard
# def run(self):
# to_be_played_key_events = []
# try:
# # Key_events ans Keys here ...
# if self._fst.arg_manager.DEBUG2:
# print(f"D2: > playing macro: {self.alias_name} :: {self.key_group}")
# for key_event in self.key_group:
# # check all constraints at start!
# constraint_fulfilled, delay_times = self._fst.output_manager.check_constraint_fulfillment(key_event, get_also_delays=True)
# if constraint_fulfilled:
# to_be_played_key_events.append([key_event, delay_times])
# if self._fst.arg_manager.DEBUG2:
# print(f"D2: >> will play '{key_event}' with delays: {delay_times}")
# for key_event, delay_times in to_be_played_key_events:
# # alias_thread_logging.append(f"{time() - starttime:.5f}: Send virtual key: {key_event.key_string}")
# if self.stop_event.is_set():
# self.stop_event.clear()
# break
# else:
# if key_event.is_toggle:
# key_event = self._fst.output_manager.get_next_toggle_state_key_event(key_event)
# # send key event and handles interruption of delay
# self._fst.output_manager.execute_key_event(key_event, delay_times, with_delay=True, stop_event=self.stop_event)
def run(self):
try:
if self._fst.arg_manager.DEBUG2:
print(f"D2: > playing macro: {self.alias_name} :: {self.key_group}")
for key_event in self.key_group:
# check all constraints at start!
constraint_fulfilled, delay_times = self._fst.output_manager.check_constraint_fulfillment(key_event, get_also_delays=True)
if constraint_fulfilled:
if self.stop_event.is_set():
self.stop_event.clear()
break
else:
if key_event.is_toggle:
key_event = self._fst.output_manager.get_next_toggle_state_key_event(key_event)
# send key event and handles interruption of delay
self._fst.output_manager.execute_key_event(key_event, delay_times, with_delay=True, stop_event=self.stop_event)
except Exception as error:
print(error)
alias_thread_logging.append(error)
class Macro_Repeat_Thread(Thread):
'''
repeatatly execute a key event based on a timer
'''
def __init__(self, alias_name, repeat_time, stop_event, fst_keyboard, time_increment=100):
Thread.__init__(self)
self.daemon = True
self.alias_name = alias_name
self.repeat_time = repeat_time
self.stop_event = stop_event
self.time_increment = time_increment
self.number_of_increments = self.repeat_time // time_increment
self._fst = fst_keyboard
self.reset = False
self.macro_stop_event = Event()
def run(self):
print(f"START REPEAT: {self.alias_name} with interval of {self.repeat_time} ms")
while not self.stop_event.is_set():
if self.reset:
self.macro_stop_event = Event()
self.reset = False
#print(f"D4: Repeat: {self.alias_name} reset")
else:
#print(f"D4: Repeat: {self.alias_name} execute")
self._fst.start_macro_playback(self.alias_name, self._fst.key_group_by_alias[self.alias_name], self.macro_stop_event)
for index in range(self.number_of_increments):
if self.stop_event.is_set():
self.macro_stop_event.set()
break
elif self.reset:
break
else:
sleep(self.time_increment / 1000)
# if stopped also stop the macro if it is still running
print(f"STOP REPEAT: {self.alias_name} with interval of {self.repeat_time} ms")
def reset_timer(self):
self.macro_stop_event.set()
self.reset = True
class Focus_Thread(Thread):
'''
Thread for observing the active window and pause toggle the evaluation of key events
can be manually overwritten by Controls on ALT+DEL
'''
def __init__(self, fst_keyboard):#, paused_lock):
Thread.__init__(self)
self.stop = False
self.daemon = True
self._fst = fst_keyboard
self.FOCUS_THREAD_PAUSED = False
# self.paused_lock = paused_lock
def run(self):
last_active_window = ''
shortened_active_window = ''
found_new_focus_app = False
manually_paused = False
while not self.stop:
try:
active_window = gw.getActiveWindow().title
except AttributeError:
pass
if active_window != last_active_window or manually_paused:
if active_window != last_active_window:
last_active_window = active_window
# shorten the active window name
###XXX 241105-1858
# if len(active_window) >= 25:
# reverse = active_window[::-1]
# del1 = reverse.find('–')
# del2 = reverse.find('-')
# del3 = reverse.find('/')
# del4 = reverse.find('\\')
# del_min = 100
# for deliminator in [del1, del2, del3, del4]:
# if deliminator != -1 and deliminator < del_min:
# del_min = deliminator
# reverse_shortened = reverse[:del_min]
# shortened_active_window = reverse_shortened[::-1].strip()
# else:
# shortened_active_window = '--'
if active_window not in ["FST Status Indicator", "FST Crosshair"]:
if not self.FOCUS_THREAD_PAUSED and not self._fst.arg_manager.MANUAL_PAUSED:
if manually_paused:
manually_paused = False
found_new_focus_app = False
for focus_name in self._fst.focus_manager.multi_focus_dict_keys:
# check the shortened window name
# if shortened_active_window.lower().find(focus_name) >= 0:
# found_new_focus_app = True
# self._fst.focus_manager.FOCUS_APP_NAME = focus_name
# active_window = shortened_active_window
# break
# check full window name
if active_window.lower().find(focus_name) >= 0:
found_new_focus_app = True
self._fst.focus_manager.FOCUS_APP_NAME = focus_name
break
if found_new_focus_app:
try:
self._fst.update_args_and_groups(focus_name)
self._fst.cli_menu.update_group_display()
self._fst.cli_menu.display_focus_found(active_window)
self._fst.arg_manager.WIN32_FILTER_PAUSED = False
except Exception as error:
print('--- reloading of groups files failed - not resumed, still paused ---')
print(f" -> aborted reloading due to: {error}")
else:
self._fst.focus_manager.FOCUS_APP_NAME = ''
if self._fst.arg_manager.WIN32_FILTER_PAUSED:
print(f"> Active Window: {active_window}")
else:
self._fst.update_args_and_groups()
self._fst.cli_menu.update_group_display()
self._fst.cli_menu.display_focus_not_found()
###XXX give chance to the controller to release the pressed keys
sleep(0.2)
self._fst.arg_manager.WIN32_FILTER_PAUSED = True
print(f"> Active Window: {active_window}")
else:
manually_paused = True
sleep(0.5)
def pause(self):
# with self.paused_lock:
self.FOCUS_THREAD_PAUSED = True
def restart(self):
if self.FOCUS_THREAD_PAUSED:
# with self.paused_lock:
self.FOCUS_THREAD_PAUSED = False
self._fst.arg_manager.MANUAL_PAUSED = False
def end(self):
self.stop = True