Lead Sheet Reader endpoint source (AGPL-3.0), from private repo commit 840bdf0e

6a1e48b verified 14 days ago

12.4 kB

"""Hugging Face Inference Endpoints custom handler: lead sheet image -> MusicXML with chord symbols.
Model: JuanCarlosMartinezSevilla/jazzmus-model (MIT): a YOLO staff detector, then the SMT
image-to-**kern model on each staff. The detector runs as ONNX on onnxruntime (MIT) with the
letterbox, box decoding and NMS below in numpy; the ultralytics package is not used. The
staff-cropping functions are copied from ../predict.py (2026-10-18 fixes: STAFF_PAD,
select_staff_boxes) and work on an in-memory image.
Errors are raised as EndpointError with a fixed short code. The Inference Toolkit turns any
exception from a handler into HTTP 400 with body {"error": str(exception)}; a handler cannot
choose the status code (see README.md).
"""
import base64
import binascii
import io
import logging
import os
import threading
from pathlib import Path

os.environ.setdefault("HF_HUB_OFFLINE", "1")  # load nothing from the network

import numpy as np  # noqa: E402
import onnxruntime as ort  # noqa: E402
import torch  # noqa: E402
from PIL import Image  # noqa: E402
from safetensors.torch import load_file  # noqa: E402
from torch.nn import Conv1d  # noqa: E402

from jazzmus_smt.configuration_smt import SMTConfig  # noqa: E402
from jazzmus_smt.modeling_smt import SMTModelForCausalLM  # noqa: E402
from kern2musicxml import kern_to_musicxml  # noqa: E402

log = logging.getLogger(__name__)

TORCH_THREADS = 2
MAX_BYTES = 760 * 1024
MAX_B64_CHARS = 4 * ((MAX_BYTES + 2) // 3)
MAX_SIDE = 6000  # refused above this
# Refused above this many pixels (6000 x 4000): a well-compressed page under MAX_BYTES can decode to
# hundreds of MB, so the bound is on pixels, not bytes or sides.
MAX_PIXELS = 24_000_000
# One decode at a time, whatever the toolkit's request concurrency: each holds up to a few hundred MB
# (four concurrent 6000 x 6000 decodes measured +1.5 GB on the endpoint's 4 GB).
DECODE_LOCK = threading.Lock()
WORK_SIDE = 3000  # larger images are scaled down before detection, as in app.py
MAX_STAVES = 12
CONTENT_TYPES = {"image/png": "PNG", "image/jpeg": "JPEG", "image/webp": "WEBP"}

# ---- YOLO staff detector on onnxruntime ------------------------------------------------------
# The same steps and defaults as ultralytics 8.4 predict on the .pt model: a rectangular letterbox
# (long side 640, padded to a multiple of 32 with grey 114), class-wise NMS at IoU 0.7 over boxes
# with confidence >= 0.25, at most 300, then boxes scaled back to the image.
DETECT_SIZE = 640
STRIDE = 32
DETECT_CONF = 0.25
NMS_IOU = 0.7
MAX_DET = 300
STAFF_CLASS = 12  # "staff" in the model's names (stored in the .onnx metadata)


def resize_linear(image: Image.Image, size: tuple[int, int]) -> Image.Image:
    """Bilinear resize without antialiasing, sampling at pixel centres as cv2.INTER_LINEAR does.
    The detector and the SMT model were used with cv2.resize; PIL's resize() antialiases when it
    shrinks, which moved staff boxes by up to 18 px. An affine transform samples like cv2.
    """
    if size == image.size:
        return image
    sx, sy = image.width / size[0], image.height / size[1]
    return image.transform(size, Image.AFFINE, (sx, 0, 0, 0, sy, 0), resample=Image.BILINEAR)


def letterbox(image: Image.Image) -> tuple[np.ndarray, tuple[float, float], tuple[int, int]]:
    """RGB image -> (1, 3, H, W) float32 input, (gain_x, gain_y), (pad_x, pad_y)."""
    w, h = image.size
    r = min(DETECT_SIZE / h, DETECT_SIZE / w)
    new_w, new_h = round(w * r), round(h * r)
    dw, dh = (DETECT_SIZE - new_w) % STRIDE / 2, (DETECT_SIZE - new_h) % STRIDE / 2
    top, bottom, left, right = round(dh - 0.1), round(dh + 0.1), round(dw - 0.1), round(dw + 0.1)
    canvas = Image.new("RGB", (new_w + left + right, new_h + top + bottom), (114, 114, 114))
    canvas.paste(resize_linear(image, (new_w, new_h)), (left, top))
    x = np.asarray(canvas, dtype=np.float32).transpose(2, 0, 1)[None] / 255.0
    return x, (new_w / w, new_h / h), (left, top)


def nms(boxes: np.ndarray, scores: np.ndarray, iou: float) -> list[int]:
    order, keep = scores.argsort()[::-1], []
    areas = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])
    while order.size:
        i, rest = order[0], order[1:]
        keep.append(int(i))
        xx1 = np.maximum(boxes[i, 0], boxes[rest, 0])
        yy1 = np.maximum(boxes[i, 1], boxes[rest, 1])
        xx2 = np.minimum(boxes[i, 2], boxes[rest, 2])
        yy2 = np.minimum(boxes[i, 3], boxes[rest, 3])
        inter = np.clip(xx2 - xx1, 0, None) * np.clip(yy2 - yy1, 0, None)
        order = rest[inter / (areas[i] + areas[rest] - inter + 1e-9) <= iou]
    return keep


