lenskit-codex/tasks.py at main · lenskit/lenskit-codex · GitHub

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
import json
import logging
import os
import re
import sys
from glob import glob
from pathlib import Path
from subprocess import check_call

from invoke.context import Context
from invoke.tasks import task

from codex.layout import codex_root
from codex.pages import front_matter, render_templates
from codex.pipeline import (
    CodexPipeline,
    render_dvc_gitignores,
)

cdl = logging.getLogger("codex")
if not logging.root.handlers:
    _cdl_h = logging.StreamHandler(sys.stderr)
    _cdl_h.setLevel(logging.INFO)
    cdl.addHandler(_cdl_h)
    cdl.setLevel(logging.INFO)
logger = logging.getLogger("codex.tasks")

os.chdir(codex_root())


@task
def render_site(c: Context):
    "Render the website."
    c.run("quarto render --to html")


@task
def upload_web_assets(c: Context):
    "Upload the web assets."
    print("pushing static images")
    check_call(["dvc", "push", "--no-run-cache", "-r", "public", "-R", "images"])
    print("pushing page outputs")
    check_call(["dvc", "push", "--no-run-cache", "-r", "public", "dvc.yaml"])


@task
def fetch_web_assets(c: Context):
    "Fetch the web assets."
    c.run("dvc pull --no-run-cache -r public dvc.yaml")
    c.run("dvc pull --no-run-cache -r public -R images")


@task
def render_page_templates(c: Context, include: str | None = None):
    "Render page templates."
    pipeline = CodexPipeline.instance()
    pipeline.scan()
    for pipe_dir, pipe in pipeline.defs():
        ds_dir = Path(pipe_dir)
        if pipe.page_templates:
            assert pipe.info is not None, "no info for pipeline"
            print("rendering templates for", ds_dir)
            render_templates(pipe.info, ds_dir / pipe.page_templates, ds_dir, include)


@task(render_page_templates)
def list_documents(c: Context):
    "List documents with their metadata."
    docs = glob("**/*.qmd", recursive=True)
    print("collecting", len(docs), "documents")
    docs = {name: front_matter(name) for name in sorted(docs) if not re.match(r".*/_", name)}

    with open("manifests/documents.json", "wt") as jsf:
        json.dump(docs, jsf, indent=2)
        print(file=jsf)

    CodexPipeline.instance().reset()


@task
def list_models(c: Context):
    "List the available recommendation models."
    from codex.config import get_config
    from codex.models import discover_models, load_model

    config = get_config()

    models = {}
    for mod_name in discover_models():
        mod = load_model(mod_name)
        models[mod.name] = {
            "src_path": f"src/codex/models/{mod.module_name}.py",
            "predictor": mod.is_predictor,
            "searchable": bool(mod.search_space),
        }
        models[mod.name].update(mod.options)
        for rule in config.models.include:
            if rule.matches_model(mod.name):
                if "ds_include" not in models[mod.name]:
                    models[mod.name]["ds_include"] = []
                models[mod.name]["ds_include"].append(rule.data)

    models = {k: models[k] for k in sorted(models.keys())}

    with open("manifests/models.json", "wt") as jsf:
        json.dump(models, jsf, indent=2)
        print(file=jsf)

    CodexPipeline.instance().reset()


@task(list_documents, list_models)
def render_pipeline(c: Context):
    pipeline = CodexPipeline.instance()
    pipeline.scan()

    for path, pipe in pipeline.defs():
        pipe.render()


@task(render_pipeline)
def update_gitignore(c):
    render_dvc_gitignores()


@task(update_gitignore)
def rerender(c):
    print("Project layout updated")