-
Notifications
You must be signed in to change notification settings - Fork 0
/
watchdog_file.py
65 lines (52 loc) · 1.95 KB
/
watchdog_file.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
import os
import time
from watchdog.events import FileSystemEventHandler
from watchdog.observers import Observer
import CustomException as ce
class WatchDog:
def __init__(self, file_name, m_path):
self._b_lock_modified = True
self._b_lock_moved = True
self.__file_name = file_name
self.__observer = Observer()
self.__observer.schedule(self.FileMonitor(self), m_path, True)
self.__observer.start()
def get_file_name(self) -> str:
return self.__file_name
class FileMonitor(FileSystemEventHandler):
def __init__(self, instance):
self.__instance = instance
def on_modified(self, event):
# print("modified " + os.path.basename(event.src_path))
if (not event.is_directory) and os.path.basename(event.src_path) == self.__instance.get_file_name():
print("modified unlocked")
self.__instance._b_lock_modified = False
pass
def on_moved(self, event):
# print("moved " + os.path.basename(event.dest_path))
if (not event.is_directory) and os.path.basename(event.dest_path) == self.__instance.get_file_name():
print("moved unlocked")
self.__instance._b_lock_moved = False
pass
def on_any_event(self, event):
pass
def on_deleted(self, event):
pass
def on_created(self, event):
pass
def get_interrupt(self):
count = 100 * 30
while self._b_lock_modified or self._b_lock_moved:
count -= 1
time.sleep(0.01)
if not count:
self.__observer.stop()
self.__observer.join()
raise ce.WatchdogTimeoutWarning()
self.__observer.stop()
self.__observer.join()
time.sleep(0.5)
return True
def __del__(self):
self.__observer.stop()
self.__observer.join()