• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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(object):  # pylint: disable=useless-object-inheritance
24    """Spinner!"""
25    def __init__(self, quiet=False):
26        self._done = None
27        self._thread = None
28        self._quiet = quiet
29
30    def _disabled(self):
31        if os.environ.get('PW_ENVSETUP_DISABLE_SPINNER'):
32            return True
33        if os.environ.get('PW_ENVSETUP_QUIET'):
34            return True
35        if self._quiet:
36            return True
37        if not sys.stdout.isatty():
38            return True
39        return False
40
41    def __del__(self):
42        self._done = True
43
44    def _spin(self):
45        i = 0
46        chars = '|/-\\'
47        while not self._done:
48            sys.stdout.write('[{}]'.format(chars[i]))
49            sys.stdout.flush()
50            time.sleep(0.1)
51            sys.stdout.write('\b\b\b')
52            i = (i + 1) % len(chars)
53
54    def start(self):
55        if self._disabled():
56            return
57
58        self._done = False
59        self._thread = threading.Thread(target=self._spin)
60        self._thread.start()
61
62    def stop(self):
63        if self._disabled():
64            return
65
66        assert self._thread
67        self._done = True
68        self._thread.join()
69        self._thread = None
70
71    @contextlib.contextmanager
72    def __call__(self):
73        try:
74            self.start()
75            yield self
76        finally:
77            self.stop()
78
79    @contextlib.contextmanager
80    def pause(self):
81        try:
82            self.stop()
83            yield self
84        finally:
85            self.start()
86