1"""Preprocessor for merging consecutive stream outputs for easier handling."""
2
3import re
4
5# Copyright (c) IPython Development Team.
6# Distributed under the terms of the Modified BSD License.
7from nbconvert.preprocessors import Preprocessor
8
9CR_PAT = re.compile(r".*\r(?=[^\n])")
10
11
12class CoalesceStreamsPreprocessor(Preprocessor):
13 """
14 Merge consecutive sequences of stream output into single stream
15 to prevent extra newlines inserted at flush calls
16 """
17
18 def preprocess_cell(self, cell, resources, cell_index):
19 """
20 Apply a transformation on each cell. See base.py for details.
21 """
22 outputs = cell.get("outputs", [])
23 if not outputs:
24 return cell, resources
25
26 last = outputs[0]
27 new_outputs = [last]
28 for output in outputs[1:]:
29 if (
30 output.output_type == "stream"
31 and last.output_type == "stream"
32 and last.name == output.name
33 ):
34 last.text += output.text
35 else:
36 new_outputs.append(output)
37 last = output
38
39 # process \r characters
40 for output in new_outputs:
41 if output.output_type == "stream" and "\r" in output.text:
42 output.text = CR_PAT.sub("", output.text)
43
44 cell.outputs = new_outputs
45 return cell, resources