quarto.py

"""Install and interact with Quarto from nbdev"""

# AUTOGENERATED! DO NOT EDIT! File to edit: ../nbs/api/14_quarto.ipynb.

# %% ../nbs/api/14_quarto.ipynb #6a35c7c4-748f-4c82-a9bf-c780a8d83e90
from __future__ import annotations
import subprocess,sys,shutil,ast,warnings,traceback
from os import system
from contextlib import contextmanager

from .config import *
from .doclinks import *

from fastcore.utils import *
from fastcore.script import call_parse
from fastcore.shutil import rmtree,move,copytree
from fastcore.meta import delegates
from .serve import proc_nbs,_proc_file
from . import serve_drv
import yaml

# %% auto #0
__all__ = ['BASE_QUARTO_URL', 'install_quarto', 'install', 'IndentDumper', 'nbdev_sidebar', 'refresh_quarto_yml',
           'nbdev_proc_nbs', 'nbdev_readme', 'nbdev_contributing', 'nbdev_docs', 'prepare', 'fs_watchdog',
           'nbdev_preview']

# %% ../nbs/api/14_quarto.ipynb #aae2d2be-ad03-4536-bf70-c4575f39cea3
def _sprun(cmd):
    try: subprocess.check_output(cmd, shell=True)
    except subprocess.CalledProcessError as cpe: sys.exit(cpe.returncode)

# %% ../nbs/api/14_quarto.ipynb #75e4b6a1
BASE_QUARTO_URL='https://www.quarto.org/download/latest/'

def _install_linux():
    from os import uname
    machine = 'arm' if uname().machine in ('arm64', 'aarch64', 'armv8', 'armv8l') else 'amd'
    system(f'curl -LO {BASE_QUARTO_URL}quarto-linux-{machine}64.deb')
    system(f'sudo dpkg -i quarto-linux-{machine}64.deb && rm quarto-linux-{machine}64.deb')
    
def _install_mac():
    system(f'curl -LO {BASE_QUARTO_URL}quarto-macos.pkg')
    system('sudo installer -pkg quarto-macos.pkg -target / && rm quarto-macos.pkg')

@call_parse
def install_quarto():
    "Install latest Quarto on macOS or Linux, prints instructions for Windows"
    if sys.platform not in ('darwin','linux'):
        return print('Please visit https://quarto.org/docs/get-started/ to install quarto')
    print("Installing or upgrading quarto -- this requires root access.")
    system('sudo touch .installing')
    try:
        installing = Path('.installing')
        if not installing.exists(): return print("Cancelled. Please download and install Quarto from quarto.org.")
        if 'darwin' in sys.platform: _install_mac()
        elif 'linux' in sys.platform: _install_linux()
    finally: system('sudo rm -f .installing')

# %% ../nbs/api/14_quarto.ipynb #7580d48f-e10e-4bb0-937e-90645b5dfd08
@call_parse
def install():
    "Install Quarto and the current library"
    install_quarto.__wrapped__()
    d = get_config().lib_path
    if (d/'__init__.py').exists(): system(f'pip install -e "{d.parent}[dev]"')

# %% ../nbs/api/14_quarto.ipynb #b93f6def
def _pre(p,b=True): return '    ' * (len(p.parts)) + ('- ' if b else '  ')
def _sort(a):
    x,y = a
    if y.startswith('index.'): return x,'00'
    return a

_def_file_re = r'\.(?:ipynb|qmd|html)$'

@delegates
def _nbglob_docs(
    path:str=None, # Path to notebooks
    file_glob:str=None, # Only include files matching glob    
    file_re:str=_def_file_re, # Only include files matching regex
    **kwargs):
    return nbglob(path, file_glob=file_glob, file_re=file_re, **kwargs)

