Rendering code refactor

folkugat_web/services/files.py

@@ -3,8 +3,10 @@ import os
 import re
 import uuid
 from collections.abc import Iterator
+from contextlib import asynccontextmanager
 from pathlib import Path
 
+import aiofiles
 import magic
 from fastapi import HTTPException, UploadFile
 from folkugat_web.config import db
@@ -57,8 +59,8 @@ async def store_file(tema_id: int, upload_file: UploadFile) -> str:
 
 def create_tema_filename(tema_id: int, extension: str = "") -> Path:
     filename = str(uuid.uuid4().hex) + extension
-    filedir = db.DB_FILES_DIR / str(tema_id)
-    filedir.mkdir(exist_ok=True)
+    filedir = db.DB_FILES_DIR / "tema" / str(tema_id)
+    filedir.mkdir(parents=True, exist_ok=True)
     filepath = filedir / filename
     return filepath
 
@@ -71,6 +73,18 @@ def create_tmp_filename(extension: str = "") -> Path:
     return filepath
 
 
+@asynccontextmanager
+async def tmp_file(content: str):
+    input_filename = create_tmp_filename(extension=".ly")
+    async with aiofiles.open(input_filename, "w") as f:
+        _ = await f.write(content)
+    try:
+        yield input_filename
+    finally:
+        if input_filename.exists():
+            os.remove(input_filename)
+
+
 def list_files(tema_id: str) -> list[str]:
     filedir = db.DB_FILES_DIR / str(tema_id)
     return [get_db_file_path(f) for f in filedir.iterdir()]

folkugat_web/services/lilypond.py