def detect(session: ort.InferenceSession, image: Image.Image) -> list[tuple[float, int, tuple[float, float, float, float]]]:
    """All detections as (confidence, class, (x1, y1, x2, y2)) in image pixels."""
    x, (gx, gy), (px, py) = letterbox(image)
    pred = session.run(None, {"images": x})[0][0].T  # (anchors, 4 + classes): cx, cy, w, h, scores
    cls = pred[:, 4:].argmax(1)
    conf = pred[np.arange(len(pred)), 4 + cls]
    m = conf >= DETECT_CONF
    pred, cls, conf = pred[m], cls[m], conf[m]
    boxes = np.stack([pred[:, 0] - pred[:, 2] / 2, pred[:, 1] - pred[:, 3] / 2,
                      pred[:, 0] + pred[:, 2] / 2, pred[:, 1] + pred[:, 3] / 2], 1)
    keep = nms(boxes + cls[:, None] * 7680.0, conf, NMS_IOU)[:MAX_DET]  # offset by class: class-wise NMS
    w, h = image.size
    out = []
    for i in keep:
        x1, y1, x2, y2 = boxes[i]
        out.append((float(conf[i]), int(cls[i]), (
            float(np.clip((x1 - px) / gx, 0, w)), float(np.clip((y1 - py) / gy, 0, h)),
            float(np.clip((x2 - px) / gx, 0, w)), float(np.clip((y2 - py) / gy, 0, h)))))
    return out


# ---- copied from ../predict.py ------------------------------------------------------------
FIXED_IMG_HEIGHT = 128
MAX_IMG_WIDTH = 1000
STAFF_PAD = 0.30
MIN_CONFIDENCE = 0.5
MAX_OVERLAP = 0.5


def _vertical_overlap(a: tuple[int, int, int, int], b: tuple[int, int, int, int]) -> float:
    inter = min(a[3], b[3]) - max(a[1], b[1])
    return max(0, inter) / min(a[3] - a[1], b[3] - b[1])


def select_staff_boxes(detections: list[tuple[float, tuple[int, int, int, int]]]) -> list[tuple[int, int, int, int]]:
    """Keep confident boxes, best first, dropping any that overlap a kept one; return them top to bottom."""
    kept: list[tuple[int, int, int, int]] = []
    for conf, box in sorted(detections, key=lambda d: -d[0]):
        if conf < MIN_CONFIDENCE:
            continue
        if any(_vertical_overlap(box, k) > MAX_OVERLAP for k in kept):
            continue
        kept.append(box)
    return sorted(kept, key=lambda b: b[1])