# %% ../nbs/api/14_quarto.ipynb #37ab90f4
def _recursive_parser(
        dir_dict: dict, # Directory structure as a dict.
        contents: list, # `contents` list from `sidebar.yaml` template dict.
        dirpath: Path,  # Directory path.
        section = None, # `section` mapping.
        set_index: bool = True): # If `True`, `index` file will be set to href.
    for name, val in dir_dict.items():
        if type(val) is str:
            if re.search(r'index\..*', re.sub(r'^\d+_', '', val)) and set_index and section:
                section.update({'href': str(dirpath/val)})
            else:
                contents.append(str(dirpath/val))
        elif type(val) is dict:
            name = re.sub(r'^\d+_', '', name)
            section = {'section': name, 'contents': []}
            contents.append(section)
            _recursive_parser(val, section['contents'], dirpath/name, section=section)

class IndentDumper(yaml.Dumper):
    def increase_indent(self, flow=False, indentless=False):
        return super(IndentDumper, self).increase_indent(flow, False)

# %% ../nbs/api/14_quarto.ipynb #d879c7e1
@call_parse
@delegates(_nbglob_docs)
def nbdev_sidebar(
    path:str=None, # Path to notebooks
    printit:bool=False,  # Print YAML for debugging
    force:bool=False,  # Create sidebar even if custom_sidebar=false in pyproject.toml
    skip_folder_re:str=r'(?:^[_.]|^www\$)', # Skip folders matching regex
    **kwargs):
    "Create sidebar.yml"
    if not force and get_config().custom_sidebar: return
    path = get_config().nbs_path if not path else Path(path)
    def _f(a,b): return Path(a),b
    files = nbglob(path, func=_f, skip_folder_re=skip_folder_re, **kwargs).sorted(key=_sort)
    lastd, res = Path(), []

    # Parse directory structure to dict.
    # dir => dict(), file => file.
    parsed_struct = {'website': {'sidebar': {'contents': []}}}
    _contents = parsed_struct['website']['sidebar']['contents']
    dir_struct = dict()
    for dabs, name in files:
        drel = dabs.relative_to(path)
        _dir = dir_struct
        for subdir in drel.parts:
            _dir = _dir.setdefault(subdir, dict())
        _dir[name] = name

    _recursive_parser(dir_struct, _contents, Path())
    yml_path = path/'sidebar.yml'
    yml = yaml.dump(parsed_struct, Dumper=IndentDumper, sort_keys=False)

    if printit: return print(yml)
    yml_path.write_text(yml)

# %% ../nbs/api/14_quarto.ipynb #aabf2f15
_quarto_yml="""project:
  type: website

format:
  html:
    theme: cosmo
    css: styles.css
    toc: true
    keep-md: true
  commonmark: default

website:
  twitter-card: true
  open-graph: true
  repo-actions: [issue]
  navbar:
    background: primary
    search: true
  sidebar:
    style: floating

metadata-files: [nbdev.yml, sidebar.yml]"""

# %% ../nbs/api/14_quarto.ipynb #faabf6e6
_nbdev_yml="""project:
  output-dir: {doc_path}

website:
  title: "{title}"
  site-url: "{doc_host}{doc_baseurl}"
  description: "{description}"
  repo-branch: {branch}
  repo-url: "{git_url}"
"""

# %% ../nbs/api/14_quarto.ipynb #38124450
def refresh_quarto_yml():
    "Generate `_quarto.yml` from `pyproject.toml`."
    cfg = get_config()
    ny = cfg.nbs_path/'nbdev.yml'
    vals = {k:cfg[k] for k in ['title', 'description', 'branch', 'git_url', 'doc_host', 'doc_baseurl']}
    vals['doc_path'] = cfg.doc_path.name
    if 'title' not in vals: vals['title'] = vals['lib_name']
    ny.write_text(_nbdev_yml.format(**vals))
    qy = cfg.nbs_path/'_quarto.yml'
    if 'custom_quarto_yml' in cfg: print("NB: `_quarto.yml` is no longer auto-updated. Remove `custom_quarto_yml` from `pyproject.toml`")
    if qy.exists() and not str2bool(cfg.get('custom_quarto_yml', True)): qy.unlink()
    if not qy.exists(): qy.write_text(_quarto_yml)

# %% ../nbs/api/14_quarto.ipynb #975d370e
def _ensure_quarto():
    if shutil.which('quarto'): return
    print("Quarto is not installed. We will download and install it for you.")
    install.__wrapped__()

