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
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
from __future__ import annotations
import logging
from typing import Callable, ContextManager, List, Optional, Sequence
from rope.base.taskhandle import BaseJobSet, BaseTaskHandle
from pylsp._utils import throttle
from pylsp.workspace import Workspace
log = logging.getLogger(__name__)
Report = Callable[[str, int], None]
class PylspJobSet(BaseJobSet):
count: int = 0
done: int = 0
_reporter: Report
_report_iter: ContextManager
job_name: str = ""
def __init__(self, count: Optional[int], report_iter: ContextManager) -> None:
if count is not None:
self.count = count
self._reporter = report_iter.__enter__()
self._report_iter = report_iter
def started_job(self, name: Optional[str]) -> None:
if name:
self.job_name = name
def finished_job(self) -> None:
self.done += 1
if self.get_percent_done() is not None and int(self.get_percent_done()) >= 100:
if self._report_iter is None:
return
self._report_iter.__exit__(None, None, None)
self._report_iter = None
else:
self._report()
def check_status(self) -> None:
pass
def get_percent_done(self) -> Optional[float]:
if self.count == 0:
return 0
return (self.done / self.count) * 100
def increment(self) -> None:
"""
Increment the number of tasks to complete.
This is used if the number is not known ahead of time.
"""
self.count += 1
self._report()
@throttle(0.5)
def _report(self) -> None:
percent = int(self.get_percent_done())
message = f"{self.job_name} {self.done}/{self.count}"
log.debug(f"Reporting {message} {percent}%")
self._reporter(message, percent)
class PylspTaskHandle(BaseTaskHandle):
name: str
observers: List
job_sets: List[PylspJobSet]
stopped: bool
workspace: Workspace
_report: Callable[[str, str], None]
def __init__(self, workspace: Workspace) -> None:
self.workspace = workspace
self.job_sets = []
self.observers = []
def create_jobset(self, name="JobSet", count: Optional[int] = None):
report_iter = self.workspace.report_progress(
name, None, None, skip_token_initialization=True
)
result = PylspJobSet(count, report_iter)
self.job_sets.append(result)
self._inform_observers()
return result
def stop(self) -> None:
pass
def current_jobset(self) -> Optional[BaseJobSet]:
pass
def add_observer(self) -> None:
pass
def is_stopped(self) -> bool:
pass
def get_jobsets(self) -> Sequence[BaseJobSet]:
pass
def _inform_observers(self) -> None:
for observer in self.observers:
observer()
|
