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
|
import inspect
from abc import abstractmethod
from functools import singledispatch
from typing import Any, List, Sequence
from parsl.utils import RepresentationMixin
# Traverse the configuration hierarchy, returning a JSON component
# for each one. Configuration components which implement
# RepresentationMixin will be in the right form for inspecting
# object attributes. Configuration components which are lists or tuples
# are traversed in sequence. Other types default to reporting no
# usage information.
@singledispatch
def get_parsl_usage(obj) -> List[Any]:
return []
@get_parsl_usage.register
def get_parsl_usage_representation_mixin(obj: RepresentationMixin) -> List[Any]:
t = type(obj)
qualified_name = t.__module__ + "." + t.__name__
# me can contain anything that can be rendered as JSON
me: List[Any] = []
if isinstance(obj, UsageInformation):
# report rich usage information for this component
attrs = {'c': qualified_name}
attrs.update(obj.get_usage_information())
me = [attrs]
else:
# report the class name of this component
me = [qualified_name]
# unwrap typeguard-style unwrapping
init: Any = type(obj).__init__
if hasattr(init, '__wrapped__'):
init = init.__wrapped__
argspec = inspect.getfullargspec(init)
for arg in argspec.args[1:]: # skip first arg, self
arg_value = getattr(obj, arg)
d = get_parsl_usage(arg_value)
me += d
return me
@get_parsl_usage.register(list)
@get_parsl_usage.register(tuple)
def get_parsl_usage_sequence(obj: Sequence) -> List[Any]:
result = []
for v in obj:
result += get_parsl_usage(v)
return result
class UsageInformation:
@abstractmethod
def get_usage_information(self) -> dict:
pass
|