# %% ../nbs/api/14_quarto.ipynb #6b880922
def _pre_docs(path=None, n_workers:int=defaults.cpus, **kwargs):
    cfg = get_config()
    path = Path(path) if path else cfg.nbs_path
    _ensure_quarto()
    refresh_quarto_yml()
    import nbdev.doclinks
    nbdev.doclinks._build_modidx()
    nbdev_sidebar.__wrapped__(path=path, **kwargs)
    cache = proc_nbs(path, n_workers=n_workers, **kwargs)
    return cache,cfg,path

# %% ../nbs/api/14_quarto.ipynb #e3806fa9
@call_parse
@delegates(proc_nbs)
def nbdev_proc_nbs(**kwargs):
    "Process notebooks in `path` for docs rendering"
    _pre_docs(**kwargs)[0]

# %% ../nbs/api/14_quarto.ipynb #c5c7bb0e
def _doc_mtime_not_older(readme_path, readme_nb_path):
    if not readme_nb_path.exists():
        print(f"Could not find {readme_nb_path}")
        return True
    return readme_path.exists() and readme_path.stat().st_mtime>=readme_nb_path.stat().st_mtime

# %% ../nbs/api/14_quarto.ipynb #7b70b5c8
class _SidebarYmlRemoved:
    "Context manager for `nbdev-readme` to avoid rendering whole docs website"
    def __init__(self,path): self._path=path
    def __enter__(self):
        self._yml_path = self._path/'sidebar.yml'
        self._moved=False
        if self._yml_path.exists():
            self._yml_path.rename(self._path/'sidebar.yml.bak')
            self._moved=True
    def __exit__(self, exc_type, exc_value, exc_tb):
        if self._moved: (self._path/'sidebar.yml.bak').rename(self._yml_path)

# %% ../nbs/api/14_quarto.ipynb #49661bbf
def _copytree(a,b):
    if sys.version_info.major >=3 and sys.version_info.minor >=8: copytree(a, b, dirs_exist_ok=True)
    else:
        from distutils.dir_util import copy_tree
        copy_tree(a, b)

# %% ../nbs/api/14_quarto.ipynb #caeaa153
def _save_cached_readme(cache, cfg):
    tmp_doc_path = cache/cfg.doc_path.name
    readme = tmp_doc_path/'README.md'
    if readme.exists():
        readme_path = cfg.config_path/'README.md'
        if readme_path.exists(): readme_path.unlink() # py37 doesn't have `missing_ok`
        move(readme, cfg.config_path)
        _rdmi = tmp_doc_path/((cache/cfg.readme_nb).stem + '_files') # Supporting files for README
        if _rdmi.exists(): _copytree(_rdmi, cfg.config_path/_rdmi.name)

# %% ../nbs/api/14_quarto.ipynb #45d6bb5d
@call_parse
def nbdev_readme(
    path:str=None, # Path to notebooks
    chk_time:bool=False): # Only build if out of date
    "Create README.md from readme_nb (index.ipynb by default)"
    cfg = get_config()
    path = Path(path) if path else cfg.nbs_path
    if chk_time and _doc_mtime_not_older(cfg.config_path/'README.md', path/cfg.readme_nb): return

    with _SidebarYmlRemoved(path): # to avoid rendering whole website
        cache = proc_nbs(path)
        _sprun(f'cd "{cache}" && quarto render "{cache/cfg.readme_nb}" -o README.md -t gfm --no-execute')
        
    _save_cached_readme(cache, cfg)

# %% ../nbs/api/14_quarto.ipynb #ef3f1e1f
def _save_cached_contributing(cache, cfg, contrib_nb):
    "Move CONTRIBUTING.md (and any `_files` assets) from the Quarto build cache to the repo root."
    tmp_doc_path = cache / cfg.doc_path.name
    contrib_file = tmp_doc_path / 'CONTRIBUTING.md'
    if contrib_file.exists():
        final_path = cfg.config_path / 'CONTRIBUTING.md'
        if final_path.exists(): final_path.unlink() # py37 doesn't have `missing_ok`
        move(contrib_file, final_path)
        assets_folder = tmp_doc_path / (Path(contrib_nb).stem + '_files') # Supporting files for CONTRIBUTING
        if assets_folder.exists(): _copytree(assets_folder, cfg.config_path / assets_folder.name)