def detect_staves(image: Image.Image, session: ort.InferenceSession) -> list[Image.Image]:
    """predict.detect_staves on an RGB PIL image, with the ONNX detector; returns greyscale crops."""
    gray = image.convert("L")
    detections = [(conf, tuple(map(int, box))) for conf, cls, box in detect(session, image) if cls == STAFF_CLASS]

    crops = []
    for x1, y1, x2, y2 in select_staff_boxes(detections)[:MAX_STAVES]:
        pad = int(round(STAFF_PAD * (y2 - y1)))
        crops.append(gray.crop((x1, max(0, y1 - pad), x2, min(gray.height, y2 + pad))))
    return crops


def preprocess_staff(img: Image.Image) -> torch.Tensor:
    width = int(np.ceil(img.width * FIXED_IMG_HEIGHT / img.height))
    width = min(width, MAX_IMG_WIDTH)
    img = resize_linear(img, (width, FIXED_IMG_HEIGHT))
    # jazzmus convert_img_to_tensor: greyscale, (1, H, W) float in [0, 1]
    return torch.from_numpy(np.asarray(img, dtype=np.float32) / 255.0)[None]


@torch.no_grad()
def transcribe_staff(model: SMTModelForCausalLM, image_tensor: torch.Tensor) -> str:
    predicted_sequence, _ = model.predict(input=image_tensor, convert_to_str=True)
    # jazzmus.dataset.tokenizer.untokenize
    return (
        "".join(predicted_sequence).replace("<t>", "\t").replace("<n>", "\n").replace("<s>", " ")
        .replace("<chord-pitch>", "").replace("<chord-extension>", "")
    )


def _sync_pe_device(model: SMTModelForCausalLM, device: torch.device):
    model.positional_2D.pe = model.positional_2D.pe.to(device)
    model.decoder.positional_1D.pe = model.decoder.positional_1D.pe.to(device)
# ---- end of copy ----------------------------------------------------------------------------


class EndpointError(ValueError):
    """A client-facing failure; str() is only the fixed code, never request content."""


MAGIC = {
    "image/png": lambda b: b.startswith(b"\x89PNG\r\n\x1a\n"),
    "image/jpeg": lambda b: b.startswith(b"\xff\xd8\xff"),
    "image/webp": lambda b: len(b) >= 12 and b[:4] == b"RIFF" and b[8:12] == b"WEBP",
}


def decode_image(data) -> Image.Image:
    """Validate the request body and return the image as RGB; raise EndpointError with a fixed code."""
    if not isinstance(data, dict):
        raise EndpointError("invalid_request")
    inputs = data.get("inputs")
    params = data.get("parameters") or {}
    if not isinstance(inputs, str) or not isinstance(params, dict):
        raise EndpointError("invalid_request")
    content_type = params.get("contentType")
    if content_type not in CONTENT_TYPES:
        raise EndpointError("unsupported_content_type")
    if len(inputs) > MAX_B64_CHARS:
        raise EndpointError("image_too_large")
    try:
        raw = base64.b64decode(inputs, validate=True)
    except (binascii.Error, ValueError):
        raise EndpointError("invalid_base64") from None
    if len(raw) > MAX_BYTES:
        raise EndpointError("image_too_large")
    if not MAGIC[content_type](raw):
        raise EndpointError("content_type_mismatch")
    with DECODE_LOCK:
        try:
            with Image.open(io.BytesIO(raw), formats=[CONTENT_TYPES[content_type]]) as im:
                # The header's size, checked before any pixel is decoded.
                if max(im.size) > MAX_SIDE or im.width * im.height > MAX_PIXELS:
                    raise EndpointError("image_dimensions_too_large")
                im.load()
                rgba = im.convert("RGBA")
        except EndpointError:
            raise
        except Exception:
            raise EndpointError("invalid_image") from None
        # Transparent areas become white paper, not black.
        image = Image.new("RGB", rgba.size, "white")
        image.paste(rgba, mask=rgba.getchannel("A"))
        del rgba
        image.thumbnail((WORK_SIDE, WORK_SIDE))
        return image


