app.py

"""
Melodimodell 67M – Hugging Face Space

Modellen arbetar på byte-nivå: varje token är en byte i UTF-8-texten, precis som när
du skickar "prompt": list(text.encode()) till llama-server. Här körs samma GGUF-fil
direkt i Space:t med llama-cpp-python, så ingen separat server behövs.

Notformat: [oktav][notnamn][volym][notlängd][tidstämpel], t.ex. 4G5d21
  • varje rad = 1 sekund, tidstämpel 0–59 i steg om 1/60 s
  • notnamn C I D J E F K G L A H B = C C# D D# E F F# G G# A A# B
  • volym 0–10, notlängd a–i = 1, 2, 4 … 256 × 33 ms
  • "*" skiljer låtar åt, så "*\n" betyder att en ny låt börjar

Miljövariabler (Settings → Variables and secrets i Space:t):
  MODEL_REPO, MODEL_FILE  vilken GGUF-fil som laddas ner
  MODEL_PATH              lokal GGUF-fil i stället för nedladdning
  N_CTX                   kontextlängd (standard 8192, som -c 8192)
  MAX_SECONDS             tak för reglaget "Sekunder ny musik"
  N_THREADS               antal CPU-trådar (annars CPU-kvoten, högst 4, eller 2)
  SOUNDFONT               sökväg till en .sf2-fil för FluidSynth
"""

import atexit
import codecs
import glob
import inspect
import os
import queue
import random
import re
import shutil
import subprocess
import tempfile
import threading
import time
import traceback
import wave
from collections import defaultdict
from functools import lru_cache
from pathlib import Path
from statistics import median

try:  # finns på ZeroGPU-Spaces och ska importeras före gradio
    import spaces
except ImportError:
    spaces = None

import gradio as gr
import mido
import numpy as np
from matplotlib.collections import PolyCollection
from matplotlib.figure import Figure
from matplotlib.patches import Patch

# ------------------------------------------------------------
#  Modell och körning
# ------------------------------------------------------------
MODEL_REPO = os.environ.get("MODEL_REPO", "qwrt/Melodimodell-67M")
MODEL_FILE = os.environ.get("MODEL_FILE", "midi-67m-f16.gguf")
MODEL_PATH = os.environ.get("MODEL_PATH", "")
N_CTX = int(os.environ.get("N_CTX", "8192"))
MAX_SECONDS = int(os.environ.get("MAX_SECONDS", "120"))
KEEP_FRACTION = 0.5  # andel av kontexten som behålls när den blir full

# ------------------------------------------------------------
#  Notformatet (samma konstanter som i dina konverteringsskript)
# ------------------------------------------------------------
STAMP_SPAN_MS = 1_000
TAKTER_PER_STAMP = 60
TACKT_MS = STAMP_SPAN_MS / TAKTER_PER_STAMP
BASE_DUR_MS = 33
VELOCITY_SCALE = 12.8
TEMPO_US_PER_BEAT = 500_000
TICKS_PER_BEAT = 480

NOTE_MAP = {
    "C": 0, "I": 1, "D": 2, "J": 3, "E": 4, "F": 5,
    "K": 6, "G": 7, "L": 8, "A": 9, "H": 10, "B": 11,
}
MIDI_TO_NOTE = {v: k for k, v in NOTE_MAP.items()}
DURATION_MAP = {"a": 1, "b": 2, "c": 4, "d": 8, "e": 16, "f": 32, "g": 64, "h": 128, "i": 256}
TOKEN_RE = re.compile(r"(\d+)([A-L])(\d+)([a-i])(\d+)")

# MIDI -> text, samma val som i convert_midi_to_custom_string.py
FILTER_DRUMS = True
FILTER_NON_PIANO = False
MAX_CHORD_NOTES = 10
MIN_VOLUME = 0
MAX_OCTAVE = 6

