from bopytex.tasks import Task, activate_corr_on, compile_pdf, generate, join_pdfs class PlannerMissingOption(Exception): pass def naming_template2source(template: str, metadatas: dict): return metadatas["number"] + template[3:] def naming_corr(source): return "corr_" + source def naming_source2pdf(source): return source[:-4] + ".pdf" def naming_join(template): return naming_source2pdf("joined" + template[3:]) def default_planner( options: dict, ) -> list[Task]: opt = { "corr": False, "no_join": False, "no_pdf": False, } opt.update(options) try: template = opt["template"] subjects = opt["subjects"] corr = opt["corr"] no_join = opt["no_join"] no_pdf = opt["no_pdf"] except KeyError: raise PlannerMissingOption("An option is missing") tasks = [] pdfs = [] corr_pdfs = [] for subject in subjects: source = naming_template2source(template, subject) tasks.append(generate(template, subject, source)) if not no_pdf: pdf = naming_source2pdf(source) tasks.append(compile_pdf(source, pdf)) pdfs.append(pdf) if corr: corr_source = naming_corr(source) tasks.append(activate_corr_on(source, corr_source)) if not no_pdf: corr_pdf = naming_source2pdf(corr_source) tasks.append(compile_pdf(corr_source, corr_pdf)) corr_pdfs.append(corr_pdf) if not no_join: joined = naming_join(template) if pdfs: tasks.append(join_pdfs(pdfs, joined)) if corr_pdfs: corr_joined = naming_corr(joined) tasks.append(join_pdfs(corr_pdfs, corr_joined)) return tasks def only_corr_planner( options: dict, ) -> list[Task]: opt = { "no_join": False, "no_pdf": False, } opt.update(options) try: sources = opt["sources"] no_join = opt["no_join"] no_pdf = opt["no_pdf"] except KeyError: raise PlannerMissingOption("An option is missing") tasks = [] corr_pdfs = [] for source in sources: corr_source = naming_corr(source) tasks.append(activate_corr_on(source, corr_source)) if not no_pdf: corr_pdf = naming_source2pdf(corr_source) tasks.append(compile_pdf(corr_source, corr_pdf)) corr_pdfs.append(corr_pdf) if not no_join: joined = "joined.pdf" if corr_pdfs: corr_joined = naming_corr(joined) tasks.append(join_pdfs(corr_pdfs, corr_joined)) return tasks