class EndpointHandler:
    def __init__(self, path: str = ""):
        torch.set_num_threads(TORCH_THREADS)
        model_dir = Path(path)
        self.device = torch.device("cpu")
        opts = ort.SessionOptions()
        opts.intra_op_num_threads = TORCH_THREADS
        opts.inter_op_num_threads = 1
        self.detector = ort.InferenceSession(
            str(model_dir / "yolo_staff_detector.onnx"), opts, providers=["CPUExecutionProvider"]
        )

        config = SMTConfig.from_pretrained(str(model_dir))
        sd = load_file(str(model_dir / "model.safetensors"))
        embed_size = sd["decoder.embedding.weight"].shape[0]
        out_size = sd["decoder.out_layer.weight"].shape[0]
        orig_out = config.out_categories
        config.out_categories = embed_size
        model = SMTModelForCausalLM(config)
        if embed_size != out_size:
            model.decoder.out_layer = Conv1d(config.d_model, out_size, kernel_size=1)
        model.load_state_dict(sd, strict=True)
        config.out_categories = orig_out
        model.to(self.device)
        _sync_pe_device(model, self.device)
        model.eval()
        self.smt = model
        self.lock = threading.Lock()  # one page at a time on the two cores

    def __call__(self, data: dict) -> dict:
        image = decode_image(data)
        try:
            with self.lock:
                staves = detect_staves(image, self.detector)
                if not staves:
                    raise EndpointError("no_staff_found")
                kerns = [transcribe_staff(self.smt, preprocess_staff(s).to(self.device)) for s in staves]
            musicxml, _ = kern_to_musicxml(kerns)
        except EndpointError:
            raise
        except Exception:
            log.exception("transcription failed")  # the traceback holds no request content
            raise EndpointError("transcription_failed") from None
        return {"musicxml": musicxml, "staves": len(staves)}

	"""Hugging Face Inference Endpoints custom handler: lead sheet image -> MusicXML with chord symbols.

	Model: JuanCarlosMartinezSevilla/jazzmus-model (MIT): a YOLO staff detector, then the SMT
	image-to-**kern model on each staff. The detector runs as ONNX on onnxruntime (MIT) with the
	letterbox, box decoding and NMS below in numpy; the ultralytics package is not used. The
	staff-cropping functions are copied from ../predict.py (2026-10-18 fixes: STAFF_PAD,
	select_staff_boxes) and work on an in-memory image.

	Errors are raised as EndpointError with a fixed short code. The Inference Toolkit turns any
	exception from a handler into HTTP 400 with body {"error": str(exception)}; a handler cannot
	choose the status code (see README.md).
	"""
	import base64
	import binascii
	import io
	import logging
	import os
	import threading
	from pathlib import Path

	os.environ.setdefault("HF_HUB_OFFLINE", "1") # load nothing from the network

	import numpy as np # noqa: E402
	import onnxruntime as ort # noqa: E402
	import torch # noqa: E402
	from PIL import Image # noqa: E402
	from safetensors.torch import load_file # noqa: E402
	from torch.nn import Conv1d # noqa: E402

	from jazzmus_smt.configuration_smt import SMTConfig # noqa: E402
	from jazzmus_smt.modeling_smt import SMTModelForCausalLM # noqa: E402
	from kern2musicxml import kern_to_musicxml # noqa: E402

	log = logging.getLogger(__name__)

	TORCH_THREADS = 2
	MAX_BYTES = 760 * 1024
	MAX_B64_CHARS = 4 * ((MAX_BYTES + 2) // 3)
	MAX_SIDE = 6000 # refused above this
	# Refused above this many pixels (6000 x 4000): a well-compressed page under MAX_BYTES can decode to
	# hundreds of MB, so the bound is on pixels, not bytes or sides.
	MAX_PIXELS = 24_000_000
	# One decode at a time, whatever the toolkit's request concurrency: each holds up to a few hundred MB
	# (four concurrent 6000 x 6000 decodes measured +1.5 GB on the endpoint's 4 GB).
	DECODE_LOCK = threading.Lock()
	WORK_SIDE = 3000 # larger images are scaled down before detection, as in app.py
	MAX_STAVES = 12
	CONTENT_TYPES = {"image/png": "PNG", "image/jpeg": "JPEG", "image/webp": "WEBP"}

	# ---- YOLO staff detector on onnxruntime ------------------------------------------------------
	# The same steps and defaults as ultralytics 8.4 predict on the .pt model: a rectangular letterbox
	# (long side 640, padded to a multiple of 32 with grey 114), class-wise NMS at IoU 0.7 over boxes
	# with confidence >= 0.25, at most 300, then boxes scaled back to the image.
	DETECT_SIZE = 640
	STRIDE = 32
	DETECT_CONF = 0.25
	NMS_IOU = 0.7
	MAX_DET = 300
	STAFF_CLASS = 12 # "staff" in the model's names (stored in the .onnx metadata)


	def resize_linear(image: Image.Image, size: tuple[int, int]) -> Image.Image:
	"""Bilinear resize without antialiasing, sampling at pixel centres as cv2.INTER_LINEAR does.

	The detector and the SMT model were used with cv2.resize; PIL's resize() antialiases when it
	shrinks, which moved staff boxes by up to 18 px. An affine transform samples like cv2.
	"""
	if size == image.size:
	return image
	sx, sy = image.width / size[0], image.height / size[1]
	return image.transform(size, Image.AFFINE, (sx, 0, 0, 0, sy, 0), resample=Image.BILINEAR)


	def letterbox(image: Image.Image) -> tuple[np.ndarray, tuple[float, float], tuple[int, int]]:
	"""RGB image -> (1, 3, H, W) float32 input, (gain_x, gain_y), (pad_x, pad_y)."""
	w, h = image.size
	r = min(DETECT_SIZE / h, DETECT_SIZE / w)
	new_w, new_h = round(w * r), round(h * r)
	dw, dh = (DETECT_SIZE - new_w) % STRIDE / 2, (DETECT_SIZE - new_h) % STRIDE / 2
	top, bottom, left, right = round(dh - 0.1), round(dh + 0.1), round(dw - 0.1), round(dw + 0.1)
	canvas = Image.new("RGB", (new_w + left + right, new_h + top + bottom), (114, 114, 114))
	canvas.paste(resize_linear(image, (new_w, new_h)), (left, top))
	x = np.asarray(canvas, dtype=np.float32).transpose(2, 0, 1)[None] / 255.0
	return x, (new_w / w, new_h / h), (left, top)


	def nms(boxes: np.ndarray, scores: np.ndarray, iou: float) -> list[int]:
	order, keep = scores.argsort()[::-1], []
	areas = (boxes[:, 2] - boxes[:, 0]) * (boxes[:, 3] - boxes[:, 1])
	while order.size:
	i, rest = order[0], order[1:]
	keep.append(int(i))
	xx1 = np.maximum(boxes[i, 0], boxes[rest, 0])
	yy1 = np.maximum(boxes[i, 1], boxes[rest, 1])
	xx2 = np.minimum(boxes[i, 2], boxes[rest, 2])
	yy2 = np.minimum(boxes[i, 3], boxes[rest, 3])
	inter = np.clip(xx2 - xx1, 0, None) * np.clip(yy2 - yy1, 0, None)
	order = rest[inter / (areas[i] + areas[rest] - inter + 1e-9) <= iou]
	return keep


	def detect(session: ort.InferenceSession, image: Image.Image) -> list[tuple[float, int, tuple[float, float, float, float]]]:
	"""All detections as (confidence, class, (x1, y1, x2, y2)) in image pixels."""
	x, (gx, gy), (px, py) = letterbox(image)
	pred = session.run(None, {"images": x})[0][0].T # (anchors, 4 + classes): cx, cy, w, h, scores
	cls = pred[:, 4:].argmax(1)
	conf = pred[np.arange(len(pred)), 4 + cls]
	m = conf >= DETECT_CONF
	pred, cls, conf = pred[m], cls[m], conf[m]
	boxes = np.stack([pred[:, 0] - pred[:, 2] / 2, pred[:, 1] - pred[:, 3] / 2,
	pred[:, 0] + pred[:, 2] / 2, pred[:, 1] + pred[:, 3] / 2], 1)
	keep = nms(boxes + cls[:, None] * 7680.0, conf, NMS_IOU)[:MAX_DET] # offset by class: class-wise NMS
	w, h = image.size
	out = []
	for i in keep:
	x1, y1, x2, y2 = boxes[i]
	out.append((float(conf[i]), int(cls[i]), (
	float(np.clip((x1 - px) / gx, 0, w)), float(np.clip((y1 - py) / gy, 0, h)),
	float(np.clip((x2 - px) / gx, 0, w)), float(np.clip((y2 - py) / gy, 0, h)))))
	return out


	# ---- copied from ../predict.py ------------------------------------------------------------
	FIXED_IMG_HEIGHT = 128
	MAX_IMG_WIDTH = 1000
	STAFF_PAD = 0.30
	MIN_CONFIDENCE = 0.5
	MAX_OVERLAP = 0.5


	def _vertical_overlap(a: tuple[int, int, int, int], b: tuple[int, int, int, int]) -> float:
	inter = min(a[3], b[3]) - max(a[1], b[1])
	return max(0, inter) / min(a[3] - a[1], b[3] - b[1])


	def select_staff_boxes(detections: list[tuple[float, tuple[int, int, int, int]]]) -> list[tuple[int, int, int, int]]:
	"""Keep confident boxes, best first, dropping any that overlap a kept one; return them top to bottom."""
	kept: list[tuple[int, int, int, int]] = []
	for conf, box in sorted(detections, key=lambda d: -d[0]):
	if conf < MIN_CONFIDENCE:
	continue
	if any(_vertical_overlap(box, k) > MAX_OVERLAP for k in kept):
	continue
	kept.append(box)
	return sorted(kept, key=lambda b: b[1])


	def detect_staves(image: Image.Image, session: ort.InferenceSession) -> list[Image.Image]:
	"""predict.detect_staves on an RGB PIL image, with the ONNX detector; returns greyscale crops."""
	gray = image.convert("L")
	detections = [(conf, tuple(map(int, box))) for conf, cls, box in detect(session, image) if cls == STAFF_CLASS]

	crops = []
	for x1, y1, x2, y2 in select_staff_boxes(detections)[:MAX_STAVES]:
	pad = int(round(STAFF_PAD * (y2 - y1)))
	crops.append(gray.crop((x1, max(0, y1 - pad), x2, min(gray.height, y2 + pad))))
	return crops


	def preprocess_staff(img: Image.Image) -> torch.Tensor:
	width = int(np.ceil(img.width * FIXED_IMG_HEIGHT / img.height))
	width = min(width, MAX_IMG_WIDTH)
	img = resize_linear(img, (width, FIXED_IMG_HEIGHT))
	# jazzmus convert_img_to_tensor: greyscale, (1, H, W) float in [0, 1]
	return torch.from_numpy(np.asarray(img, dtype=np.float32) / 255.0)[None]


	@torch.no_grad()
	def transcribe_staff(model: SMTModelForCausalLM, image_tensor: torch.Tensor) -> str:
	predicted_sequence, _ = model.predict(input=image_tensor, convert_to_str=True)
	# jazzmus.dataset.tokenizer.untokenize
	return (
	"".join(predicted_sequence).replace("<t>", "\t").replace("<n>", "\n").replace("<s>", " ")
	.replace("<chord-pitch>", "").replace("<chord-extension>", "")
	)


	def _sync_pe_device(model: SMTModelForCausalLM, device: torch.device):
	model.positional_2D.pe = model.positional_2D.pe.to(device)
	model.decoder.positional_1D.pe = model.decoder.positional_1D.pe.to(device)
	# ---- end of copy ----------------------------------------------------------------------------


	class EndpointError(ValueError):
	"""A client-facing failure; str() is only the fixed code, never request content."""


	MAGIC = {
	"image/png": lambda b: b.startswith(b"\x89PNG\r\n\x1a\n"),
	"image/jpeg": lambda b: b.startswith(b"\xff\xd8\xff"),
	"image/webp": lambda b: len(b) >= 12 and b[:4] == b"RIFF" and b[8:12] == b"WEBP",
	}


	def decode_image(data) -> Image.Image:
	"""Validate the request body and return the image as RGB; raise EndpointError with a fixed code."""
	if not isinstance(data, dict):
	raise EndpointError("invalid_request")
	inputs = data.get("inputs")
	params = data.get("parameters") or {}
	if not isinstance(inputs, str) or not isinstance(params, dict):
	raise EndpointError("invalid_request")
	content_type = params.get("contentType")
	if content_type not in CONTENT_TYPES:
	raise EndpointError("unsupported_content_type")
	if len(inputs) > MAX_B64_CHARS:
	raise EndpointError("image_too_large")
	try:
	raw = base64.b64decode(inputs, validate=True)
	except (binascii.Error, ValueError):
	raise EndpointError("invalid_base64") from None
	if len(raw) > MAX_BYTES:
	raise EndpointError("image_too_large")
	if not MAGIC[content_type](raw):
	raise EndpointError("content_type_mismatch")
	with DECODE_LOCK:
	try:
	with Image.open(io.BytesIO(raw), formats=[CONTENT_TYPES[content_type]]) as im:
	# The header's size, checked before any pixel is decoded.
	if max(im.size) > MAX_SIDE or im.width * im.height > MAX_PIXELS:
	raise EndpointError("image_dimensions_too_large")
	im.load()
	rgba = im.convert("RGBA")
	except EndpointError:
	raise
	except Exception:
	raise EndpointError("invalid_image") from None
	# Transparent areas become white paper, not black.
	image = Image.new("RGB", rgba.size, "white")
	image.paste(rgba, mask=rgba.getchannel("A"))
	del rgba
	image.thumbnail((WORK_SIDE, WORK_SIDE))
	return image


	class EndpointHandler:
	def __init__(self, path: str = ""):
	torch.set_num_threads(TORCH_THREADS)
	model_dir = Path(path)
	self.device = torch.device("cpu")
	opts = ort.SessionOptions()
	opts.intra_op_num_threads = TORCH_THREADS
	opts.inter_op_num_threads = 1
	self.detector = ort.InferenceSession(
	str(model_dir / "yolo_staff_detector.onnx"), opts, providers=["CPUExecutionProvider"]
	)

	config = SMTConfig.from_pretrained(str(model_dir))
	sd = load_file(str(model_dir / "model.safetensors"))
	embed_size = sd["decoder.embedding.weight"].shape[0]
	out_size = sd["decoder.out_layer.weight"].shape[0]
	orig_out = config.out_categories
	config.out_categories = embed_size
	model = SMTModelForCausalLM(config)
	if embed_size != out_size:
	model.decoder.out_layer = Conv1d(config.d_model, out_size, kernel_size=1)
	model.load_state_dict(sd, strict=True)
	config.out_categories = orig_out
	model.to(self.device)
	_sync_pe_device(model, self.device)
	model.eval()
	self.smt = model
	self.lock = threading.Lock() # one page at a time on the two cores

	def __call__(self, data: dict) -> dict:
	image = decode_image(data)
	try:
	with self.lock:
	staves = detect_staves(image, self.detector)
	if not staves:
	raise EndpointError("no_staff_found")
	kerns = [transcribe_staff(self.smt, preprocess_staff(s).to(self.device)) for s in staves]
	musicxml, _ = kern_to_musicxml(kerns)
	except EndpointError:
	raise
	except Exception:
	log.exception("transcription failed") # the traceback holds no request content
	raise EndpointError("transcription_failed") from None
	return {"musicxml": musicxml, "staves": len(staves)}