EXAMPLE_PROMPT = """*
4D4d0 4D4c18 4D4d27 4D4c45 4D4d54
4D4c12 4D4c21 4D4c30 4D4c39 4D4d48
4D4c6 4D4d15 4D4c33 4D4d42 4D4c59
4D4c9 4D4c18 4D4c27 4D4d36 2D4f36 1D4f36 4D4c54
4D4d3 4D4c21 4D4d30 4H4d30 2D4d30 1D4d30 4D4c48 4D4c57 4A4d57 2D4d57 1D4d57
3A4c6 4A4d10 4C4c15 4D4d24 3A4d24 3F4d24 3D4d24 2D4d24 4D4d42 3A4d42 3F4d42 3D4c42 2D4c42 3D4d51 2D4d51 4D4c59 3A4c59 3F4c59
4E4c9 4C4c9 3A4c9 3C4c9 2C4c9 4F4d18 4D4d18 3H4d18 2H4d18 1H4d18 4F4d36 4D4d36 3H4d36 2H4c36 1H4c36 2H4d45 1H4d45 4F4c54 4D4c54 3H4c54
4G4c3 4D4c3 2H4c3 1H4c3 4E4d12 4C4d12 3A4d12 2A4d12 1A4d12 4E4d30 4C4d30 3A4d30 2A4c30 1A4c30 2A4d39 1A4d39 4D4c48 3A4c48 4C4c57 3G4c57 2A4c57 1A4c57
4C4c6 3A4c6 3D4d6 2D4d6 4D4d15 3A4d15 3D4c24 2D4c24 3D4d33 2D4d33 3A4c42 4C4c51 3D4c51 2D4c51
4D4d0 3H4d0 3F4d0 2H4d0 1H4d0 4D4d18 3H4d18 3F4d18 2H4c18 1H4c18 2H4d27 1H4d27 4D4c36 3H4c36 4E4c45 3H4c45 2H4c45 1H4c45 4F4d54 4C4d54 3A4d54 2H4d54 1H4d54
4F4d12 4C4d12 3A4d12 2H4c12 1H4c12 2H4d21 1H4d21 4F4c30 4C4c30 4G4c39 4C4c39 2H4c39 1H4c39 4E4d48 4C4d48 3A4d48 2A4d48 1A4d48
4E4d6 4C4d6 3A4d6 2A4c6 1A4c6 2A4d15 1A4d15 4D4c24 3A4c24 4C4c33 3G4c33 2A4c33 1A4c33 4D4d42 3A4d42 3F4d42 3D4d42 2D4d42 3D4c59 2D4c59
3D4d9 2D4d9 3A4c18 4C4c27 3D4c27 2D4c27 4D4d36 3A4d36 3F4d36 3D4d36 2D4d36 4D4d54 3A4d54 3F4d54 3D4c54 2D4c54
3D4d3 2D4d3 4D4c12 3A4c12 4F4c21 3A4c21 3D4c21 2D4c21 4G4d30 4D4d30 3H4d30 2H4d30 1H4d30 4G4d48 4D4d48 3H4d48 2H4c48 1H4c48 2H4d57 1H4d57
4G4c6 4D4c6 4A4c15 4D4c15 2H4c15 1H4c15 4H4d24 4G4d24 4D4d24 2G4d24 1G4d24 4H4d42 4G4d42 4D4d42 2G4c42 1G4c42 2G4d51 1G4d51 4A4c59 4F4c59
4G4c9 4E4c9 2G4c9 1G4c9 4A4c18 4F4c18 3D4d18 2D4d18 4D4d27 3D4c36 2D4c36 3D4d45 2D4d45 4D4c54"""


# ------------------------------------------------------------
#  Text -> noter -> MIDI
# ------------------------------------------------------------
def normalize_prompt(text):
    """Enhetliga radbrytningar, bara ASCII och ett mellanslag mellan tokens."""
    text = (text or "").replace("\r\n", "\n").replace("\r", "\n")
    text = text.encode("ascii", "ignore").decode("ascii")
    text = "\n".join(" ".join(line.split()) for line in text.split("\n")).lstrip("\n")
    return text if text.strip() else "*\n"


def parse_notes(text, prompt_len=0):
    """Tolkar texten som parse_custom gör, men hoppar över trasiga tokens i stället för att krascha.

    Returnerar ([(start_ms, pitch, velocity, duration_ms, från_prompten)], antal_ogiltiga).
    """
    text = text.replace("*", " ")  # samma längd, så teckenpositionerna stämmer
    pos = len(text) - len(text.lstrip())  # som .strip(): inledande tomrader räknas inte
    events, invalid = [], 0
    for stamp, line in enumerate(text[pos:].split("\n")):
        for m in re.finditer(r"\S+", line):
            tok = TOKEN_RE.fullmatch(m.group())
            if not tok:
                invalid += 1
                continue
            octave, letter, vol, dur, takt = tok.groups()
            pitch = NOTE_MAP[letter] + (int(octave) + 1) * 12
            velocity = min(127, int(round(int(vol) * VELOCITY_SCALE)))
            if pitch > 127:
                invalid += 1
                continue
            if velocity == 0:  # volym 0 hörs inte
                continue
            start_ms = stamp * STAMP_SPAN_MS + int(takt) * TACKT_MS
            events.append((start_ms, pitch, velocity, DURATION_MAP[dur] * BASE_DUR_MS,
                           pos + m.start() < prompt_len))
        pos += len(line) + 1
    return events, invalid


def build_midi(events, out_path):
    """Ett spår med piano. Överlappande noter på samma tangent kapas, så att en
    note_off aldrig stänger av en senare not, och note_off kommer före note_on."""
    per_pitch = defaultdict(list)
    for start, pitch, velocity, duration, *_ in events:
        per_pitch[pitch].append((start, start + duration, velocity))

    messages = []
    for pitch, notes in per_pitch.items():
        notes.sort()
        for i, (start, end, velocity) in enumerate(notes):
            if i + 1 < len(notes):
                end = min(end, notes[i + 1][0])
            if end - start < 1:  # samma tangent två gånger på samma tid
                continue
            messages.append((start, 1, pitch, velocity))
            messages.append((end, 0, pitch, 0))
    messages.sort(key=lambda m: (m[0], m[1]))

    mid = mido.MidiFile(ticks_per_beat=TICKS_PER_BEAT)
    track = mido.MidiTrack()
    mid.tracks.append(track)
    track.append(mido.MetaMessage("set_tempo", tempo=TEMPO_US_PER_BEAT, time=0))
    track.append(mido.Message("program_change", program=0, time=0))
    last_tick = 0
    for ms, is_on, pitch, velocity in messages:
        tick = int(round(mido.second2tick(ms / 1000, TICKS_PER_BEAT, TEMPO_US_PER_BEAT)))
        track.append(mido.Message("note_on" if is_on else "note_off", note=pitch,
                                  velocity=velocity, time=max(0, tick - last_tick)))
        last_tick = max(last_tick, tick)
    mid.save(str(out_path))


