-
Notifications
You must be signed in to change notification settings - Fork 71
/
observer.py
71 lines (51 loc) · 1.77 KB
/
observer.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
import pathlib
import time
from pymedphys._imports import watchdog
import pymedphys._utilities.filesystem
def read_and_trigger_callback(event, callback):
event_source_path = pathlib.Path(event.src_path)
ip = event_source_path.parent.name
with pymedphys._utilities.filesystem.open_no_lock( # pylint: disable = protected-access
event_source_path, "rb"
) as f:
data = f.read()
callback(ip, data)
def create_event_handler(callback):
def on_created(event):
print(f"File created: {event.src_path}")
read_and_trigger_callback(event, callback)
def on_deleted(_):
pass
def on_modified(event):
print(f"File modified: {event.src_path}")
read_and_trigger_callback(event, callback)
def on_moved(_):
pass
event_handler = watchdog.events.PatternMatchingEventHandler(
patterns="*/[0-2][0-9][0-9].txt",
ignore_patterns="",
ignore_directories=True,
case_sensitive=True,
)
event_handler.on_created = on_created
event_handler.on_deleted = on_deleted
event_handler.on_modified = on_modified
event_handler.on_moved = on_moved
return event_handler
def observe_with_callback(directories_to_watch, callback):
event_handler = create_event_handler(callback)
observers = []
for watch_path in directories_to_watch:
observer = watchdog.observers.Observer()
# observer = watchdog.observers.polling.PollingObserver()
observer.schedule(event_handler, watch_path, recursive=True)
observers.append(observer)
for observer in observers:
observer.start()
try:
while True:
time.sleep(1)
finally:
for observer in observers:
observer.stop()
observer.join()