Coverage for src/loman/ui/viewmodel.py: 100%

1"""Pure view-model builders shared by the widget and its tests.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any 

6 

7from loman.computeengine import Error 

8from loman.consts import NodeAttributes, States 

9from loman.nodekey import NodeKey 

10from loman.visualization import ColorByState, GraphView, aggregate_states 

11 

12from .builder import describe_definition 

13from .value import to_wire 

14 

15if TYPE_CHECKING: 

16 from collections.abc import Mapping, Sequence 

17 

18 from loman.computeengine import Computation 

19 

20#: Label used where the graph renderer uses ``None``: a collapsed block whose 

21#: members are in genuinely different states. 

22MIXED_STATE_LABEL = "MIXED" 

23 

24 

25def state_label(states: Sequence[States | None]) -> str: 

26 """Return the display label for one rendered node or collapsed block. 

27 

28 Thin wrapper over :func:`loman.visualization.aggregate_states` that renames 

29 the mixed marker from ``None`` to a JSON-safe string, so the widget and the 

30 Graphviz picture always agree on what a rendered node represents. 

31 

32 :param states: States of the members behind one rendered node. 

33 :return: A :class:`~loman.consts.States` name, or ``MIXED``. 

34 """ 

35 state = aggregate_states(states) 

36 return MIXED_STATE_LABEL if state is None else state.name 

37 

38 

39def state_colors(cmap: dict[States | None, str] | None = None) -> dict[str, str]: 

40 """Convert Loman's state-colour mapping to JSON-safe string keys. 

41 

42 :param cmap: Custom state colours, or ``None`` for Loman's defaults. 

43 :return: Colours keyed by the labels :func:`state_label` produces. 

44 """ 

45 colors = ColorByState.DEFAULT_STATE_COLORS if cmap is None else cmap 

46 return {(MIXED_STATE_LABEL if state is None else state.name): color for state, color in colors.items()} 

47 

48 

49def node_states(view: GraphView) -> dict[str, str]: 

50 """Build the small rendered-ID to state map used for repainting. 

51 

52 :param view: The graph view whose rendered nodes should be described. 

53 :return: One state label per rendered node ID. 

54 """ 

55 result: dict[str, str] = {} 

56 for visible_key, node_id in view.node_index_map.items(): 

57 members = view.original_nodes[visible_key] 

58 states = [view.computation.dag.nodes[node][NodeAttributes.STATE] for node in members] 

59 result[node_id] = state_label(states) 

60 return result 

61 

62 

63def _safe_source(computation: Computation, node_key: NodeKey) -> str: 

64 """Return source text, tolerating interactive and restored callables. 

65 

66 Loman users define lambdas constantly, and :func:`inspect.getsource` cannot 

67 recover the text of one typed at a REPL or rehydrated from dill. The detail 

68 panel says so rather than failing. 

69 

70 :param computation: Computation owning the node. 

71 :param node_key: Node whose source is wanted. 

72 :return: Source text, or a short explanation of why it is unavailable. 

73 """ 

74 try: 

75 return computation.get_source(node_key) 

76 except (OSError, TypeError, SyntaxError) as exc: 

77 return f"Source unavailable for this callable ({type(exc).__name__})" 

78 

79 

80def build_detail( 

81 view: GraphView, 

82 node_id: str, 

83 *, 

84 editable: bool, 

85 id_to_visible: Mapping[str, NodeKey] | None = None, 

86 root: NodeKey | None = None, 

87) -> dict[str, Any]: 

88 """Build the lazily populated detail panel for one rendered node. 

89 

90 :param view: The graph view the rendered node belongs to. 

91 :param node_id: Rendered node ID, as carried by the SVG title element. 

92 :param editable: Whether the widget permits edits at all. 

93 :param id_to_visible: Reverse of ``view.node_index_map``. The widget keeps 

94 one and passes it in; omit it and this rebuilds it. 

95 :param root: The block the view is rooted on, so the node's definition is 

96 described in the same relative names the node form accepts. 

97 :return: The detail payload, or an empty dict for an unknown ID. 

98 """ 

99 if id_to_visible is None: 

100 id_to_visible = {value: key for key, value in view.node_index_map.items()} 

101 visible_key = id_to_visible.get(node_id) 

102 if visible_key is None: 

103 return {} 

104 computation = view.computation 

105 members = view.original_nodes[visible_key] 

106 member_states = [computation.dag.nodes[node][NodeAttributes.STATE] for node in members] 

107 detail: dict[str, Any] = { 

108 "id": node_id, 

109 "name": str(visible_key), 

110 "state": state_label(member_states), 

111 "members": [str(member) for member in members], 

112 "composite": visible_key in view.composite_nodes, 

113 "editable": False, 

114 } 

115 if len(members) != 1: 

116 return detail 

117 

118 node_key = members[0] 

119 node = computation.dag.nodes[node_key] 

120 value = node.get(NodeAttributes.VALUE) 

121 if isinstance(value, Error): 

122 # The repr of an Error carries the whole traceback as an escaped 

123 # one-line string. The traceback is reported separately and properly 

124 # formatted, so this keeps the value to the headline. 

125 value_wire = { 

126 "kind": "repr", 

127 "type": type(value.exception).__name__, 

128 "repr": f"{type(value.exception).__name__}: {value.exception}", 

129 } 

130 else: 

131 value_wire = to_wire(value) 

132 timing = node.get(NodeAttributes.TIMING) 

133 detail.update( 

134 { 

135 "name": str(node_key), 

136 "value": value_wire, 

137 "timing": None 

138 if timing is None 

139 else {"start": timing.start.isoformat(), "end": timing.end.isoformat(), "duration": timing.duration}, 

140 "source": _safe_source(computation, node_key), 

141 "inputs": [str(name) for name in computation.get_inputs(node_key)], 

142 "outputs": [str(name) for name in computation.get_outputs(node_key)], 

143 "definition": describe_definition(computation, node_key, root), 

144 } 

145 ) 

146 if isinstance(value, Error): 

147 detail["error"] = value.traceback 

148 # Editing is offered on input nodes only. Editing a calculated node would be 

149 # silently discarded by the next compute, which is worse than not offering it. 

150 writable_node = ( 

151 editable and node.get(NodeAttributes.FUNC) is None and node.get(NodeAttributes.STATE) != States.PLACEHOLDER 

152 ) 

153 detail["editable"] = bool(writable_node and value_wire["kind"] == "scalar") 

154 detail["cells_editable"] = bool(writable_node and value_wire["kind"] == "table" and value_wire.get("editable")) 

155 return detail