1
+ from collections .abc import Callable
1
2
from concurrent .futures import CancelledError , Future
2
- from typing import Callable , Generic , List , Optional , Tuple , TypeVar , cast
3
+ from typing import Any , Generic , Literal , Optional , TypeVar , Union
3
4
4
- T = TypeVar ('T' )
5
+ T = TypeVar ("T" )
5
6
6
7
7
8
class Job (Generic [T ]):
@@ -23,17 +24,12 @@ class Job(Generic[T]):
23
24
Picamera2.switch_mode_and_capture_array.
24
25
"""
25
26
26
- _functions : List [Callable [[], Tuple [bool , T ]]]
27
- _future : Future [T ]
28
- _signal_function : Optional [Callable [['Job[T]' ], None ]]
29
- _result : Optional [T ]
30
- calls : int
31
-
32
- def __init__ (
33
- self ,
34
- functions : List [Callable [[], Tuple [bool , T ]]],
35
- signal_function : Optional [Callable [['Job[T]' ], None ]] = None ,
36
- ) -> None :
27
+ def __init__ (self , functions : list [Callable [..., Union [
28
+ tuple [Literal [False ], None ],
29
+ tuple [Literal [True ], Any ],
30
+ tuple [Literal [True ], T ]
31
+ ]]], signal_function = None
32
+ ):
37
33
self ._functions = functions
38
34
self ._future = Future ()
39
35
self ._future .set_running_or_notify_cancel ()
@@ -80,7 +76,7 @@ def signal(self) -> None:
80
76
assert not self ._functions , "Job not finished!"
81
77
82
78
if not self ._future .done ():
83
- self ._future .set_result (cast ( T , self ._result ) )
79
+ self ._future .set_result (self ._result )
84
80
if self ._signal_function :
85
81
self ._signal_function (self )
86
82
0 commit comments