xref: /aosp_15_r20/external/pigweed/pw_env_setup/py/pw_env_setup/spinner.py (revision 61c4878ac05f98d0ceed94b57d316916de578985)
1# Copyright 2020 The Pigweed Authors
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may not
4# use this file except in compliance with the License. You may obtain a copy of
5# the License at
6#
7#     https://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, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations under
13# the License.
14"""Spinner!"""
15
16import contextlib
17import os
18import sys
19import threading
20import time
21
22
23class Spinner:
24    """Spinner!"""
25
26    def __init__(self, quiet=False):
27        self._done = None
28        self._thread = None
29        self._quiet = quiet
30
31    def _disabled(self):
32        if os.environ.get('PW_ENVSETUP_DISABLE_SPINNER'):
33            return True
34        if os.environ.get('PW_ENVSETUP_QUIET'):
35            return True
36        if self._quiet:
37            return True
38        if not sys.stdout.isatty():
39            return True
40        return False
41
42    def __del__(self):
43        self._done = True
44
45    def _spin(self):
46        i = 0
47        chars = '|/-\\'
48        while not self._done:
49            sys.stdout.write('[{}]'.format(chars[i]))
50            sys.stdout.flush()
51            time.sleep(0.1)
52            sys.stdout.write('\b\b\b')
53            i = (i + 1) % len(chars)
54
55    def start(self):
56        if self._disabled():
57            return
58
59        self._done = False
60        self._thread = threading.Thread(target=self._spin)
61        self._thread.start()
62
63    def stop(self):
64        if self._disabled():
65            return
66
67        assert self._thread
68        self._done = True
69        self._thread.join()
70        self._thread = None
71
72    @contextlib.contextmanager
73    def __call__(self):
74        try:
75            self.start()
76            yield self
77        finally:
78            self.stop()
79
80    @contextlib.contextmanager
81    def pause(self):
82        try:
83            self.stop()
84            yield self
85        finally:
86            self.start()
87