# %% ../nbs/api/14_quarto.ipynb #729a0fa1
@call_parse
def nbdev_contributing(
    path:str=None,  # Path to notebooks
    chk_time:bool=False  # Only build if out-of-date
):
    """Create CONTRIBUTING.md from contributing_nb (defaults to 'contributing.ipynb' if present). Skips if the file doesn't exist."""
    cfg = get_config()
    path = Path(path) if path else cfg.nbs_path
    contrib_nb_name = cfg.get('contributing_nb', 'contributing.ipynb')
    contrib_nb_path = path / contrib_nb_name
    if not contrib_nb_path.exists(): return
    if chk_time and _doc_mtime_not_older(cfg.config_path / 'CONTRIBUTING.md' , contrib_nb_path): return
    
    with _SidebarYmlRemoved(path): # to avoid rendering whole website
        cache = proc_nbs(path)
        _sprun(f'cd "{cache}" && quarto render "{cache/contrib_nb_name}" -o CONTRIBUTING.md -t gfm --no-execute')
        
    _save_cached_contributing(cache, cfg, contrib_nb_name)

# %% ../nbs/api/14_quarto.ipynb #37d16049
@call_parse
@delegates(_nbglob_docs)
def nbdev_docs(
    path:str=None, # Path to notebooks
    n_workers:int=defaults.cpus,  # Number of workers
    **kwargs):
    "Create Quarto docs and README.md"
    cache,cfg,path = _pre_docs(path, n_workers=n_workers, **kwargs)
    nbdev_readme.__wrapped__(path=path, chk_time=True)
    nbdev_contributing.__wrapped__(path=path, chk_time=True)
    _sprun(f'cd "{cache}" && quarto render --no-cache')
    shutil.rmtree(cfg.doc_path, ignore_errors=True)
    move(cache/cfg.doc_path.name, cfg.config_path)

# %% ../nbs/api/14_quarto.ipynb #23886f9c
@call_parse
def prepare():
    "Export, test, and clean notebooks, and render README if needed"
    import nbdev.test, nbdev.clean
    nbdev_export.__wrapped__()
    nbdev.test.nbdev_test.__wrapped__()
    nbdev.clean.nbdev_clean.__wrapped__()
    refresh_quarto_yml()
    nbdev_readme.__wrapped__(chk_time=True)
    nbdev_contributing.__wrapped__(chk_time=True)

# %% ../nbs/api/14_quarto.ipynb #5f701608
@contextmanager
def fs_watchdog(func, path, recursive:bool=True):
    "File system watchdog dispatching to `func`"
    from watchdog.observers import Observer
    from watchdog.events import FileSystemEventHandler
    class _ProcessHandler(FileSystemEventHandler): dispatch=func
    observer = Observer()
    observer.schedule(_ProcessHandler, path, recursive=True)
    observer.start()
    try: yield
    except KeyboardInterrupt: pass
    finally:
        observer.stop()
        observer.join()

# %% ../nbs/api/14_quarto.ipynb #b6ddeadd
@call_parse
@delegates(_nbglob_docs)
def nbdev_preview(
    path:str=None, # Path to notebooks
    port:int=None, # The port on which to run preview
    host:str=None, # The host on which to run preview
    no_browser:bool=False, # Do not open a browser
    n_workers:int=defaults.cpus,  # Number of workers
    **kwargs):
    "Preview docs locally"
    os.environ['QUARTO_PREVIEW']='1'
    cache,cfg,path = _pre_docs(path, n_workers=n_workers, **kwargs)
    xtra = []
    if port: xtra += ['--port', str(port)]
    if host: xtra += ['--host', host]
    if no_browser: xtra += ['--no-browser']

    def _f(e):
        res = _proc_file(Path(e.src_path), cache, path)
        if res:
            try: serve_drv.main(res)
            except: traceback.print_exc()

    os.chdir(cache)
    xtra = xtra or []
    with fs_watchdog(_f, path): subprocess.run(['quarto','preview']+xtra)