112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
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:
|
|
logger.warning(f"Lilypond process failed with return code: {proc.returncode}")
|
|
log_stream(stdout, "stdout")
|
|
log_stream(stderr, "stderr")
|
|
errors = await _build_errors(input_file=input_file, stderr=stderr)
|
|
return Result(error=errors)
|
|
|
|
return Result(result=output_file)
|
|
|
|
|
|
def log_stream(stream: bytes, name: str):
|
|
stream_lines = stream.decode("utf-8").splitlines()
|
|
for line in stream_lines:
|
|
logger.warning(f"[Lilypond {name}] {line}")
|
|
|
|
|
|
async def _build_errors(input_file: Path, stderr: bytes) -> list[RenderError]:
|
|
stderr_lines = stderr.decode("utf-8").splitlines()
|
|
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
|