xref: /aosp_15_r20/external/grpc-grpc/tools/run_tests/python_utils/watch_dirs.py (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2)
1# Copyright 2015 gRPC authors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""Helper to watch a (set) of directories for modifications."""
15
16import os
17import time
18
19from six import string_types
20
21
22class DirWatcher(object):
23    """Helper to watch a (set) of directories for modifications."""
24
25    def __init__(self, paths):
26        if isinstance(paths, string_types):
27            paths = [paths]
28        self._done = False
29        self.paths = list(paths)
30        self.lastrun = time.time()
31        self._cache = self._calculate()
32
33    def _calculate(self):
34        """Walk over all subscribed paths, check most recent mtime."""
35        most_recent_change = None
36        for path in self.paths:
37            if not os.path.exists(path):
38                continue
39            if not os.path.isdir(path):
40                continue
41            for root, _, files in os.walk(path):
42                for f in files:
43                    if f and f[0] == ".":
44                        continue
45                    try:
46                        st = os.stat(os.path.join(root, f))
47                    except OSError as e:
48                        if e.errno == os.errno.ENOENT:
49                            continue
50                        raise
51                    if most_recent_change is None:
52                        most_recent_change = st.st_mtime
53                    else:
54                        most_recent_change = max(
55                            most_recent_change, st.st_mtime
56                        )
57        return most_recent_change
58
59    def most_recent_change(self):
60        if time.time() - self.lastrun > 1:
61            self._cache = self._calculate()
62            self.lastrun = time.time()
63        return self._cache
64