Coverage for src / lstautorta / files_callback.py: 22%
77 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-10 11:56 +0000
1import logging
2from collections.abc import Callable
3from multiprocessing import Queue as MPQueue
4from multiprocessing.synchronize import Event as SyncEvent # for type hints
5from pathlib import Path
6from queue import Empty, Queue
7from time import sleep
8from typing import Any, NamedTuple
10from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
11from watchdog.observers import Observer
12from watchdog.utils.patterns import match_any_paths
14from lstautorta.utils.logging import init_logging
15from lstautorta.utils.queue import process_all_items
18class QueueingFilePatternMatchingEventHandler(PatternMatchingEventHandler):
19 """Handler that queues (blocking) the file paths of created files that matched the patterns
21 Parameters
22 ----------
23 file_queue : Queue
24 The queue to use to queue the files created.
25 """
27 def __init__(self, file_queue: Queue, **kwargs):
28 super().__init__(**kwargs)
29 self._file_queue = file_queue
31 def on_created(self, event: FileSystemEvent) -> None:
32 super().on_created(event)
33 # add file in queue, blocking until put can happen (should never block if queue has infinite size)
34 self._file_queue.put(Path(event.src_path))
37class DirMonitoringInfo(NamedTuple):
38 """Information required by `dir_monitoring` to start monitoring a directory. Typically received from a multiprocessing Queue."""
40 data_dir: Path
41 process_file_fct_extra_kwargs: dict[str, Any]
42 on_dir_change_fct_extra_kwargs: dict[str, Any]
45def dir_monitoring(
46 log_level: str,
47 log_filename: str,
48 log_format_title: str,
49 info_queue: MPQueue,
50 stop_event: SyncEvent,
51 process_file_fct: Callable,
52 on_dir_change_fct: Callable,
53 file_patterns: list[str] = None,
54 file_ignore_patterns: list[str] = None,
55 monitoring_ignore_directories=True,
56 pattern_case_sensitive: bool = True,
57 query_interval_s: float = 1.0,
58 monitoring_info_received_event: SyncEvent | None = None,
59):
60 """Monitor the last directory pushed on `info_queue` and applies `process_file_fct` on created files that match the patterns, until stop event is set. Also executes `on_dir_change_fct` when the monitoring directoy is changed.
62 Notes
63 -----
64 Files already present in the monitored directory when monitoring starts will be processed as well.
66 Parameters
67 ----------
68 log_level : str
69 Log level to use for the logger of this function
70 log_filename : str
71 Path to the log files to write logs to.
72 log_format_title : str
73 "Title" to use in the log files lines.
74 info_queue : MPQueue[DirMonitoringInfo]
75 Multiprocessing queue to use to retrieve the directory to monitor and relevant informations to do it.
76 stop_event : SyncEvent
77 When this event is set, the monitoring will stop and the function will return.
78 process_file_fct : Callable
79 Function to apply to the created files (and already present files) in the monitored directory, passing the file path
80 as argument and extra kwargs based on information in `info_queue`
81 on_dir_change_fct : Callable
82 Function to execute when the directory to monitor is changed (a new one is available in `info_queue`) The monitored directory
83 path (before change) is passed to the function as argument, with additional kwargs based on information in `info_queue`.
84 file_patterns : List[str], optional
85 List of patterns to filter the created files. See watchdogs PatternMatchingEventHandler documentation. By default None
86 file_ignore_patterns : List[str], optional
87 List of patterns to ignore the created files. See watchdogs PatternMatchingEventHandler documentation. By default None
88 monitoring_ignore_directories : bool, optional
89 Whether the monitoring should ignore the created subdirectories, by default True
90 pattern_case_sensitive : bool, optional
91 If True, the patterns applied to filter or ignore files will be case insensitive, by default True
92 query_interval_s : float, optional
93 Amount of time in seconds to wait between 2 processing rounds of created files, by default 1.0
94 """
95 # init logging in a separate file for this monitoring process
96 # Note: this requires that the process is started with the "spawn" method, so it does not inherit the parent
97 # process file handles of logging !
98 init_logging(log_level=log_level, log_filename=log_filename, format_title=log_format_title)
100 if not isinstance(file_patterns, (list, None)):
101 raise ValueError(f"File patterns must be a list of string or None, got {file_patterns}")
102 if not isinstance(file_ignore_patterns, (list, None)):
103 raise ValueError(f"Ignore pattern must be a list of string or None, got {file_ignore_patterns}")
105 # queue to store paths of created files in the folder (that match pattern)
106 # with infinite size so we can use put/get non-blocking
107 new_file_queue: Queue[Path] = Queue(maxsize=-1)
109 # Once a monitoring is started, it continues until another directory to monitor is put in the info_queue, or the
110 # stop event is set. However, when the process is started, there may not yet be a DirMonitoringInfo in the queue yet.
111 monitoring_started = False
113 # This loop code order doesn't reflects the chronology of the code execution (in particular variable instantiation)
114 # before the 2nd DirMonitoringInfo:
115 # - Before the first DirMonitoringInfo is retrieved from the queue, we will simply do the "except" clause, which will
116 # do nothing because monitoring_started is False.
117 # - When the first DirMonitoringInfo is retrieved from the queue, we execute the "else" clause, but since
118 # monitoring_started is still False, we don't do the observer stop + final files processing. We directly go to the
119 # processing info execution and create the processing_info and all variables there.
120 # - When retrieving the 2nd and more DirMonitoringInfo from the queue, we execute the entire try + else clauses, and all
121 # required variables are defined.
122 while not stop_event.is_set():
123 try:
124 # Check if a new processing info was put in the queue
125 # Otherwise Empty is raised
126 new_processing_info = info_queue.get(block=False)
127 except Empty:
128 # no new processing info: simply process any queued files if we are monitoring something, then sleep
129 if monitoring_started:
130 nb_processed = process_all_items(
131 new_file_queue,
132 processed_files_set,
133 process_file_fct,
134 processing_info.process_file_fct_extra_kwargs,
135 )
136 if nb_processed > 0:
137 logging.debug(f"Processed {nb_processed} files.")
138 sleep(query_interval_s)
139 else:
140 if monitoring_info_received_event is not None:
141 monitoring_info_received_event.set()
142 # There is a new processing info:
143 # If monitoring:
144 # - stop monitoring
145 # - processed queued files
146 # - run on_dir_change function
147 # In any case:
148 # - start new monitoring
149 # - re-set all monitoring variables (processed files set)
150 # - queue all files already present in folder "before" monitoring started (can be some overlap, handled by the processed files set)
151 logging.info(f"New monitoring dir {new_processing_info.data_dir}")
152 logging.info(f"New process_file_fct_extra_kwargs {new_processing_info.process_file_fct_extra_kwargs}")
153 logging.info(f"New on_dir_change_fct_extra_kwargs {new_processing_info.on_dir_change_fct_extra_kwargs}")
154 if monitoring_started:
155 logging.info(f"Stopping monitoring of {processing_info.data_dir}")
156 observer.stop()
157 observer.join()
158 logging.info(f"Processing remaining items of {processing_info.data_dir}")
159 nb_processed = process_all_items(
160 new_file_queue,
161 processed_files_set,
162 process_file_fct,
163 processing_info.process_file_fct_extra_kwargs,
164 )
165 logging.debug(f"Processed {nb_processed} files.")
166 logging.info(
167 f"Executing on_dir_change_fct on data_dir {processing_info.data_dir}, with kwargs {processing_info.on_dir_change_fct_extra_kwargs}"
168 )
169 on_dir_change_fct(processing_info.data_dir, **processing_info.on_dir_change_fct_extra_kwargs)
171 logging.info(f"Starting monitoring new dir {new_processing_info.data_dir}")
172 # Start new monitoring
173 monitoring_started = True
174 processing_info = new_processing_info
175 file_event_handler = QueueingFilePatternMatchingEventHandler(
176 file_queue=new_file_queue, # queue should be empty at this point: either 1st monitoring so empty, or emptied by process_all_items above
177 patterns=file_patterns,
178 ignore_patterns=file_ignore_patterns,
179 ignore_directories=monitoring_ignore_directories,
180 case_sensitive=pattern_case_sensitive,
181 )
182 observer = Observer()
183 observer.schedule(file_event_handler, processing_info.data_dir)
184 observer.start()
185 logging.info("Monitoring started!")
187 # Add already present files in the queue
188 # It is possible to have duplicated files because when we start monitoring there could be files already present,
189 # which are then added to the queue. If a file is created between the monitoring start and the
190 # glob for already present files, they will be added twice to the queue, but processed_file_set will prevent them from
191 # been processed twice.
192 processed_files_set: set[Path] = set()
193 initial_files = [
194 f
195 for f in processing_info.data_dir.glob("*")
196 if f.is_file()
197 and match_any_paths( # use watchdogs pattern utils to apply same rules as during monitoring
198 [f],
199 included_patterns=file_patterns,
200 excluded_patterns=file_ignore_patterns,
201 case_sensitive=pattern_case_sensitive,
202 )
203 ]
204 logging.info(f"Adding {len(initial_files)} already present files to queue")
205 for f in initial_files:
206 new_file_queue.put_nowait(f) # maxsize is infinite so will never block
208 # move on to next loop iteration, where we will process the files in the "else" clause
210 logging.info("Stop event is set, stopping monitoring and processing remaining files")
211 # finish processing files and execute on_dir_change before stopping
212 if monitoring_started:
213 observer.stop()
214 observer.join()
215 nb_processed = process_all_items(
216 new_file_queue,
217 processed_files_set,
218 process_file_fct,
219 processing_info.process_file_fct_extra_kwargs,
220 )
221 logging.debug(f"Processed {nb_processed} files.")
222 logging.info(
223 f"Executing on_dir_change_fct on data_dir {processing_info.data_dir}, with kwargs {processing_info.on_dir_change_fct_extra_kwargs}"
224 )
225 on_dir_change_fct(processing_info.data_dir, **processing_info.on_dir_change_fct_extra_kwargs)
226 logging.info("Stopping monitoring")