-
Notifications
You must be signed in to change notification settings - Fork 517
Expand file tree
/
Copy pathclean.py
More file actions
331 lines (292 loc) · 14.8 KB
/
Copy pathclean.py
File metadata and controls
331 lines (292 loc) · 14.8 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
"""Strip superfluous metadata from notebooks
Docs: https://nbdev.fast.ai/api/clean.html.md"""
# AUTOGENERATED! DO NOT EDIT! File to edit: ../nbs/api/11_clean.ipynb.
# %% auto #0
__all__ = ['nbdev_trust', 'clean_nb', 'process_write', 'nbdev_clean', 'clean_jupyter', 'nbdev_install_hooks']
# %% ../nbs/api/11_clean.ipynb #07637414
import ast,warnings,stat
from astunparse import unparse
from textwrap import indent
from fastcore.nbio import *
from fastcore.nbio import _directive, _dir_line, _meta_directives, _unparse_dir
from fastcore.script import *
from fastcore.utils import *
from fastcore.xtras import *
from .imports import *
from .config import *
from .sync import *
from .process import first_code_ln
# %% ../nbs/api/11_clean.ipynb #4259f92c
@call_parse
def nbdev_trust(
fname:str=None, # A notebook name or glob to trust
force_all:bool=False # Also trust notebooks that haven't changed
):
"Trust notebooks matching `fname`."
try: from nbformat.sign import NotebookNotary
except:
import warnings
warnings.warn("Please install jupyter and try again")
return
from nbformat import read
fname = Path(fname if fname else get_config().nbs_path)
path = fname if fname.is_dir() else fname.parent
check_fname = path/".last_checked"
last_checked = os.path.getmtime(check_fname) if check_fname.exists() else None
nbs = globtastic(fname, file_glob='*.ipynb', skip_folder_re='^[_.]') if fname.is_dir() else [fname]
for fn in nbs:
if last_checked and not force_all:
last_changed = os.path.getmtime(fn)
if last_changed < last_checked: continue
with open(fn, 'r', encoding='utf-8') as f: nb = read(f, as_version=4)
if not NotebookNotary().check_signature(nb): NotebookNotary().sign(nb)
check_fname.touch(exist_ok=True)
# %% ../nbs/api/11_clean.ipynb #a2ba2b4c
_repr_id_re = re.compile('(<.*?)( at 0x[0-9a-fA-F]+)(>)')
_sub = partial(_repr_id_re.sub, r'\1\3')
def _skip_or_sub(x): return _sub(x) if "at 0x" in x else x
def _clean_cell_output_id(lines):
return _skip_or_sub(lines) if isinstance(lines,str) else [_skip_or_sub(o) for o in lines]
# %% ../nbs/api/11_clean.ipynb #b4cde615
def _clean_cell_output(cell, clean_ids, allowed_out_meta_keys):
"Remove `cell` output execution count and optionally ids from text reprs"
outputs = cell.get('outputs', [])
for o in outputs:
if 'execution_count' in o: o['execution_count'] = None
data = o.get('data', {})
data.pop("application/vnd.google.colaboratory.intrinsic+json", None)
for k in data:
if k.startswith('text') and clean_ids: data[k] = _clean_cell_output_id(data[k])
if k.startswith('image') and "svg" not in k: data[k] = data[k].rstrip()
if 'text' in o and clean_ids: o['text'] = _clean_cell_output_id(o['text'])
if 'metadata' in o: o['metadata'] = {k:v for k,v in o['metadata'].items() if k in allowed_out_meta_keys}
# %% ../nbs/api/11_clean.ipynb #2ba79c93
def _clean_cell(cell, clear_all, allowed_metadata_keys, clean_ids, allowed_out_meta_keys):
"Clean `cell` by removing superfluous metadata or everything except the input if `clear_all`"
if 'execution_count' in cell: cell['execution_count'] = None
if 'outputs' in cell:
if clear_all: cell['outputs'] = []
else: _clean_cell_output(cell, clean_ids, allowed_out_meta_keys)
if cell['source'] == ['']: cell['source'] = []
cell['metadata'] = {} if clear_all else {
k:v for k,v in cell['metadata'].items() if k in allowed_metadata_keys}
if 'id' not in cell: cell['id'] = rtoken_hex(4)
# %% ../nbs/api/11_clean.ipynb #e8101222
def clean_nb(
nb, # The notebook to clean
clear_all=False, # Remove all cell metadata and cell outputs?
allowed_metadata_keys:list=None, # Preserve the list of keys in the main notebook metadata
allowed_cell_metadata_keys:list=None, # Preserve the list of keys in cell level metadata
clean_ids=True, # Remove ids from plaintext reprs?
allowed_out_metadata_keys:list=None, # Preserve the list of keys in output metadata
repair:bool=True, # Fix structural problems first (see `repair_nb`)?
):
"Clean `nb` from superfluous metadata"
if repair: repair_nb(nb)
metadata_keys = {"kernelspec", "jekyll", "jupytext", "doc", "widgets", "nbdev"}
if allowed_metadata_keys: metadata_keys.update(allowed_metadata_keys)
cell_metadata_keys = {"hide_input", "nbdev"}
if allowed_cell_metadata_keys: cell_metadata_keys.update(allowed_cell_metadata_keys)
out_meta_keys = set()
if allowed_out_metadata_keys: out_meta_keys.update(allowed_out_metadata_keys)
for c in nb['cells']: _clean_cell(c, clear_all, cell_metadata_keys, clean_ids, out_meta_keys)
if nb.get('metadata', {}).get('kernelspec', {}).get('name', None):
nb['metadata']['kernelspec']['display_name'] = nb["metadata"]["kernelspec"]["name"]
nb['metadata'] = {k:v for k,v in nb['metadata'].items() if k in metadata_keys}
# Cell IDs were added in nbformat 4.5
if nb.get('nbformat') == 4 and nb.get('nbformat_minor', 0) < 5: nb['nbformat_minor'] = 5
# %% ../nbs/api/11_clean.ipynb #604d83e6
def _reconfigure(*strms):
for s in strms:
if hasattr(s,'reconfigure'): s.reconfigure(encoding='utf-8')
# %% ../nbs/api/11_clean.ipynb #d251837f
def process_write(warn_msg, proc_nb, f_in, f_out=None, disp=False):
if not f_out: f_out = f_in
if isinstance(f_in, (str,Path)): f_in = Path(f_in).open(encoding="utf-8")
try:
_reconfigure(f_in, f_out)
nb = loads(f_in.read())
proc_nb(nb)
write_nb(nb, f_out) if not disp else sys.stdout.write(nb2str(nb))
except Exception as e:
warn(f'{warn_msg}')
warn(e)
# %% ../nbs/api/11_clean.ipynb #f6e854ac
def _cmt_dirs(cell):
"Comment directives in `cell` as `{name: value}`, plus the partitioned `(dirs,code)` lines"
dirs,code = cell._partition()
return dict(t for t in (_directive(s, cell.lang_) for s in dirs) if t),dirs,code
def _rm_dir_lines(cell, dirs, code, names):
"Rewrite `cell` source without the directive lines in `names`"
cell.set_source(''.join([o for o in dirs if (t:=_directive(o, cell.lang_)) is None or t[0] not in names] + code))
def _to_meta(cell, names):
"Move comment directives in `names` to the cell's `nbdev` metadata key"
cmts,dirs,code = _cmt_dirs(cell)
move = {k:v for k,v in cmts.items() if k in names}
if not move: return
cell.setdefault('metadata',{}).setdefault('nbdev',{}).update({k:_unparse_dir(v) for k,v in move.items()})
_rm_dir_lines(cell, dirs, code, move)
def _to_comments(cell, names):
"Move directives in `names` from the cell's `nbdev` metadata key to comments"
move = {k:v for k,v in _meta_directives(cell.get('metadata')).items() if k in names}
if not move: return
nbd = cell.metadata['nbdev']
for k in move: nbd.pop(k, None)
if not nbd: del cell.metadata['nbdev']
dirs,code = cell._partition()
cell.set_source(''.join(dirs + [_dir_line(k, v, cell.lang_) for k,v in move.items()] + code))
# %% ../nbs/api/11_clean.ipynb #6e8d013d
def _canon_dirs(cell):
"Rewrite `cell`'s comment directives in canonical form (colon-separated, bare for true)"
dirs,code = cell._partition()
new = [o if (t:=_directive(o, cell.lang_)) is None else _dir_line(*t, lang=cell.lang_) for o in dirs]
if new != dirs: cell.set_source(''.join(new + code))
def _hoist_nb_meta(nb, names=('default_exp',)):
"Move notebook-scope directives in `names` to notebook metadata, dropping any cell left empty"
for c in list(nb.cells):
cmts,dirs,code = _cmt_dirs(c)
move = {k:v for k,v in cmts.items() if k in names}
if not move: continue
nb.setdefault('metadata',{}).setdefault('nbdev',{}).update({k:_unparse_dir(v) for k,v in move.items()})
_rm_dir_lines(c, dirs, code, move)
left = set(c.directives) - {'hide'}
if not left and not ''.join(c._partition()[1]).strip(): nb.cells.remove(c)
def _dir_moves(nb, dirs=False, to_meta=None, to_comments=None, nb_meta=False):
"Apply directive migrations to loaded notebook dict `nb` in place"
nbo = dict2nb(nb)
if to_meta:
for c in nbo.cells: _to_meta(c, to_meta.split())
if to_comments:
for c in nbo.cells: _to_comments(c, to_comments.split())
if nb_meta: _hoist_nb_meta(nbo)
if dirs:
for c in nbo.cells: _canon_dirs(c)
nb['cells'],nb['metadata'] = nbo.cells,nbo.metadata
# %% ../nbs/api/11_clean.ipynb #714357ce
def _nbdev_clean(nb, path=None, clear_all=None, repair=True, dirs=False, to_meta=None, to_comments=None, nb_meta=False):
cfg = get_config(path=path)
clear_all = clear_all or cfg.clear_all
allowed_metadata_keys = cfg.get("allowed_metadata_keys") or []
allowed_cell_metadata_keys = cfg.get("allowed_cell_metadata_keys") or []
allowed_out_metadata_keys = cfg.get("allowed_out_metadata_keys") or []
if dirs or to_meta or to_comments or nb_meta: _dir_moves(nb, dirs, to_meta, to_comments, nb_meta)
clean_nb(nb, clear_all, allowed_metadata_keys, allowed_cell_metadata_keys, cfg.clean_ids, allowed_out_metadata_keys, repair=repair)
if path: nbdev_trust.__wrapped__(path)
# %% ../nbs/api/11_clean.ipynb #6af3b9d4
@call_parse
def nbdev_clean(
fname:str=None, # A notebook name or glob to clean
clear_all:bool=False, # Remove all cell metadata and cell outputs?
disp:bool=False, # Print the cleaned outputs
stdin:bool=False, # Read notebook from input stream
repair:bool_arg=True, # Fix structural problems, e.g. stray outputs on non-code cells (see `repair_nb`)?
dirs:bool=False, # Rewrite comment directives in canonical form?
to_meta:str=None, # Space-separated directive names to move from comments to cell metadata
to_comments:str=None, # Space-separated directive names to move from cell metadata to comments
nb_meta:bool=False # Move `default_exp` into notebook metadata?
):
"Clean all notebooks in `fname` to avoid merge conflicts"
# Git hooks will pass the notebooks in stdin
_clean = partial(_nbdev_clean, clear_all=clear_all, repair=repair, dirs=dirs, to_meta=to_meta, to_comments=to_comments, nb_meta=nb_meta)
_write = partial(process_write, warn_msg='Failed to clean notebook', proc_nb=_clean)
if stdin: return _write(f_in=sys.stdin, f_out=sys.stdout)
if fname is None: fname = get_config().nbs_path
for f in globtastic(fname, file_glob='*.ipynb', skip_folder_re='^[_.]'): _write(f_in=f, disp=disp)
# %% ../nbs/api/11_clean.ipynb #f84289fc
def clean_jupyter(path, model, **kwargs):
"Clean Jupyter `model` pre save to `path`"
if not (model['type']=='notebook' and model['content']['nbformat']==4): return
jupyter_hooks = get_config(path=path).jupyter_hooks
if jupyter_hooks: _nbdev_clean(model['content'], path=path)
# %% ../nbs/api/11_clean.ipynb #b7c19563
_pre_save_hook_src = '''
def nbdev_clean_jupyter(**kwargs):
try: from nbdev.clean import clean_jupyter
except ModuleNotFoundError: return
clean_jupyter(**kwargs)
c.ContentsManager.pre_save_hook = nbdev_clean_jupyter'''.strip()
_pre_save_hook_re = re.compile(r'c\.(File)?ContentsManager\.pre_save_hook')
# %% ../nbs/api/11_clean.ipynb #fcb8df4b
def _add_jupyter_hooks(src, path):
if _pre_save_hook_src in src: return
mod = ast.parse(src)
for node in ast.walk(mod):
if not isinstance(node,ast.Assign): continue
target = only(node.targets)
if _pre_save_hook_re.match(unparse(target)):
pre = ' '*2
old = indent(unparse(node), pre)
new = indent(_pre_save_hook_src, pre)
sys.stderr.write(f"Can't install hook to '{path}' since it already contains:\n{old}\n"
f"Manually update to the following (without indentation) for this functionality:\n\n{new}\n\n")
return
src = src.rstrip()
if src: src+='\n\n'
return src+_pre_save_hook_src
# %% ../nbs/api/11_clean.ipynb #cc677e44
def _git_root():
try: return Path(run('git rev-parse --show-toplevel'))
except OSError: return None
# %% ../nbs/api/11_clean.ipynb #e6083614
def _add_attrs(path, attrs):
"Append missing attribute lines to git attributes file at `path`"
txt = path.read_text() if path.exists() else ''
have = [l.split() for l in txt.splitlines()] # whitespace-insensitive: nbdime writes its lines with tabs
for attr in attrs:
if attr.split() not in have:
if txt and not txt.endswith('\n'): txt+='\n'
txt += attr+'\n'
path.write_text(txt)
def _cfg_drivers(loc, merge, diff):
"Define the `jupyternotebook` merge/diff drivers via `git config <loc>`"
if merge:
run(f'git config {loc} merge.jupyternotebook.name "resolve conflicts with nbdev_fix"')
run(f'git config {loc} merge.jupyternotebook.driver "nbdev-merge %O %A %B %P"')
if diff: run(f'git config {loc} diff.jupyternotebook.command nbdev-diff-driver')
@call_parse
def nbdev_install_hooks(
merge:bool_arg=True, # Install the notebook merge driver?
diff:bool_arg=True, # Install the notebook diff driver?
globally:bool_arg=False # Define the drivers in `~/.gitconfig` and the global attributes file, instead of repo files?
):
"Install Jupyter and git hooks to automatically clean, trust, and fix merge conflicts in notebooks"
cfg_path = Path.home()/'.jupyter'
cfg_path.mkdir(exist_ok=True)
cfg_fns = [cfg_path/f'jupyter_{o}_config.py' for o in ('notebook','server')]
for fn in cfg_fns:
src = fn.read_text() if fn.exists() else ''
upd = _add_jupyter_hooks(src, fn)
if upd is not None: fn.write_text(upd)
nbdev_attrs = (['*.ipynb merge=jupyternotebook'] if merge else []) + (['*.ipynb diff=jupyternotebook'] if diff else [])
if globally:
_cfg_drivers('--global', merge, diff)
rc,p = run('git config --global --get core.attributesFile', ignore_ex=True)
if rc: p = os.environ.get('XDG_CONFIG_HOME', '~/.config') + '/git/attributes'
attrs_path = Path(p).expanduser()
attrs_path.parent.mkdir(parents=True, exist_ok=True)
_add_attrs(attrs_path, nbdev_attrs)
return print("Hooks are installed globally.")
repo_path = _git_root()
if repo_path is None:
sys.stderr.write('Not in a git repository, git hooks cannot be installed.\n')
return
hook_path = repo_path/'.git'/'hooks'
fn = hook_path/'post-merge'
hook_path.mkdir(parents=True, exist_ok=True)
fn.write_text("#!/bin/bash\nnbdev-trust")
os.chmod(fn, os.stat(fn).st_mode | stat.S_IEXEC)
cmd = 'git config --local include.path ../.gitconfig'
cfg_fn = repo_path/'.gitconfig'
cfg_fn.write_text(f'''# Generated by nbdev-install-hooks
#
# If you need to disable this instrumentation do:
# git config --local --unset include.path
#
# To restore:
# {cmd}
#
''')
_cfg_drivers(f'--file "{cfg_fn}"', merge, diff)
run(cmd)
_add_attrs(repo_path/'.gitattributes', nbdev_attrs)
print("Hooks are installed.")