# ------------------------------------------------------------
#  MIDI -> text (din convert_midi_to_custom_string, något förenklad)
# ------------------------------------------------------------
def midi_to_events(path):
    """MIDI -> [(start_ms, pitch, velocity, duration_ms)]. mido räknar om ticks till
    sekunder och tar hänsyn till alla tempobyten."""
    mid = mido.MidiFile(path, clip=True)
    programs, active, events, now = {}, {}, [], 0.0
    for msg in mid:  # alla spår sammanslagna, msg.time = sekunder sedan förra meddelandet
        now += msg.time
        if msg.type == "program_change":
            programs[msg.channel] = msg.program
            continue
        if msg.type not in ("note_on", "note_off"):
            continue
        if FILTER_DRUMS and msg.channel == 9:
            continue
        if FILTER_NON_PIANO and programs.get(msg.channel, 0) > 7:
            continue
        key = (msg.note, msg.channel)
        if key in active:  # note_off, eller ett nytt anslag på en tangent som redan låter
            start, velocity = active.pop(key)
            events.append((round(start * 1000, 3), msg.note, velocity, (now - start) * 1000))
        if msg.type == "note_on" and msg.velocity > 0:
            active[key] = (now, msg.velocity)
    for (note, _), (start, velocity) in active.items():
        events.append((round(start * 1000, 3), note, velocity, (now - start) * 1000))
    # Avrundningen tar bort flyttalsbrus från summeringen ovan, annars kan en not
    # på exakt 9000 ms hamna på 8999.9999 ms, alltså sista takten på raden innan.
    return sorted(events, key=lambda e: e[0])


def remove_duplicate_notes(events):
    best = {}
    for ev in events:
        key = (round(ev[0], 1), ev[1])
        if key not in best or ev[2] > best[key][2]:
            best[key] = ev
    return sorted(best.values(), key=lambda e: e[0])


def filter_large_chords(events, max_simultaneous):
    groups = defaultdict(list)
    for ev in events:
        groups[round(ev[0], 1)].append(ev)
    kept = []
    for group in groups.values():
        if len(group) > max_simultaneous:
            middle = median(e[1] for e in group)
            group = sorted(group, key=lambda e: abs(e[1] - middle))[:max_simultaneous]
        kept.extend(group)
    return sorted(kept, key=lambda e: e[0])


def duration_to_letter(duration_ms):
    units = duration_ms / BASE_DUR_MS
    return min(DURATION_MAP, key=lambda letter: abs(units - DURATION_MAP[letter]))


