Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/nbclient/output_widget.py: 20%

1"""An output widget mimic.""" 

2from typing import Any, Dict, List, Optional 

3 

4from jupyter_client.client import KernelClient 

5from nbformat.v4 import output_from_msg 

6 

7from .jsonutil import json_clean 

8 

9 

10class OutputWidget: 

11 """This class mimics a front end output widget""" 

12 

13 def __init__( 

14 self, comm_id: str, state: Dict[str, Any], kernel_client: KernelClient, executor: Any 

15 ) -> None: 

16 """Initialize the widget.""" 

17 self.comm_id: str = comm_id 

18 self.state: Dict[str, Any] = state 

19 self.kernel_client: KernelClient = kernel_client 

20 self.executor = executor 

21 self.topic: bytes = ('comm-%s' % self.comm_id).encode('ascii') 

22 self.outputs: List = self.state['outputs'] 

23 self.clear_before_next_output: bool = False 

24 

25 def clear_output(self, outs: List, msg: Dict, cell_index: int) -> None: 

26 """Clear output.""" 

27 self.parent_header = msg['parent_header'] 

28 content = msg['content'] 

29 if content.get('wait'): 

30 self.clear_before_next_output = True 

31 else: 

32 self.outputs = [] 

33 # sync back the state to the kernel 

34 self.sync_state() 

35 if hasattr(self.executor, 'widget_state'): 

36 # sync the state to the nbconvert state as well, since that is used for testing 

37 self.executor.widget_state[self.comm_id]['outputs'] = self.outputs 

38 

39 def sync_state(self) -> None: 

40 """Sync state.""" 

41 state = {'outputs': self.outputs} 

42 msg = {'method': 'update', 'state': state, 'buffer_paths': []} 

43 self.send(msg) 

44 

45 def _publish_msg( 

46 self, 

47 msg_type: str, 

48 data: Optional[Dict] = None, 

49 metadata: Optional[Dict] = None, 

50 buffers: Optional[List] = None, 

51 **keys: Any, 

52 ) -> None: 

53 """Helper for sending a comm message on IOPub""" 

54 data = {} if data is None else data 

55 metadata = {} if metadata is None else metadata 

56 content = json_clean(dict(data=data, comm_id=self.comm_id, **keys)) 

57 msg = self.kernel_client.session.msg( 

58 msg_type, content=content, parent=self.parent_header, metadata=metadata 

59 ) 

60 self.kernel_client.shell_channel.send(msg) 

61 

62 def send( 

63 self, 

64 data: Optional[Dict] = None, 

65 metadata: Optional[Dict] = None, 

66 buffers: Optional[List] = None, 

67 ) -> None: 

68 """Send a comm message.""" 

69 self._publish_msg('comm_msg', data=data, metadata=metadata, buffers=buffers) 

70 

71 def output(self, outs: List, msg: Dict, display_id: str, cell_index: int) -> None: 

72 """Handle output.""" 

73 if self.clear_before_next_output: 

74 self.outputs = [] 

75 self.clear_before_next_output = False 

76 self.parent_header = msg['parent_header'] 

77 output = output_from_msg(msg) 

78 

79 if self.outputs: 

80 # try to coalesce/merge output text 

81 last_output = self.outputs[-1] 

82 if ( 

83 last_output['output_type'] == 'stream' 

84 and output['output_type'] == 'stream' 

85 and last_output['name'] == output['name'] 

86 ): 

87 last_output['text'] += output['text'] 

88 else: 

89 self.outputs.append(output) 

90 else: 

91 self.outputs.append(output) 

92 self.sync_state() 

93 if hasattr(self.executor, 'widget_state'): 

94 # sync the state to the nbconvert state as well, since that is used for testing 

95 self.executor.widget_state[self.comm_id]['outputs'] = self.outputs 

96 

97 def set_state(self, state: Dict) -> None: 

98 """Set the state.""" 

99 if 'msg_id' in state: 

100 msg_id = state.get('msg_id') 

101 if msg_id: 

102 self.executor.register_output_hook(msg_id, self) 

103 self.msg_id = msg_id 

104 else: 

105 self.executor.remove_output_hook(self.msg_id, self) 

106 self.msg_id = msg_id 

107 

108 def handle_msg(self, msg: Dict) -> None: 

109 """Handle a message.""" 

110 content = msg['content'] 

111 comm_id = content['comm_id'] 

112 if comm_id != self.comm_id: 

113 raise AssertionError('Mismatched comm id') 

114 data = content['data'] 

115 if 'state' in data: 

116 self.set_state(data['state'])