rwbase.py
102 lines
| 3.6 KiB
| text/x-python
|
PythonLexer
|
r18573 | """Base classes and utilities for readers and writers.""" | ||
|
r18568 | |||
|
r18573 | # Copyright (c) IPython Development Team. | ||
# Distributed under the terms of the Modified BSD License. | ||||
|
r18568 | |||
|
r18602 | from IPython.utils.py3compat import string_types, cast_unicode_py2 | ||
|
r18568 | |||
def rejoin_lines(nb): | ||||
"""rejoin multiline text into strings | ||||
For reversing effects of ``split_lines(nb)``. | ||||
This only rejoins lines that have been split, so if text objects were not split | ||||
they will pass through unchanged. | ||||
Used when reading JSON files that may have been passed through split_lines. | ||||
""" | ||||
|
r18573 | for cell in nb.cells: | ||
if 'source' in cell and isinstance(cell.source, list): | ||||
|
r18602 | cell.source = ''.join(cell.source) | ||
|
r18596 | if cell.get('cell_type', None) == 'code': | ||
for output in cell.get('outputs', []): | ||||
output_type = output.get('output_type', '') | ||||
if output_type in {'execute_result', 'display_data'}: | ||||
for key, value in output.get('data', {}).items(): | ||||
|
r18589 | if key != 'application/json' and isinstance(value, list): | ||
|
r18602 | output.data[key] = ''.join(value) | ||
|
r18596 | elif output_type: | ||
if isinstance(output.get('text', ''), list): | ||||
|
r18602 | output.text = ''.join(output.text) | ||
|
r18568 | return nb | ||
|
r20843 | _non_text_split_mimes = { | ||
'application/javascript', | ||||
'image/svg+xml', | ||||
} | ||||
|
r18568 | |||
def split_lines(nb): | ||||
"""split likely multiline text into lists of strings | ||||
For file output more friendly to line-based VCS. ``rejoin_lines(nb)`` will | ||||
reverse the effects of ``split_lines(nb)``. | ||||
Used when writing JSON files. | ||||
""" | ||||
|
r18573 | for cell in nb.cells: | ||
source = cell.get('source', None) | ||||
if isinstance(source, string_types): | ||||
cell['source'] = source.splitlines(True) | ||||
if cell.cell_type == 'code': | ||||
for output in cell.outputs: | ||||
|
r18589 | if output.output_type in {'execute_result', 'display_data'}: | ||
for key, value in output.data.items(): | ||||
|
r20843 | if isinstance(value, string_types) and ( | ||
key.startswith('text/') or key in _non_text_split_mimes | ||||
): | ||||
|
r18589 | output.data[key] = value.splitlines(True) | ||
elif output.output_type == 'stream': | ||||
if isinstance(output.text, string_types): | ||||
output.text = output.text.splitlines(True) | ||||
|
r18568 | return nb | ||
|
r18581 | def strip_transient(nb): | ||
"""Strip transient values that shouldn't be stored in files. | ||||
This should be called in *both* read and write. | ||||
""" | ||||
nb.metadata.pop('orig_nbformat', None) | ||||
nb.metadata.pop('orig_nbformat_minor', None) | ||||
|
r19625 | nb.metadata.pop('signature', None) | ||
|
r18581 | for cell in nb.cells: | ||
cell.metadata.pop('trusted', None) | ||||
return nb | ||||
|
r18568 | class NotebookReader(object): | ||
"""A class for reading notebooks.""" | ||||
def reads(self, s, **kwargs): | ||||
"""Read a notebook from a string.""" | ||||
raise NotImplementedError("loads must be implemented in a subclass") | ||||
def read(self, fp, **kwargs): | ||||
"""Read a notebook from a file like object""" | ||||
|
r18602 | nbs = cast_unicode_py2(fp.read()) | ||
|
r18568 | return self.reads(nbs, **kwargs) | ||
class NotebookWriter(object): | ||||
"""A class for writing notebooks.""" | ||||
def writes(self, nb, **kwargs): | ||||
"""Write a notebook to a string.""" | ||||
raise NotImplementedError("loads must be implemented in a subclass") | ||||
def write(self, nb, fp, **kwargs): | ||||
"""Write a notebook to a file like object""" | ||||
|
r18602 | nbs = cast_unicode_py2(self.writes(nb, **kwargs)) | ||
|
r18568 | return fp.write(nbs) | ||