def events_to_text(events):
    stamps = defaultdict(list)
    for start_ms, pitch, velocity, duration_ms in events:
        octave = pitch // 12 - 1
        volume = max(0, min(10, int(round(velocity / VELOCITY_SCALE))))
        if octave < 0 or octave > MAX_OCTAVE or volume <= MIN_VOLUME:
            continue
        stamp = int(start_ms // STAMP_SPAN_MS)
        takt = max(0, min(59, int(round((start_ms - stamp * STAMP_SPAN_MS) / TACKT_MS))))
        token = f"{octave}{MIDI_TO_NOTE[pitch % 12]}{volume}{duration_to_letter(duration_ms)}{takt}"
        if token not in stamps[stamp]:
            stamps[stamp].append(token)
    if not stamps:
        return ""
    return "\n".join(" ".join(stamps.get(i, [])) for i in range(max(stamps) + 1))


def midi_to_text(path):
    events = remove_duplicate_notes(midi_to_events(path))
    events = [e for e in events if e[1] >= 12]  # negativ oktav bort
    if MAX_CHORD_NOTES > 0:
        events = filter_large_chords(events, MAX_CHORD_NOTES)
    return events_to_text(events)


# ------------------------------------------------------------
#  Ljud: FluidSynth om det finns (packages.txt), annars en enkel syntes
# ------------------------------------------------------------
AUDIO_SR = 22050
FLUIDSYNTH = shutil.which("fluidsynth")


def _find_soundfont():
    candidates = [
        os.environ.get("SOUNDFONT", ""),
        "/usr/share/sounds/sf2/FluidR3_GM.sf2",
        "/usr/share/soundfonts/FluidR3_GM.sf2",
        "/usr/share/sounds/sf2/default-GM.sf2",
        "/usr/share/soundfonts/default.sf2",
    ]
    candidates += sorted(glob.glob("/usr/share/sounds/sf2/*.sf2"))
    candidates += sorted(glob.glob("/usr/share/soundfonts/*.sf2"))
    return next((p for p in candidates if p and os.path.isfile(p)), None)


SOUNDFONT = _find_soundfont()


def _render_fluidsynth(midi_path):
    raw_path = Path(midi_path).with_suffix(".f32")
    cmd = [FLUIDSYNTH, "-ni", "-g", "0.6", "-r", str(AUDIO_SR), "-T", "raw", "-O", "float",
           "-F", str(raw_path), SOUNDFONT, str(midi_path)]
    subprocess.run(cmd, check=True, timeout=300,
                   stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    samples = np.fromfile(raw_path, dtype="<f4")
    raw_path.unlink(missing_ok=True)
    if samples.size < 2:
        raise RuntimeError("FluidSynth gav inget ljud")
    return samples[: samples.size // 2 * 2].reshape(-1, 2)  # stereo


@lru_cache(maxsize=256)
def _piano_tone(pitch, duration_ms):
    """En enkel pianolik ton: några övertoner som klingar av, mjuk attack och släpp."""
    f0 = 440.0 * 2 ** ((pitch - 69) / 12)
    hold = duration_ms / 1000
    n = int((hold + 0.3) * AUDIO_SR)
    t = np.arange(n, dtype=np.float32) / AUDIO_SR
    tau = float(np.clip(1.4 * (261.6 / f0) ** 0.5, 0.25, 3.0))  # mörka toner klingar längre
    tone = np.zeros(n, dtype=np.float32)
    for k, amp in enumerate((1.0, 0.5, 0.28, 0.14, 0.07), start=1):
        if k * f0 >= AUDIO_SR / 2:
            break
        tone += amp * np.sin(2 * np.pi * k * f0 * t) * np.exp(-t * k / tau)
    envelope = np.minimum(1.0, t / 0.005)
    released = t > hold
    envelope[released] *= np.exp(-(t[released] - hold) / 0.08)
    return (tone * envelope).astype(np.float32)


def _render_simple(events):
    end_s = max(e[0] + e[3] for e in events) / 1000 + 0.5
    buf = np.zeros(int(end_s * AUDIO_SR) + 1, dtype=np.float32)
    for start, pitch, velocity, duration, *_ in events:
        tone = _piano_tone(pitch, int(duration))
        i = int(start / 1000 * AUDIO_SR)
        j = min(len(buf), i + len(tone))
        buf[i:j] += tone[: j - i] * (velocity / 127)
    return buf


def _write_wav(path, samples):
    peak = float(np.max(np.abs(samples))) if samples.size else 0.0
    if peak > 0:
        samples = samples * (0.89 / peak)
    pcm = (samples * 32767).astype("<i2")
    with wave.open(str(path), "wb") as w:
        w.setnchannels(1 if pcm.ndim == 1 else pcm.shape[1])
        w.setsampwidth(2)
        w.setframerate(AUDIO_SR)
        w.writeframes(np.ascontiguousarray(pcm).tobytes())


def render_audio(midi_path, events, wav_path):
    samples = None
    if FLUIDSYNTH and SOUNDFONT:
        try:
            samples = _render_fluidsynth(midi_path)
        except Exception:
            traceback.print_exc()
    if samples is None:
        samples = _render_simple(events)
    _write_wav(wav_path, samples)
    return str(wav_path)


# ------------------------------------------------------------
#  Pianorulle
# ------------------------------------------------------------
PROMPT_RGB = (0.55, 0.61, 0.71)  # filtgrå: det du gav modellen
NEW_RGB = (0.91, 0.64, 0.24)  # mässing: det modellen skrev
MUTED = "#8a8f98"


def piano_roll(events):
    fig = Figure(figsize=(10, 3.8), dpi=110)
    fig.patch.set_alpha(0)
    ax = fig.add_subplot(111)
    ax.set_facecolor((0, 0, 0, 0))
    if not events:
        ax.text(0.5, 0.5, "Inga noter att visa", ha="center", va="center",
                color=MUTED, transform=ax.transAxes)
        ax.set_axis_off()
        return fig

    verts, colors = [], []
    for start, pitch, velocity, duration, from_prompt in events:
        x0, x1 = start / 1000, (start + duration) / 1000
        verts.append(((x0, pitch - 0.42), (x1, pitch - 0.42), (x1, pitch + 0.42), (x0, pitch + 0.42)))
        colors.append((*(PROMPT_RGB if from_prompt else NEW_RGB), 0.35 + 0.65 * velocity / 127))
    ax.add_collection(PolyCollection(verts, facecolors=colors, edgecolors="none"))

    low = min(e[1] for e in events) - 2
    high = max(e[1] for e in events) + 2
    ax.set_xlim(0, max(1.0, max(e[0] + e[3] for e in events) / 1000))
    ax.set_ylim(low, high)
    c_notes = [p for p in range(low, high + 1) if p % 12 == 0]
    ax.set_yticks(c_notes)
    ax.set_yticklabels([f"{p // 12 - 1}C" for p in c_notes])  # samma skrivsätt som formatet
    ax.set_xlabel("Sekunder", color=MUTED)
    ax.tick_params(colors=MUTED, labelsize=8)
    for side, spine in ax.spines.items():
        spine.set_visible(side == "bottom")
        spine.set_color(MUTED)
    ax.grid(axis="y", color=MUTED, alpha=0.15, linewidth=0.6)

    new_starts = [e[0] for e in events if not e[4]]
    handles = []
    if len(new_starts) < len(events):
        handles.append(Patch(color=PROMPT_RGB, label="Prompt"))
    if new_starts:
        handles.append(Patch(color=NEW_RGB, label="Ny musik"))
        if len(new_starts) < len(events):
            ax.axvline(min(new_starts) / 1000, color=MUTED, linewidth=1, linestyle=(0, (3, 3)))
    ax.legend(handles=handles, loc="upper right", frameon=False, fontsize=8, labelcolor=MUTED)
    fig.subplots_adjust(left=0.06, right=0.99, top=0.96, bottom=0.14)
    return fig


# ------------------------------------------------------------
#  Modellen
# ------------------------------------------------------------
def _cpu_count():
    """Antal trådar till llama.cpp och varifrån siffran kommer.

    Gratis-Spaces har 2 vCPU, men os.cpu_count() och sched_getaffinity visar ofta hela
    värddatorn (t.ex. 16). Fler trådar än kärnor gör llama.cpp mycket långsammare, så
    om CPU-kvoten inte går att läsa används 2. Sätt N_THREADS för större hårdvara.
    """
    if os.environ.get("N_THREADS"):
        return max(1, int(os.environ["N_THREADS"])), "N_THREADS"
    try:  # cgroup v2
        quota, period = Path("/sys/fs/cgroup/cpu.max").read_text().split()[:2]
        if quota != "max":
            return max(1, min(4, round(int(quota) / int(period)))), "CPU-kvot"
    except (OSError, ValueError):
        pass
    try:  # cgroup v1
        quota = int(Path("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").read_text())
        period = int(Path("/sys/fs/cgroup/cpu/cpu.cfs_period_us").read_text())
        if quota > 0 and period > 0:
            return max(1, min(4, round(quota / period))), "CPU-kvot"
    except (OSError, ValueError):
        pass
    return 2, "standard, CPU-kvoten gick inte att läsa"


def load_model():
    from llama_cpp import Llama

    path = MODEL_PATH
    if not path:
        from huggingface_hub import hf_hub_download

        path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE)
    threads, source = _cpu_count()
    print(f"Laddar {path} med n_ctx={N_CTX} och {threads} trådar ({source})", flush=True)
    return Llama(model_path=path, n_ctx=N_CTX, n_batch=512, n_threads=threads,
                 n_threads_batch=threads, verbose=os.environ.get("LLAMA_VERBOSE") == "1")


# ZeroGPU, som är gratisalternativet för Gradio-Spaces, startar bara appar som har
# minst en funktion med @spaces.GPU. Modellen körs på CPU, så funktionen nedan anropas
# aldrig och drar ingen GPU-tid. På vanlig CPU-hårdvara och lokalt gör den ingenting.
if spaces is not None:
    @spaces.GPU
    def _zerogpu_placeholder():
        return None


LLM, LOAD_ERROR = None, ""
try:
    LLM = load_model()
except Exception as exc:  # visa felet i gränssnittet i stället för att krascha
    LOAD_ERROR = f"{type(exc).__name__}: {exc}"
    traceback.print_exc()
CTX_LIMIT = LLM.n_ctx() if LLM is not None else N_CTX


# ------------------------------------------------------------
#  Generering. All kontakt med modellen sker i en och samma tråd; Gradio-
#  funktionen läser bara tokens från en kö och uppdaterar gränssnittet.
# ------------------------------------------------------------
JOBS = queue.Queue()
_SESSION_STOPS = {}  # session -> stop-event för Stoppa-knappen
_LATEST = {"stop": None}  # stop-event för det senast startade jobbet


def _slide(ctx):
    """Behåll slutet av kontexten och börja helst vid en ny rad (= hel sekund)."""
    keep = max(1, int(CTX_LIMIT * KEEP_FRACTION))
    if len(ctx) <= keep:
        return ctx
    tail = ctx[-keep:]
    for sep in (10, 32):  # "\n", annars " "
        if sep in tail:
            return tail[tail.index(sep) + 1:]
    return tail


def _run_job(prompt, opts, out_q, stop_ev):
    reason = "eos"
    try:
        if stop_ev.is_set():
            reason = "stopped"
            return
        out_q.put(("started", None))
        ctx = list(prompt.encode("utf-8"))  # samma som "prompt": list(text.encode())
        if len(ctx) > CTX_LIMIT - 256:
            ctx = _slide(ctx)
        # Slutar prompten mitt på en rad avslutar det första "\n" bara den raden.
        newlines_left = opts["seconds"] + (0 if prompt.endswith("\n") else 1)
        eos = LLM.token_eos()
        window = 0
        while True:
            LLM.set_seed((opts["seed"] + window) % 0xFFFFFFFF)
            window += 1
            stream = LLM.generate(
                list(ctx),  # kopia: ctx byggs på nedan medan generatorn körs
                temp=opts["temperature"],
                top_k=opts["top_k"],
                top_p=opts["top_p"],
                min_p=opts["min_p"],
                repeat_penalty=opts["repeat_penalty"],
            )
            context_full = False
            try:
                for tok in stream:
                    if stop_ev.is_set():
                        reason = "stopped"
                        return
                    if tok == eos or not 0 <= tok < 256:
                        reason = "eos"
                        return
                    if tok == 42 and opts["stop_at_star"]:  # "*": nästa låt börjar
                        reason = "star"
                        return
                    ctx.append(tok)
                    out_q.put(("tok", tok))
                    if tok == 10:
                        newlines_left -= 1
                        if newlines_left <= 0:
                            reason = "done"
                            return
                    if len(ctx) >= CTX_LIMIT:
                        context_full = True
                        break
            finally:
                stream.close()
            if not context_full:
                return
            # Kontexten är full: modellen får se slutet och skriver vidare.
            # Hela texten finns ändå kvar hos Gradio-funktionen.
            ctx = _slide(ctx)
            out_q.put(("slide", None))
    except Exception as exc:
        traceback.print_exc()
        reason = "error"
        out_q.put(("error", f"{type(exc).__name__}: {exc}"))
    finally:
        out_q.put(("done", reason))


_BUSY = threading.Lock()  # hålls medan modellen genererar


def _model_loop():
    while True:
        job = JOBS.get()
        with _BUSY:
            _run_job(*job)


threading.Thread(target=_model_loop, name="llm", daemon=True).start()


@atexit.register
def _close_model():
    """Frigör modellen innan Python river ner sina moduler. Annars försöker
    llama-cpp-python göra det för sent och skriver ut ett ofarligt men
    förvirrande TypeError när Space:t stängs av eller startas om."""
    if LLM is None:
        return
    if _LATEST["stop"] is not None:
        _LATEST["stop"].set()
    if _BUSY.acquire(timeout=5):  # vänta tills en ev. pågående generering har stannat
        try:
            LLM.close()
        except Exception:
            pass

OUT_ROOT = Path(tempfile.gettempdir()) / "melodimodell"


def _new_output_dir():
    OUT_ROOT.mkdir(parents=True, exist_ok=True)
    now = time.time()
    for old in OUT_ROOT.iterdir():  # städa bort resultat äldre än en timme
        try:
            if now - old.stat().st_mtime > 3600:
                shutil.rmtree(old, ignore_errors=True)
        except OSError:
            pass
    return Path(tempfile.mkdtemp(dir=OUT_ROOT))


def _session_key(request):
    return getattr(request, "session_hash", None) or "anon"


def _new_seconds(gen_text, prompt):
    return max(0, gen_text.count("\n") - (0 if prompt.endswith("\n") else 1))


def _drop_partial_token(text):
    """Tar bort en ev. halvskriven not i slutet."""
    if not text or text[-1] in " \n":
        return text
    return text[: max(text.rfind(" "), text.rfind("\n")) + 1]


FINAL_TEXT = {
    "done": "Klart.",
    "star": "Klart. Modellen avslutade låten.",
    "eos": "Klart. Modellen slutade skriva.",
    "stopped": "Stoppad.",
}


def generate_music(prompt, seconds, temperature, top_k, top_p, min_p,
                   repeat_penalty, seed, stop_at_star, request: gr.Request = None):
    if LLM is None:
        yield (prompt, f"Modellen kunde inte laddas ({LOAD_ERROR}). Läs loggen under Logs.",
               None, None, None)
        return

    prompt = normalize_prompt(prompt)
    try:
        seed = int(seed)
    except (TypeError, ValueError):
        seed = -1
    if seed < 0:
        seed = random.randint(0, 2**31 - 1)
    opts = {
        "seconds": int(seconds),
        "temperature": float(temperature),
        "top_k": int(top_k),
        "top_p": float(top_p),
        "min_p": float(min_p),
        "repeat_penalty": float(repeat_penalty),
        "seed": seed,
        "stop_at_star": bool(stop_at_star),
    }

    # Gradio kör bara en generering åt gången (concurrency_limit=1). Hänger ett
    # gammalt jobb kvar, t.ex. från en flik som stängdes mitt i, avbryts det här.
    if _LATEST["stop"] is not None:
        _LATEST["stop"].set()
    stop_ev = threading.Event()
    _LATEST["stop"] = stop_ev
    key = _session_key(request)
    _SESSION_STOPS[key] = stop_ev

    out_q = queue.Queue()
    JOBS.put((prompt, opts, out_q, stop_ev))

    decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
    pieces, reason, error, slides = [], None, "", 0
    t_start, last_ui = None, 0.0
    try:
        yield prompt, "Väntar på modellen …", None, None, None
        while reason is None:
            items = []
            try:
                items.append(out_q.get(timeout=0.25))
                while True:
                    items.append(out_q.get_nowait())
            except queue.Empty:
                pass
            for kind, value in items:
                if kind == "tok":
                    pieces.append(decoder.decode(bytes([value])))
                elif kind == "started":
                    t_start = time.time()
                elif kind == "slide":
                    slides += 1
                elif kind == "error":
                    error = value
                elif kind == "done":
                    reason = value
            if reason is None and t_start is not None and time.time() - last_ui > 0.3:
                last_ui = time.time()
                gen_text = "".join(pieces)
                rate = len(gen_text) / max(1e-6, last_ui - t_start)
                yield (prompt + gen_text,
                       f"Genererar … {_new_seconds(gen_text, prompt)} av {opts['seconds']} s "
                       f"ny musik ({rate:.0f} tecken per sekund).",
                       None, None, None)
    finally:
        stop_ev.set()
        if _SESSION_STOPS.get(key) is stop_ev:
            del _SESSION_STOPS[key]

    elapsed = time.time() - (t_start or time.time())
    gen_text = "".join(pieces)
    if reason != "done":
        gen_text = _drop_partial_token(gen_text)
    full_text = prompt + gen_text
    print(f"[generering] seed={seed} orsak={reason} nya_tecken={len(gen_text)} "
          f"tid={elapsed:.1f}s fönsterflytt={slides}", flush=True)

    head = f"Fel under genereringen: {error}." if reason == "error" else FINAL_TEXT.get(reason, "Klart.")
    try:
        events, invalid = parse_notes(full_text, prompt_len=len(prompt))
        out_dir = _new_output_dir()
        stem = f"melodi-seed{seed}"
        txt_path = out_dir / f"{stem}.txt"
        txt_path.write_text(full_text, encoding="utf-8")
        files, audio_path = [str(txt_path)], None
        if events:
            midi_path = out_dir / f"{stem}.mid"
            build_midi(events, midi_path)
            files.insert(0, str(midi_path))
            try:
                audio_path = render_audio(midi_path, events, out_dir / f"{stem}.wav")
            except Exception:
                traceback.print_exc()
        total_s = max((e[0] + e[3] for e in events), default=0) / 1000
        parts = [head, f"{_new_seconds(gen_text, prompt)} s ny musik på {elapsed:.0f} s. "
                       f"Hela låten är {total_s:.0f} s lång med {len(events)} noter. Seed {seed}."]
        if slides:
            parts.append(f"Kontexten blev full och flyttades fram {slides} "
                         f"{'gång' if slides == 1 else 'gånger'}.")
        if invalid:
            parts.append(f"{invalid} ogiltiga tokens hoppades över.")
        yield full_text, " ".join(parts), audio_path, piano_roll(events), files
    except Exception as exc:
        traceback.print_exc()
        yield full_text, f"{head} Kunde inte skapa MIDI eller ljud: {type(exc).__name__}: {exc}", None, None, None


def stop_generation(request: gr.Request = None):
    stop_ev = _SESSION_STOPS.get(_session_key(request))
    if stop_ev is not None:
        stop_ev.set()


def use_result_as_prompt(result, prompt):
    return result if (result or "").strip() else prompt


def midi_to_prompt(midi_file, seconds, prompt, status):
    path = midi_file if isinstance(midi_file, str) else getattr(midi_file, "name", None)
    if not path:
        return prompt, status
    try:
        text = midi_to_text(path)
    except Exception as exc:
        raise gr.Error(f"Kunde inte läsa MIDI-filen ({type(exc).__name__}: {exc}). Prova en annan fil.")
    lines = text.split("\n")
    first = next((i for i, line in enumerate(lines) if line.strip()), None)
    if first is None:
        raise gr.Error("Hittade inga noter att använda. Trummor och toner över oktav 6 tas bort.")
    chunk = lines[first:first + int(seconds)]
    new_prompt = "*\n" + "\n".join(chunk) + "\n"
    notes = sum(len(line.split()) for line in chunk)
    message = (f"Prompten är de första {len(chunk)} sekunderna av {Path(path).name}: "
               f"{notes} noter, {len(new_prompt)} tecken.")
    if len(new_prompt) > CTX_LIMIT - 256:
        message += " Den är längre än modellens kontext, så modellen ser bara slutet av den."
    return new_prompt, message


# ------------------------------------------------------------
#  Gränssnitt
# ------------------------------------------------------------
HEADER_MD = """
# Melodimodell 67M
En liten språkmodell som skriver musik som text, ett tecken i taget. Ge den en början och lyssna
på hur den fortsätter. Du kan skriva själv, ladda upp en MIDI-fil eller börja från noll.
"""

FORMAT_MD = """
Varje rad är en sekund. En not skrivs som `[oktav][not][volym][längd][tid]`, till exempel `4G5d21`:

| Del | Värden | I `4G5d21` |
|---|---|---|
| Oktav | 0–6, där `4C` är mittens C | `4` |
| Not | `C I D J E F K G L A H B` = C, C♯, D, D♯, E, F, F♯, G, G♯, A, A♯, B | `G` |
| Volym | 1–10 | `5` |
| Längd | `a`–`i` = 1, 2, 4 … 256 × 33 ms | `d` = 264 ms |
| Tid | 0–59, var i sekunden noten börjar, i steg om 1/60 s | `21` = 350 ms |

En rad med bara `*` betyder att en ny låt börjar. Därför börjar prompten med `*`.
"""

THEME = gr.themes.Soft(
    primary_hue=gr.themes.colors.amber,
    neutral_hue=gr.themes.colors.slate,
    font=[gr.themes.GoogleFont("Instrument Sans"), "ui-sans-serif", "system-ui", "sans-serif"],
    font_mono=[gr.themes.GoogleFont("JetBrains Mono"), "ui-monospace", "monospace"],
)

CSS = """
#prompt textarea, #result textarea {
    font-family: var(--font-mono);
    font-size: 13px;
    line-height: 1.6;
}
"""


def _split_style_kwargs(**style):
    """Beroende på Gradio-version tas title/theme/css emot av gr.Blocks() eller av launch()."""
    blocks_params = inspect.signature(gr.Blocks.__init__).parameters
    launch_params = inspect.signature(gr.Blocks.launch).parameters
    blocks_kw = {k: v for k, v in style.items() if k in blocks_params}
    launch_kw = {k: v for k, v in style.items() if k not in blocks_kw and k in launch_params}
    return blocks_kw, launch_kw


BLOCKS_KW, LAUNCH_KW = _split_style_kwargs(title="Melodimodell 67M", theme=THEME, css=CSS)

with gr.Blocks(**BLOCKS_KW) as demo:
    gr.Markdown(HEADER_MD)
    if LOAD_ERROR:
        gr.Markdown(f"**Modellen kunde inte laddas:** `{LOAD_ERROR}`. "
                    "Kontrollera MODEL_REPO och MODEL_FILE och läs loggen under Logs.")

    with gr.Row(equal_height=False):
        with gr.Column(scale=5):
            prompt_box = gr.Textbox(
                label="Början på låten",
                info="Modellen fortsätter där texten slutar. En rad är en sekund.",
                value=EXAMPLE_PROMPT, lines=12, max_lines=12, elem_id="prompt",
            )
            with gr.Row():
                example_btn = gr.Button("Ladda exemplet", size="sm")
                scratch_btn = gr.Button("Börja från noll", size="sm")
            with gr.Accordion("Börja från en MIDI-fil", open=False):
                midi_in = gr.File(label="MIDI-fil", file_types=[".mid", ".midi"], type="filepath")
                midi_secs = gr.Slider(minimum=1, maximum=60, value=10, step=1,
                                      label="Sekunder från filen")
            seconds_in = gr.Slider(minimum=5, maximum=MAX_SECONDS, value=30, step=5,
                                   label="Sekunder ny musik")
            with gr.Accordion("Sampling", open=False):
                temperature_in = gr.Slider(minimum=0.1, maximum=1.5, value=0.9, step=0.05,
                                           label="Temperatur")
                with gr.Row():
                    top_k_in = gr.Slider(minimum=0, maximum=64, value=0, step=1,
                                         label="top_k", info="0 = av")
                    top_p_in = gr.Slider(minimum=0.1, maximum=1.0, value=1.0, step=0.01,
                                         label="top_p")
                with gr.Row():
                    min_p_in = gr.Slider(minimum=0.0, maximum=0.5, value=0.0, step=0.01,
                                         label="min_p")
                    repeat_in = gr.Slider(minimum=1.0, maximum=1.3, value=1.0, step=0.01,
                                          label="repeat_penalty")
                with gr.Row():
                    seed_in = gr.Number(value=-1, precision=0, label="Seed",
                                        info="-1 ger en ny slumpad seed varje gång")
                    star_in = gr.Checkbox(value=True, label="Sluta när låten tar slut (*)")
            with gr.Row():
                gen_btn = gr.Button("Generera musik", variant="primary")
                stop_btn = gr.Button("Stoppa", variant="stop")

        with gr.Column(scale=6):
            status_box = gr.Textbox(label="Status", interactive=False, lines=2, max_lines=4)
            audio_out = gr.Audio(label="Ljud", type="filepath", interactive=False)
            roll_out = gr.Plot(label="Pianorulle")
            files_out = gr.File(label="MIDI och text", file_count="multiple", interactive=False)
            result_box = gr.Textbox(label="Resultat", lines=12, max_lines=12,
                                    interactive=False, elem_id="result")
            continue_btn = gr.Button("Fortsätt från resultatet")

    with gr.Accordion("Om notformatet", open=False):
        gr.Markdown(FORMAT_MD)

    gen_inputs = [prompt_box, seconds_in, temperature_in, top_k_in, top_p_in,
                  min_p_in, repeat_in, seed_in, star_in]
    gen_outputs = [result_box, status_box, audio_out, roll_out, files_out]

    gen_btn.click(generate_music, gen_inputs, gen_outputs,
                  concurrency_limit=1, concurrency_id="model")
    continue_btn.click(use_result_as_prompt, [result_box, prompt_box], prompt_box).then(
        generate_music, gen_inputs, gen_outputs, concurrency_limit=1, concurrency_id="model")
    stop_btn.click(stop_generation, None, None)
    example_btn.click(lambda: EXAMPLE_PROMPT, None, prompt_box)
    scratch_btn.click(lambda: "*\n", None, prompt_box)
    midi_inputs = [midi_in, midi_secs, prompt_box, status_box]
    midi_in.change(midi_to_prompt, midi_inputs, [prompt_box, status_box])
    midi_secs.change(midi_to_prompt, midi_inputs, [prompt_box, status_box])

demo.queue(max_size=20)

if __name__ == "__main__":
    demo.launch(**LAUNCH_KW)