Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/nbconvert/preprocessors/coalescestreams.py: 34%

1"""Preprocessor for merging consecutive stream outputs for easier handling.""" 

2 

3# Copyright (c) IPython Development Team. 

4# Distributed under the terms of the Modified BSD License. 

5 

6import functools 

7import re 

8 

9from traitlets.log import get_logger 

10 

11 

12def cell_preprocessor(function): 

13 """ 

14 Wrap a function to be executed on all cells of a notebook 

15 

16 The wrapped function should have these parameters: 

17 

18 cell : NotebookNode cell 

19 Notebook cell being processed 

20 resources : dictionary 

21 Additional resources used in the conversion process. Allows 

22 preprocessors to pass variables into the Jinja engine. 

23 index : int 

24 Index of the cell being processed 

25 """ 

26 

27 @functools.wraps(function) 

28 def wrappedfunc(nb, resources): 

29 get_logger().debug("Applying preprocessor: %s", function.__name__) 

30 for index, cell in enumerate(nb.cells): 

31 nb.cells[index], resources = function(cell, resources, index) 

32 return nb, resources 

33 

34 return wrappedfunc 

35 

36 

37cr_pat = re.compile(r".*\r(?=[^\n])") 

38 

39 

40@cell_preprocessor 

41def coalesce_streams(cell, resources, index): 

42 """ 

43 Merge consecutive sequences of stream output into single stream 

44 to prevent extra newlines inserted at flush calls 

45 

46 Parameters 

47 ---------- 

48 cell : NotebookNode cell 

49 Notebook cell being processed 

50 resources : dictionary 

51 Additional resources used in the conversion process. Allows 

52 transformers to pass variables into the Jinja engine. 

53 index : int 

54 Index of the cell being processed 

55 """ 

56 

57 outputs = cell.get("outputs", []) 

58 if not outputs: 

59 return cell, resources 

60 

61 last = outputs[0] 

62 new_outputs = [last] 

63 for output in outputs[1:]: 

64 if ( 

65 output.output_type == "stream" 

66 and last.output_type == "stream" 

67 and last.name == output.name 

68 ): 

69 last.text += output.text 

70 

71 else: 

72 new_outputs.append(output) 

73 last = output 

74 

75 # process \r characters 

76 for output in new_outputs: 

77 if output.output_type == "stream" and "\r" in output.text: 

78 output.text = cr_pat.sub("", output.text) 

79 

80 cell.outputs = new_outputs 

81 return cell, resources