@@ -1,89 +0,0 @@
-import asyncio
-import dataclasses
-import os
-import re
-from pathlib import Path
-from typing import Literal
-
-import aiofiles
-from folkugat_web.model.lilypond import (LyricsLine, LyricsParagraph,
-                                         LyricsText, RenderError)
-from folkugat_web.model.temes import Tema
-from folkugat_web.services import files as files_service
-from folkugat_web.templates import templates
-from folkugat_web.utils import batched
-
-SCORE_BEGINNING = "% --- SCORE BEGINNING --- %"
-
-RenderFormat = Literal["png"] | Literal["pdf"]
-
-
-def source_to_single_score(source: str, tema: Tema) -> str:
-    if tema.lyrics:
-        lyrics_text = build_lyrics(tema.lyrics[0].content)
-    else:
-        lyrics_text = None
-    print("AAAAAA")
-    print(lyrics_text)
-    print("AAAAAA")
-    return templates.get_template("lilypond/single_score.ly").render(
-        score_beginning=SCORE_BEGINNING,
-        score_source=source,
-        tema=tema,
-        lyrics_text=lyrics_text,
-    )
-
-
-async def render(source: str, fmt: RenderFormat, output_filename: Path | None = None) -> tuple[Path | None, list[RenderError]]:
-    input_filename = files_service.create_tmp_filename(extension=".ly")
-    async with aiofiles.open(input_filename, "w") as f:
-        _ = await f.write(source)
-
-    output_filename = output_filename or files_service.create_tmp_filename()
-
-    if fmt == "png":
-        proc = await asyncio.create_subprocess_exec(
-            "lilypond", "--png", "-o", output_filename, input_filename,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-    else:
-        proc = await asyncio.create_subprocess_exec(
-            "lilypond", "-o", output_filename, input_filename,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
-        )
-
-    _stdout, stderr = await proc.communicate()
-
-    if proc.returncode:
-        esc_input_filename = re.escape(str(input_filename))
-        error_re = re.compile(rf"{esc_input_filename}:(?P<line>\d+):(?P<pos>\d+): error: (?P<error>.*)$")
-        error_matches = filter(None, map(error_re.match, stderr.decode("utf-8").splitlines()))
-        errors = map(lambda m: RenderError.from_dict(m.groupdict()), error_matches)
-        line_offset = source.splitlines().index(SCORE_BEGINNING) + 1
-        errors_offset = [dataclasses.replace(e, line=e.line - line_offset) for e in errors if e.line > line_offset]
-        return None, errors_offset
-
-    if fmt == "png":
-        # Check if the output generated multiple pages
-        pages = sorted(output_filename.parent.rglob(f"{output_filename.name}*.png"))
-        if len(pages) > 1:
-            output_filename = pages[0]
-            for page in pages[1:]:
-                os.remove(page)
-        else:
-            output_filename = output_filename.with_suffix(".png")
-    elif fmt == "pdf":
-        output_filename = output_filename.with_suffix(".pdf")
-
-    return output_filename, []
-
-
-def build_lyrics(text: str, max_cols: int = 3) -> LyricsText:
-    paragraphs = [LyricsParagraph(lines=par_str.splitlines()) for par_str in text.split("\n\n") if par_str]
-    print(list(batched(paragraphs, max_cols)))
-    return LyricsText(lines=[
-        LyricsLine(paragraphs=list(line_pars))
-        for line_pars in batched(paragraphs, max_cols)
-    ])

folkugat_web/services/lilypond/build.py

@@ -0,0 +1,102 @@
+from collections.abc import Iterable, Iterator
+
+from fastapi import HTTPException
+from folkugat_web.dal.sql.temes import scores as scores_dal
+from folkugat_web.model import playlists as playlists_model
+from folkugat_web.model import temes as model
+from folkugat_web.model.lilypond import score as lilypond_model
+from folkugat_web.services import lilypond
+from folkugat_web.services.temes import lyrics as lyrics_service
+from folkugat_web.services.temes import properties as properties_service
+from folkugat_web.services.temes import query as temes_q
+from folkugat_web.services.temes import scores as scores_service
+from folkugat_web.utils import FnChain
+
+UNKNOWN_TITLE = "Desconegut"
+
+
+def unknown_tune() -> lilypond_model.LilypondTune:
+    return lilypond_model.LilypondTune(
+        header=lilypond_model.HeaderData(
+            title=UNKNOWN_TITLE,
+            composer=None,
+        ),
+        score_source=None,
+        lyrics=None,
+        is_unknown=True,
+    )
+
+
+def tune_from_tema(tema: model.Tema | None, score_source: str | None = None) -> lilypond_model.LilypondTune:
+    """
+    The given `tema` is assumed to have properties, lyrics and scores (if source is None)
+    """
+    if tema is None:
+        return unknown_tune()
+    return lilypond_model.LilypondTune(
+        header=lilypond_model.HeaderData.from_tema(tema=tema),
+        score_source=score_source or (tema.scores[0].source if tema.scores else None),
+        lyrics=lilypond_model.LyricsText.from_lyrics(lyrics=tema.lyrics[0]) if tema.lyrics else None,
+    )
+
+
+def tunes_from_temes(temes: Iterable[model.Tema | None]) -> list[lilypond_model.LilypondTune]:
+    """
+    All `Tema` in `temes` are assumed to have properties, lyrics and scores
+    """
+    return [tune_from_tema(tema) for tema in temes]
+
+
+def tune_from_tema_id(tema_id: int, score_source: str | None = None) -> lilypond_model.LilypondTune:
+    tema = temes_q.get_tema_by_id(tema_id)
+    if not tema:
+        raise HTTPException(status_code=404, detail="Could not find tema!")
+    tema = (
+        FnChain.transform(tema) |
+        properties_service.add_properties_to_tema |
+        lyrics_service.add_lyrics_to_tema
+    ).result()
+    if score_source is None:
+        tema = scores_service.add_scores_to_tema(tema)
+    return tune_from_tema(tema=tema, score_source=score_source)
+
+
+def tunes_from_tema_ids(tema_ids: list[int]) -> list[lilypond_model.LilypondTune]:
+    temes = (
+        FnChain.transform(temes_q.get_temes_by_ids(tema_ids=tema_ids)) |
+        properties_service.add_properties_to_temes |
+        lyrics_service.add_lyrics_to_temes |
+        scores_service.add_scores_to_temes |
+        list
+    ).result()
+    return tunes_from_temes(temes=temes)
+
+
+def build_set_title(temes: list[model.Tema | None]) -> str:
+    def build_title(tema: model.Tema | None) -> str:
+        return tema.title if tema else UNKNOWN_TITLE
+    return " i ".join(filter(bool, [
+        ", ".join([build_title(tema) for tema in temes[:-1]]),
+        build_title(temes[-1])
+    ]))
+
+
+def set_from_set(set_entry: playlists_model.Set) -> lilypond_model.LilypondSet:
+    def _build_temes_by_id(temes: Iterable[model.Tema]) -> dict[int, model.Tema]:
+        return {tema.id: tema for tema in temes if tema.id is not None}
+    tema_ids = [tema_in_set.tema_id for tema_in_set in set_entry.temes]
+    temes_by_id = (
+        FnChain.transform([tema_id for tema_id in tema_ids if tema_id is not None]) |
+        temes_q.get_temes_by_ids |
+        properties_service.add_properties_to_temes |
+        lyrics_service.add_lyrics_to_temes |
+        scores_service.add_scores_to_temes |
+        _build_temes_by_id
+    ).result()
+    temes = [temes_by_id[tema_id] if tema_id is not None else None for tema_id in tema_ids]
+    set_title = build_set_title(temes=temes)
+    tunes = tunes_from_temes(temes)
+    return lilypond_model.LilypondSet(
+        title=set_title,
+        tunes=tunes
+    )

folkugat_web/services/lilypond/render.py

@@ -0,0 +1,105 @@
+import asyncio
+import dataclasses
+import enum
+import re
+from pathlib import Path
+
+import aiofiles
+from folkugat_web.log import logger
+from folkugat_web.model.lilypond.processing import RenderError
+from folkugat_web.services import files as files_service
+from folkugat_web.utils import Result
+
+from .source import SCORE_BEGINNING
+
+RenderResult = Result[Path, list[RenderError]]
+
+
+class RenderOutput(enum.Enum):
+    PDF = "pdf"
+    PNG_CROPPED = "png-cropped"
+    PNG_PREVIEW = "png-preview"
+
+
+async def render(
+    source: str,
+    output: RenderOutput,
+    output_file: Path | None = None,
+) -> RenderResult:
+    async with files_service.tmp_file(content=source) as input_file:
+        return await render_file(
+            input_file=input_file,
+            output=output,
+            output_file=output_file,
+        )
+
+
+async def render_file(
+    input_file: Path,
+    output: RenderOutput,
+    output_file: Path | None = None,
+) -> RenderResult:
+    output_file = output_file or files_service.create_tmp_filename()
+    match output:
+        case RenderOutput.PDF:
+            command = [
+                "lilypond",
+                "-f", "pdf",
+                "-o", str(output_file),
+                str(input_file)
+            ]
+            output_file = output_file.with_suffix(".pdf")
+        case RenderOutput.PNG_CROPPED:
+            command = [
+                "lilypond",
+                "-f", "png",
+                "-dcrop=#t",
+                "-dno-print-pages",
+                # "-duse-paper-size-for-page=#f",
+                "-o", str(output_file),
+                str(input_file)
+            ]
+            output_file = output_file.with_suffix(".cropped.png")
+            # output_file = output_file.with_suffix(".png")
+        case RenderOutput.PNG_PREVIEW:
+            command = [
+                "lilypond",
+                "-f", "png",
+                "-dpreview=#t",
+                "-dno-print-pages",
+                "-o", str(output_file),
+                str(input_file)
+            ]
+            output_file = output_file.with_suffix(".preview.png")
+
+    proc = await asyncio.create_subprocess_exec(
+        *command,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+
+    _stdout, stderr = await proc.communicate()
+
+    if proc.returncode:
+        errors = await _build_errors(input_file=input_file, stderr=stderr)
+        return Result(error=errors)
+
+    return Result(result=output_file)
+
+
+async def _build_errors(input_file: Path, stderr: bytes) -> list[RenderError]:
+    stderr_lines = stderr.decode("utf-8").splitlines()
+    logger.warning("Lilypond errors:")
+    for line in stderr_lines:
+        logger.warning(line)
+    async with aiofiles.open(input_file, "r") as f:
+        lines = await f.readlines()
+
+    offset_mark = SCORE_BEGINNING + "\n"
+    line_offset = lines.index(offset_mark) + 1 if offset_mark in lines else 0
+    esc_input_file = re.escape(str(input_file))
+    error_re = re.compile(rf"{esc_input_file}:(?P<line>\d+):(?P<pos>\d+): error: (?P<error>.*)$")
+    error_matches = filter(None, map(error_re.match, stderr_lines))
+    errors = map(lambda m: RenderError.from_dict(m.groupdict()), error_matches)
+    errors_offset = [dataclasses.replace(e, line=e.line - line_offset) for e in errors if e.line > line_offset]
+    return errors_offset

folkugat_web/services/lilypond/source.py

@@ -0,0 +1,18 @@
+from folkugat_web.model.lilypond.score import LilypondSet, LilypondTune
+from folkugat_web.templates import templates
+
+SCORE_BEGINNING = "% --- SCORE BEGINNING --- %"
+
+
+def tune_source(tune: LilypondTune) -> str:
+    return templates.get_template("lilypond/single_tune.ly").render(
+        score_beginning=SCORE_BEGINNING,
+        tune=tune,
+    )
+
+
+def set_source(tune_set: LilypondSet) -> str:
+    return templates.get_template("lilypond/tune_set.ly").render(
+        score_beginning=SCORE_BEGINNING,
+        tune_set=tune_set,
+    )

folkugat_web/services/temes/query.py

@@ -5,7 +5,11 @@ from folkugat_web.model import temes as model
 
 
 def get_tema_by_id(tema_id: int) -> model.Tema | None:
-    return temes_q.get_tema_by_id(tema_id)
+    return temes_q.get_tema_by_id(tema_id=tema_id)
+
+
+def get_temes_by_ids(tema_ids: list[int]) -> list[model.Tema]:
+    return temes_q.get_temes_by_ids(tema_ids=tema_ids)
 
 
 def tema_compute_stats(

folkugat_web/services/temes/scores.py

@@ -3,6 +3,7 @@ from collections.abc import Iterable, Iterator
 from fastapi import HTTPException
 from folkugat_web.dal.sql.temes import scores as scores_dal
 from folkugat_web.model import temes as model
+from folkugat_web.model.lilypond import score as lilypond_model
 from folkugat_web.services import lilypond
 from folkugat_web.services.temes import lyrics as lyrics_service
 from folkugat_web.services.temes import properties as properties_service
@@ -46,13 +47,29 @@ def create_score(tema_id: int) -> model.Score:
     return scores_dal.insert_score(score=new_score)
 
 
-def build_single_tune_full_source(tema_id: int, source: str) -> str:
-    tema = temes_q.get_tema_by_id(tema_id)
-    if not tema:
-        raise HTTPException(status_code=404, detail="Could not find tema!")
-    tema = (
-        FnChain.transform(tema) |
-        properties_service.add_properties_to_tema |
-        lyrics_service.add_lyrics_to_tema
+def build_tune_set_full_source(tema_ids: list[int]) -> str:
+    temes = (
+        FnChain.transform(temes_q.get_temes_by_ids(tema_ids=tema_ids)) |
+        properties_service.add_properties_to_temes |
+        lyrics_service.add_lyrics_to_temes |
+        add_scores_to_temes |
+        list
     ).result()
-    return lilypond.source_to_single_score(source=source, tema=tema)
+    if not temes:
+        return ""
+    set_title = " i ".join(filter(bool, [
+        ", ".join([tema.title for tema in temes[:-1]]),
+        temes[-1].title
+    ]))
+    tune_set = lilypond_model.LilypondSet(
+        title=set_title,
+        tunes=[
+            lilypond_model.LilypondTune(
+                header=lilypond_model.HeaderData.from_tema(tema=tema),
+                score_source=tema.scores[0].source if tema.scores else None,
+                lyrics=lilypond_model.LyricsText.from_lyrics(lyrics=tema.lyrics[0]) if tema.lyrics else None,
+            ) for tema in temes
+        ]
+    )
+    print("TUNE SET HASH: ", tune_set.hash().hex())
+    return lilypond.tune_set_score(tune_set=tune_set)