Initial

.gitignore

@@ -0,0 +1,144 @@
+uv.lock
+
+wandb/
+toy-duration-predictor-lightning/
+
+sample/gv/json_filled_time_gaps
+sample/gv/json_time_adjusted
+sample/gv/json_preprocessed
+sample/gv/split_json
+sample/mssv/preprocessed
+preprocessed_gv/
+preprocessed_mssv/
+
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
+
+# Byte-compiled / optimized / DLL files
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+pip-wheel-metadata/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# PyDev project settings
+.pydevproject
+
+# PyCharm specific files
+.idea/
+*.iml
+
+# VS Code specific files
+.vscode/
+
+# Sass cache files
+.sass-cache/
+
+# Conda environments
+.conda/
+
+# Mypy cache
+.mypy_cache/
+
+# pytype
+.pytype/
+
+# Cache and log files created by tools
+.cache/
+*.log
+log/
+logs/
+
+# OS generated files
+.DS_Store
+.DS_Store?
+._*
+.Spotlight-V100
+.Trashes
+ehthumbs.db
+Thumbs.db

.python-version

@@ -0,0 +1 @@
+3.12

README.md

@@ -0,0 +1,15 @@
+# a
+
+aaaaaaaaaa [midii](https://github.com/ccss17/midii) 
+
+## Installation
+
+```shell
+pip install git+https://github.com/ccss17/toy-duration-predictor.git
+```
+
+## Usage
+
+```python
+import toy_duration_predictor as tdp
+```

pyproject.toml

@@ -0,0 +1,34 @@
+[project]
+name = "toy-duration-predictor"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+authors = [
+    { name = "ccsss", email = "chansol0505@naver.com" }
+]
+requires-python = ">=3.9"
+dependencies = [
+    "datasets>=3.6.0",
+    "fastai>=2.1.10",
+    "gradio>=4.44.1",
+    "lightning>=2.5.2",
+    "midii>=0.1.19",
+    "numpy>=2.0.2",
+    "pandas>=2.3.0",
+    "ray[data,serve,train,tune]>=2.47.1",
+    "torch>=2.7.1",
+    "wandb>=0.20.1",
+]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[dependency-groups]
+dev = [
+    "black>=25.1.0",
+    "ipykernel>=6.29.5",
+    "ipywidgets>=8.1.7",
+    "rich>=14.0.0",
+    "ruff>=0.12.0",
+]

src/toy_duration_predictor/__init__.py

@@ -0,0 +1,2 @@
+# from .preprocess import mssv
+# from .preprocess import utils

src/toy_duration_predictor/_legacy/train_fastai.py

@@ -0,0 +1,169 @@
+# --- 0. 필요 라이브러리 설치 ---
+# 이 스크립트를 실행하기 전에 먼저 터미널에서 아래 명령어를 실행해주세요.
+# pip install torch fastai wandb gradio
+
+import torch
+import torch.nn as nn
+from torch.utils.data import TensorDataset
+from fastai.data.core import DataLoaders
+from fastai.learner import Learner, pérdida_Calculada
+from fastai.callback.wandb import WandbCallback
+from fastai.callback.schedule import lr_find
+import numpy as np
+import pandas as pd
+import gradio as gr
+import os
+
+# --- 1. 하이퍼파라미터 및 상수 정의 ---
+MAX_SEQ_LENGTH = 32
+NUM_SINGERS = 100
+NUM_SAMPLES = 100000
+BATCH_SIZE = 256
+
+# 모델 구조 관련 파라미터 (fastai Learner에 전달)
+SID_EMBEDDING_DIM = 16
+GRU_UNITS = 128
+NUM_GRU_LAYERS = 2
+
+# --- 2. PyTorch 모델 아키텍처 정의 (fastai는 순수 PyTorch 모델을 그대로 사용) ---
+class DurationPredictorGRU(nn.Module):
+    """
+    fastai의 Learner가 래핑할 순수 PyTorch 모델.
+    """
+    def __init__(self, num_singers, sid_embedding_dim, gru_units, num_gru_layers):
+        super().__init__()
+        self.sid_embedding = nn.Embedding(num_singers, sid_embedding_dim)
+        gru_input_dim = 1 + sid_embedding_dim
+        self.gru = nn.GRU(
+            gru_input_dim, gru_units, num_gru_layers,
+            batch_first=True, bidirectional=True
+        )
+        self.fc_out = nn.Linear(gru_units * 2, 1)
+
+    def forward(self, x):
+        # fastai는 입력을 튜플로 묶어서 전달합니다.
+        duration_input, sid_input = x
+        sid_embedded = self.sid_embedding(sid_input)
+        duration_reshaped = duration_input.unsqueeze(-1)
+        
+        features = torch.cat([duration_reshaped, sid_embedded], dim=-1)
+        gru_output, _ = self.gru(features)
+        predictions = self.fc_out(gru_output)
+        return predictions
+
+# --- 3. 데이터 준비 ---
+print("--- 데이터셋 준비 중... ---")
+
+# 가상의 전체 데이터셋 생성 (DataFrame으로 관리하면 편리)
+data = {
+    'durations': [torch.rand(MAX_SEQ_LENGTH) for _ in range(NUM_SAMPLES)],
+    'sids': [torch.randint(0, NUM_SINGERS, (MAX_SEQ_LENGTH,)) for _ in range(NUM_SAMPLES)],
+    'labels': [d * torch.rand_like(d) * 2 for d in [d['durations'] for d in [{'durations': data} for data in [{'durations': torch.rand(MAX_SEQ_LENGTH)}] * NUM_SAMPLES]]]
+}
+df = pd.DataFrame(data)
+
+# 훈련(80%), 검증(10%), 테스트(10%) 인덱스 생성
+np.random.seed(42)
+indices = np.random.permutation(len(df))
+test_split_idx = int(len(df) * 0.1)
+val_split_idx = int(len(df) * 0.2)
+
+test_indices = indices[:test_split_idx]
+val_indices = indices[test_split_idx:val_split_idx]
+train_indices = indices[val_split_idx:]
+
+# fastai의 DataLoaders 객체 생성
+# 입력(x)은 튜플, 출력(y)은 단일 텐서로 구성
+train_ds = TensorDataset(torch.stack(df.loc[train_indices, 'durations'].tolist()),
+                         torch.stack(df.loc[train_indices, 'sids'].tolist()),
+                         torch.stack(df.loc[train_indices, 'labels'].tolist()).unsqueeze(-1))
+
+val_ds = TensorDataset(torch.stack(df.loc[val_indices, 'durations'].tolist()),
+                       torch.stack(df.loc[val_indices, 'sids'].tolist()),
+                       torch.stack(df.loc[val_indices, 'labels'].tolist()).unsqueeze(-1))
+
+test_ds = TensorDataset(torch.stack(df.loc[test_indices, 'durations'].tolist()),
+                        torch.stack(df.loc[test_indices, 'sids'].tolist()),
+                        torch.stack(df.loc[test_indices, 'labels'].tolist()).unsqueeze(-1))
+
+# fastai의 DataLoaders로 래핑
+# 입력(x)을 튜플로 묶기 위해 x_cat=2
+dls = DataLoaders.from_dsets(train_ds, val_ds, bs=BATCH_SIZE, device='cuda' if torch.cuda.is_available() else 'cpu')
+test_dl = dls.test_dl(test_ds, with_labels=True)
+
+print(f"훈련 데이터 샘플 수: {len(train_ds)}")
+print(f"검증 데이터 샘플 수: {len(val_ds)}")
+print(f"테스트 데이터 샘플 수: {len(test_ds)}")
+
+# --- 4. fastai Learner 생성 및 훈련 ---
+
+# 모델 인스턴스화
+model = DurationPredictorGRU(NUM_SINGERS, SID_EMBEDDING_DIM, GRU_UNITS, NUM_GRU_LAYERS)
+
+# Learner 생성 (모델, 데이터, 손실 함수, 콜백 등을 모두 묶음)
+learn = Learner(dls, model, loss_func=nn.MSELoss(), cbs=WandbCallback(log_preds=False))
+
+# --- 4a. 최적의 학습률 탐색 (Optuna 대신 사용) ---
+print("\n--- 1. 최적의 학습률 탐색 시작 (fastai lr_find) ---")
+# lr_find() 실행 후, 가장 가파른 기울기를 가진 지점의 학습률을 사용하는 것이 일반적
+suggested_lr = learn.lr_find(suggest_funcs=(lr_find.valley, lr_find.slide))
+print(f"fastai가 제안하는 최적 학습률: {suggested_lr.valley:.2e}")
+
+# --- 4b. 모델 훈련 ---
+print("\n--- 2. 제안된 학습률로 모델 훈련 시작 ---")
+# fine_tune은 헤드는 제안된 학습률로, 몸통은 더 낮은 학습률��� 훈련하는 등
+# 여러 best practice가 적용된 강력한 훈련 메소드
+learn.fine_tune(10, base_lr=suggested_lr.valley)
+
+print("모델 훈련 완료!")
+
+# --- 5. 최종 성능 평가 (테스트셋) ---
+print("\n--- 3. 최종 모델 평가 시작 (테스트 데이터셋 사용) ---")
+# get_preds를 사용하여 테스트셋에 대한 예측 및 손실 계산
+preds, targs, test_loss = learn.get_preds(dl=test_dl, with_loss=True)
+print(f"최종 테스트 손실 (MSE): {test_loss.item():.6f}")
+
+# --- 6. Gradio 데모 실행 ---
+print("\n--- 4. Gradio 데모 인터페이스 실행 ---")
+learn.model.eval() # 추론을 위해 모델을 평가 모드로 전환
+
+def predict_duration_fastai(singer_id_str, duration_sequence_str):
+    try:
+        # 입력 파싱 및 텐서화
+        singer_id = int(singer_id_str)
+        durations = [float(d.strip()) for d in duration_sequence_str.split(',')]
+        
+        if len(durations) > MAX_SEQ_LENGTH:
+            durations = durations[:MAX_SEQ_LENGTH]
+        else:
+            durations += [0] * (MAX_SEQ_LENGTH - len(durations))
+
+        duration_tensor = torch.tensor(durations, dtype=torch.float32).unsqueeze(0)
+        sid_tensor = torch.full_like(duration_tensor, singer_id, dtype=torch.long)
+
+        # fastai Learner를 사용한 예측
+        # learn.predict는 단일 아이템에 대한 예측과 디코딩을 수행
+        # 여기서는 모델 직접 호출이 더 간단
+        with torch.no_grad():
+            prediction = learn.model((duration_tensor.to(learn.dls.device), sid_tensor.to(learn.dls.device)))
+        
+        output_sequence = prediction.squeeze().cpu().tolist()
+        return ", ".join([f"{x:.4f}" for x in output_sequence])
+
+    except Exception as e:
+        return f"오류 발생: {e}"
+
+# Gradio 인터페이스 생성 및 실행
+iface = gr.Interface(
+    fn=predict_duration_fastai,
+    inputs=[
+        gr.Textbox(label="가수 ID (Singer ID)", value="10"),
+        gr.Textbox(label="음표 길이 시퀀스 (쉼표로 구분)", 
+                   value="0.1, 0.2, 0.15, 0.5, 0.4, 0.12, 0.1, 0.25")
+    ],
+    outputs=gr.Textbox(label="예측된 음표 길이 시퀀스"),
+    title="🎵 Duration Predictor (fastai + MLOps)",
+    description="fastai로 훈련된 모델입니다. 가수 ID와 정규 음표 길이 시퀀스를 입력하면, 해당 가수의 고유한 리듬 표현이 적용된 음표 길이를 예측합니다."
+)
+iface.launch()
+

src/toy_duration_predictor/_legacy/train_jax.py

@@ -0,0 +1,173 @@
+# --- 0. 필요 라이브러리 설치 ---
+# 이 스크립트를 실행하기 전에 먼저 터미널에서 아래 명령어를 실행해주세요.
+# CPU 버전: pip install jax flax optax elegy
+# GPU 버전: pip install jax[cuda11_pip] -f https://storage.googleapis.com/jax-releases/jax_cuda_releases.html
+#           pip install flax optax elegy
+
+import jax
+import jax.numpy as jnp
+import flax.linen as nn
+import optax
+import elegy
+import numpy as np
+from sklearn.model_selection import train_test_split
+
+# --- 1. 하이퍼파라미터 및 상수 정의 ---
+MAX_SEQ_LENGTH = 32
+NUM_SINGERS = 100
+NUM_SAMPLES = 100000
+BATCH_SIZE = 256
+
+# 모델 구조 관련 파라미터
+SID_EMBEDDING_DIM = 16
+GRU_UNITS = 128
+NUM_GRU_LAYERS = 2  # Flax의 GRUCell은 num_layers를 직접 지원하지 않으므로, 루프로 구현해야 합니다.
+                  # 이 예제에서는 간결성을 위해 1개 층으로 구현합니다.
+LEARNING_RATE = 0.001
+NUM_EPOCHS = 20
+
+# --- 2. Flax를 사용한 모델 아키텍처 정의 ---
+class DurationPredictorGRU(nn.Module):
+    """
+    Flax를 사용하여 정의한 Duration Predictor 모델.
+    JAX의 순수 함수 철학을 따릅니다.
+    """
+    num_singers: int
+    sid_embedding_dim: int
+    gru_units: int
+    
+    @nn.compact
+    def __call__(self, x):
+        # Elegy는 입력을 튜플/리스트 대신 딕셔너리로 받는 것을 선호합니다.
+        duration_input = x['duration_input']
+        sid_input = x['sid_input']
+
+        # 1. SID 임베딩
+        sid_embedded = nn.Embed(
+            num_embeddings=self.num_singers,
+            features=self.sid_embedding_dim
+        )(sid_input)
+
+        # 2. 음표 길이 차원 확장
+        duration_reshaped = jnp.expand_dims(duration_input, axis=-1)
+        
+        # 3. 피처 연결
+        features = jnp.concatenate([duration_reshaped, sid_embedded], axis=-1)
+
+        # 4. 양방향 GRU
+        # Flax의 Bidirectional 래퍼는 RNNCell을 감싸서 양방향으로 만듭니다.
+        gru_cell = nn.GRUCell(features=self.gru_units)
+        gru_output = nn.Bidirectional(gru_cell)(features)
+        
+        # 5. 출력층
+        # Flax의 Dense는 시퀀스 입력에 대해 자동으로 Time-Distributed처럼 작동합니다.
+        predictions = nn.Dense(features=1)(gru_output)
+        
+        return predictions
+
+# --- 3. 데이터 준비 ---
+print("--- 데이터셋 준비 중... ---")
+
+# 가상의 Numpy 데이터셋 생성
+durations = np.random.rand(NUM_SAMPLES, MAX_SEQ_LENGTH).astype(np.float32)
+sids = np.random.randint(0, NUM_SINGERS, (NUM_SAMPLES, MAX_SEQ_LENGTH)).astype(np.int32)
+labels = (durations * np.random.rand(NUM_SAMPLES, MAX_SEQ_LENGTH) * 2).astype(np.float32)
+
+# 훈련(80%), 검증/테스트(20%)로 먼저 분할
+dur_train, dur_rem, sids_train, sids_rem, y_train, y_rem = train_test_split(
+    durations, sids, labels, test_size=0.2, random_state=42)
+
+# 검증(10%), 테스트(10%)로 분할
+dur_val, dur_test, sids_val, sids_test, y_val, y_test = train_test_split(
+    dur_rem, sids_rem, y_rem, test_size=0.5, random_state=42)
+
+# Elegy가 사용할 수 있도록 입력 데이터를 딕셔너리 형태로 묶습니다.
+X_train = {'duration_input': dur_train, 'sid_input': sids_train}
+X_val = {'duration_input': dur_val, 'sid_input': sids_val}
+X_test = {'duration_input': dur_test, 'sid_input': sids_test}
+
+print(f"훈련 데이터 샘플 수: {len(y_train)}")
+print(f"검증 데이터 샘플 수: {len(y_val)}")
+print(f"테스트 데이터 샘플 수: {len(y_test)}")
+
+# --- 4. Elegy를 사용한 모델 훈련 및 평가 ---
+
+# Elegy 모델 생성
+# Keras와 매우 유사하게, 모듈, 손실함수, 옵티마이저, 메트릭을 정의합니다.
+model = elegy.Model(
+    module=DurationPredictorGRU(
+        num_singers=NUM_SINGERS,
+        sid_embedding_dim=SID_EMBEDDING_DIM,
+        gru_units=GRU_UNITS
+    ),
+    loss=elegy.losses.MeanSquaredError(),
+    optimizer=optax.adam(learning_rate=LEARNING_RATE),
+    metrics=[elegy.metrics.MeanAbsoluteError()]
+)
+
+print("\n--- 모델 훈련 시작 (Elegy)... ---")
+# Keras와 거의 동일한 .fit() API를 사용
+# (참고: WandB, Gradio 등은 Elegy의 콜백 시스템을 통해 연동 가능합니다)
+history = model.fit(
+    x=X_train,
+    y=y_train,
+    epochs=NUM_EPOCHS,
+    batch_size=BATCH_SIZE,
+    validation_data=(X_val, y_val),
+    callbacks=[elegy.callbacks.EarlyStopping(monitor='val_loss', patience=5)],
+    shuffle=True
+)
+
+print("모델 훈련 완료!")
+
+# --- 5. 최종 성능 평가 (테스트셋) ---
+print("\n--- 최종 모델 평가 시작 (테스트 데이터셋 사용)... ---")
+
+# .evaluate() API를 사용하여 최종 성능 측정
+test_metrics = model.evaluate(X_test, y_test)
+print("-" * 50)
+print(f"최종 테스트 결과: {test_metrics}")
+print("-" * 50)
+
+# --- 6. Gradio 데모를 위한 예측 함수 (예시) ---
+# Elegy 모델은 내부적으로 JAX의 JIT 컴파일을 사용하여 예측 속도가 매우 빠릅니다.
+@jax.jit
+def predict_fn(params, x):
+    return model.module.apply({'params': params}, x)
+
+def gradio_predict(singer_id_str, duration_sequence_str):
+    try:
+        singer_id = int(singer_id_str)
+        durations = [float(d.strip()) for d in duration_sequence_str.split(',')]
+        
+        if len(durations) > MAX_SEQ_LENGTH:
+            durations = durations[:MAX_SEQ_LENGTH]
+        else:
+            durations += [0] * (MAX_SEQ_LENGTH - len(durations))
+
+        duration_np = np.array(durations, dtype=np.float32).reshape(1, -1)
+        sid_np = np.full_like(duration_np, singer_id, dtype=np.int32)
+        
+        input_dict = {'duration_input': duration_np, 'sid_input': sid_np}
+        
+        # JIT 컴파일된 함수로 예측 실행
+        prediction = predict_fn(model.states.params, input_dict)
+        
+        output_sequence = np.asarray(prediction).flatten().tolist()
+        return ", ".join([f"{x:.4f}" for x in output_sequence])
+    except Exception as e:
+        return f"오류 발생: {e}"
+
+# (Gradio 실행 부분은 주석 처리. 필요시 주석 해제하여 사용)
+# print("\n--- Gradio 데모 인터페이스 실행 ---")
+# iface = gr.Interface(
+#     fn=gradio_predict,
+#     inputs=[
+#         gr.Textbox(label="가수 ID (Singer ID)", value="10"),
+#         gr.Textbox(label="음표 길이 시퀀스 (쉼표로 구분)", value="0.1, 0.2, 0.15, 0.5")
+#     ],
+#     outputs=gr.Textbox(label="예측된 음표 길이 시퀀스"),
+#     title="🎵 Duration Predictor (JAX/Flax/Elegy)",
+#     description="JAX 생태계로 훈련된 모델입니다. 가수 ID와 정규 음표 길이 시퀀스를 입력하면, 해당 가수의 고유한 리듬 표현이 적용된 음표 길이를 예측합니다."
+# )
+# iface.launch()

src/toy_duration_predictor/_legacy/train_tensorflow.py

@@ -0,0 +1,183 @@
+# --- 0. 필요 라이브러리 설치 ---
+# 이 스크립트를 실행하기 전에 먼저 터미널에서 아래 명령어를 실행해주세요.
+# pip install tensorflow numpy wandb keras-tuner gradio
+
+import tensorflow as tf
+from tensorflow import keras
+from tensorflow.keras import layers, Model
+import numpy as np
+import keras_tuner as kt
+from wandb.keras import WandbCallback
+import gradio as gr
+import os
+
+# --- 1. 하이퍼파라미터 및 상수 정의 ---
+MAX_SEQ_LENGTH = 32
+NUM_SINGERS = 100
+NUM_SAMPLES = 100000
+BATCH_SIZE = 256
+BUFFER_SIZE = 10000 # tf.data.Dataset 셔플을 위한 버퍼 크기
+
+# --- 2. 데이터 준비 (tf.data.Dataset 사용) ---
+print("--- 데이터셋 준비 중... ---")
+
+def generate_dummy_data():
+    """가상의 데이터셋을 생성하는 제너레이터 함수"""
+    for _ in range(NUM_SAMPLES):
+        duration = np.random.rand(MAX_SEQ_LENGTH).astype(np.float32)
+        sid = np.random.randint(0, NUM_SINGERS, (MAX_SEQ_LENGTH,)).astype(np.int32)
+        label = (duration * np.random.rand(MAX_SEQ_LENGTH) * 2).astype(np.float32)
+        # Keras 모델은 입력과 출력을 딕셔너리 형태로 받는 것이 편리합니다.
+        yield {'duration_input': duration, 'sid_input': sid}, label
+
+# tf.data.Dataset 객체 생성
+full_dataset = tf.data.Dataset.from_generator(
+    generate_dummy_data,
+    output_signature=(
+        {'duration_input': tf.TensorSpec(shape=(MAX_SEQ_LENGTH,), dtype=tf.float32),
+         'sid_input': tf.TensorSpec(shape=(MAX_SEQ_LENGTH,), dtype=tf.int32)},
+        tf.TensorSpec(shape=(MAX_SEQ_LENGTH,), dtype=tf.float32)
+    )
+)
+
+# 훈련(80%), 검증(10%), 테스트(10%) 데이터셋으로 분할
+full_dataset = full_dataset.shuffle(BUFFER_SIZE, seed=42) # 분할 전 전체 셔플
+train_size = int(0.8 * NUM_SAMPLES)
+val_size = int(0.1 * NUM_SAMPLES)
+
+train_dataset = full_dataset.take(train_size)
+val_and_test_dataset = full_dataset.skip(train_size)
+val_dataset = val_and_test_dataset.take(val_size)
+test_dataset = val_and_test_dataset.skip(val_size)
+
+# 데이터로더 생성 (배치, 프리페치 등 최적화)
+train_loader = train_dataset.batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)
+val_loader = val_dataset.batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)
+test_loader = test_dataset.batch(BATCH_SIZE).prefetch(tf.data.AUTOTUNE)
+
+print(f"훈련 데이터 샘플 수: {train_size}")
+print(f"검증 데이터 샘플 수: {val_size}")
+print(f"테스트 데이터 샘플 수: {NUM_SAMPLES - train_size - val_size}")
+
+# --- 3. KerasTuner를 사용한 하이퍼파라미터 최적화 ---
+
+def build_model(hp: kt.HyperParameters):
+    """KerasTuner가 하이퍼파라미터를 탐색하기 위한 모델 빌드 함수"""
+    
+    # 입력층 정의
+    duration_input = layers.Input(shape=(MAX_SEQ_LENGTH,), name='duration_input')
+    sid_input = layers.Input(shape=(MAX_SEQ_LENGTH,), name='sid_input')
+
+    # 하이퍼파라미터 탐색 공간 정의
+    sid_embedding_dim = hp.Choice('sid_embedding_dim', values=[8, 16, 32])
+    gru_units = hp.Choice('gru_units', values=[64, 128])
+    learning_rate = hp.Float('learning_rate', min_value=1e-4, max_value=1e-2, sampling='log')
+    
+    # 모델 레이어
+    sid_embedding = layers.Embedding(input_dim=NUM_SINGERS, output_dim=sid_embedding_dim)(sid_input)
+    duration_reshaped = layers.Reshape((MAX_SEQ_LENGTH, 1))(duration_input)
+    
+    x = layers.Concatenate()([duration_reshaped, sid_embedding])
+    
+    # Keras의 GRU는 기본적으로 dropout 인자를 가짐
+    x = layers.Bidirectional(layers.GRU(gru_units, return_sequences=True))(x)
+    x = layers.Bidirectional(layers.GRU(gru_units, return_sequences=True))(x)
+    
+    outputs = layers.TimeDistributed(layers.Dense(1, activation='linear'))(x)
+
+    model = Model(inputs=[duration_input, sid_input], outputs=outputs)
+    
+    model.compile(optimizer=keras.optimizers.Adam(learning_rate), loss='mean_squared_error')
+    
+    return model
+
+print("\n--- 1. 하이퍼파라미터 최적화 시작 (KerasTuner) ---")
+tuner = kt.Hyperband(
+    build_model,
+    objective='val_loss',
+    max_epochs=10,
+    factor=3,
+    directory='keras_tuner_dir',
+    project_name='duration_predictor'
+)
+
+# KerasTuner 실행
+tuner.search(train_loader, epochs=10, validation_data=val_loader, callbacks=[keras.callbacks.EarlyStopping(patience=3)])
+
+# 최적 하이퍼파라미터 추출
+best_hps = tuner.get_best_hyperparameters(num_trials=1)[0]
+print("최적화 완료!")
+print(f"최적의 학습률: {best_hps.get('learning_rate')}")
+print(f"최적의 임베딩 차원: {best_hps.get('sid_embedding_dim')}")
+print(f"최적의 GRU 유닛 수: {best_hps.get('gru_units')}")
+
+# --- 4. 최종 모델 훈련 및 평가 ---
+print("\n--- 2. 최적의 하이퍼파라미터로 최종 모델 학습 및 평가 시작 ---")
+
+# WandB 초기화
+import wandb
+wandb.init(project="duration_predictor_tf_keras", config=best_hps.values)
+
+# 최적의 하이퍼파라미터로 최종 모델 빌드
+final_model = tuner.hypermodel.build(best_hps)
+
+# 체크포인트 및 조기 종료 콜백 설정
+checkpoint_cb = keras.callbacks.ModelCheckpoint("best_model.keras", save_best_only=True)
+early_stopping_cb = keras.callbacks.EarlyStopping(patience=5, restore_best_weights=True)
+
+# 모델 훈련
+final_model.fit(
+    train_loader,
+    epochs=50,
+    validation_data=val_loader,
+    callbacks=[WandbCallback(), checkpoint_cb, early_stopping_cb]
+)
+print("최종 모델 학습 완료!")
+
+# --- 5. 최종 성능 평가 (테스트셋) ---
+print("\n--- 3. 최종 모델 평가 시작 (테스트 데이터셋 사용) ---")
+best_model = keras.models.load_model("best_model.keras")
+test_loss = best_model.evaluate(test_loader)
+print("-" * 50)
+print(f"최종 테스트 손실 (MSE): {test_loss:.6f}")
+wandb.log({"test_loss": test_loss})
+wandb.finish()
+
+# --- 6. Gradio 데모 실행 ---
+print("\n--- 4. Gradio 데모 인터페이스 실행 ---")
+
+def predict_duration_keras(singer_id_str, duration_sequence_str):
+    try:
+        singer_id = int(singer_id_str)
+        durations = [float(d.strip()) for d in duration_sequence_str.split(',')]
+        
+        if len(durations) > MAX_SEQ_LENGTH:
+            durations = durations[:MAX_SEQ_LENGTH]
+        else:
+            durations += [0] * (MAX_SEQ_LENGTH - len(durations))
+
+        # Keras 모델 입력 형태로 변환 (Numpy 배열)
+        duration_np = np.array(durations, dtype=np.float32).reshape(1, -1)
+        sid_np = np.full_like(duration_np, singer_id, dtype=np.int32)
+
+        # 예측 실행
+        prediction = best_model.predict({'duration_input': duration_np, 'sid_input': sid_np})
+        
+        output_sequence = prediction.flatten().tolist()
+        return ", ".join([f"{x:.4f}" for x in output_sequence])
+    except Exception as e:
+        return f"오류 발생: {e}"
+
+iface = gr.Interface(
+    fn=predict_duration_keras,
+    inputs=[
+        gr.Textbox(label="가수 ID (Singer ID)", value="10"),
+        gr.Textbox(label="음표 길이 시퀀스 (쉼표로 구분)", 
+                   value="0.1, 0.2, 0.15, 0.5, 0.4, 0.12, 0.1, 0.25")
+    ],
+    outputs=gr.Textbox(label="예측된 음표 길이 시퀀스"),
+    title="🎵 Duration Predictor (Keras + MLOps)",
+    description="Keras로 훈련된 모델입니다. 가수 ID와 정규 음표 길이 시퀀스를 입력하면, 해당 가수의 고유한 리듬 표현이 적용된 음표 길이를 예측합니다."
+)
+
+iface.launch()

src/toy_duration_predictor/_legacy/train_torch_mlops.py

@@ -0,0 +1,219 @@
+# --- 0. 필요 라이브러리 설치 ---
+# 이 스크립트를 실행하기 전에 먼저 터미널에서 아래 명령어를 실행해주세요.
+# pip install torch numpy pytorch-lightning wandb optuna gradio
+
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader, TensorDataset, random_split
+import pytorch_lightning as pl
+from pytorch_lightning.loggers import WandbLogger
+from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStopping
+import numpy as np
+import optuna
+import gradio as gr
+import os
+
+# --- 1. 가상 데이터 생성 및 PyTorch Lightning 데이터 모듈 (업데이트) ---
+# 실제로는 이 부분에 MIDI 데이터를 전처리하고 로드하는 코드가 들어갑니다.
+
+# 데이터 관련 상수
+MAX_SEQ_LENGTH = 32
+NUM_SINGERS = 100
+NUM_SAMPLES = 100000  # 전체 샘플 수 (10만개로 증가)
+BATCH_SIZE = 256      # 배치 크기
+
+class DurationDataModule(pl.LightningDataModule):
+    def __init__(self, batch_size):
+        super().__init__()
+        self.batch_size = batch_size
+        self.full_dataset = None
+        self.train_dataset = None
+        self.val_dataset = None
+        self.test_dataset = None
+
+    def prepare_data(self):
+        # 이 메소드는 단일 프로세스에서만 실행됩니다.
+        # 데이터를 다운로드하거나 생성하는 로직을 여기에 넣습니다.
+        pass
+
+    def setup(self, stage=None):
+        # 모든 GPU/TPU에서 실행됩니다. 데이터를 분할하고 할당합니다.
+        if not self.full_dataset:
+            # 가상의 전체 데이터셋 생성
+            durations = torch.rand(NUM_SAMPLES, MAX_SEQ_LENGTH)
+            sids = torch.randint(0, NUM_SINGERS, (NUM_SAMPLES, MAX_SEQ_LENGTH))
+            labels = durations * torch.rand_like(durations) * 2
+            self.full_dataset = TensorDataset(durations, sids, labels)
+
+        # 훈련(80%), 검증(10%), 테스트(10%) 데이터셋으로 분할
+        train_size = int(0.8 * len(self.full_dataset))
+        val_size = int(0.1 * len(self.full_dataset))
+        test_size = len(self.full_dataset) - train_size - val_size
+        
+        # random_split을 사용하여 데이터를 나눔 (매번 동일한 분할을 위해 시드 고정)
+        self.train_dataset, self.val_dataset, self.test_dataset = random_split(
+            self.full_dataset, [train_size, val_size, test_size],
+            generator=torch.Generator().manual_seed(42)
+        )
+
+    def train_dataloader(self):
+        return DataLoader(self.train_dataset, batch_size=self.batch_size, shuffle=True, num_workers=os.cpu_count()//2)
+
+    def val_dataloader(self):
+        return DataLoader(self.val_dataset, batch_size=self.batch_size, num_workers=os.cpu_count()//2)
+        
+    def test_dataloader(self):
+        # 테스트 데이터로더 추가
+        return DataLoader(self.test_dataset, batch_size=self.batch_size, num_workers=os.cpu_count()//2)
+
+# --- 2. PyTorch Lightning 모델 (업데이트) ---
+# 테스트 스텝 추가
+
+class DurationPredictor(pl.LightningModule):
+    def __init__(self, hparams):
+        super().__init__()
+        self.save_hyperparameters(hparams)
+        self.sid_embedding = nn.Embedding(self.hparams.num_singers, self.hparams.sid_embedding_dim)
+        gru_input_dim = 1 + self.hparams.sid_embedding_dim
+        self.gru = nn.GRU(gru_input_dim, self.hparams.gru_units, self.hparams.num_gru_layers, 
+                          batch_first=True, bidirectional=True, 
+                          dropout=self.hparams.dropout_rate if self.hparams.num_gru_layers > 1 else 0)
+        self.fc_out = nn.Linear(self.hparams.gru_units * 2, 1)
+        self.loss_fn = nn.MSELoss()
+
+    def forward(self, duration_input, sid_input):
+        sid_embedded = self.sid_embedding(sid_input)
+        duration_reshaped = duration_input.unsqueeze(-1)
+        x = torch.cat([duration_reshaped, sid_embedded], dim=-1)
+        gru_output, _ = self.gru(x)
+        predictions = self.fc_out(gru_output)
+        return predictions
+
+    def _common_step(self, batch, batch_idx):
+        durations, sids, labels = batch
+        predictions = self.forward(durations, sids)
+        loss = self.loss_fn(predictions.squeeze(-1), labels)
+        return loss
+
+    def training_step(self, batch, batch_idx):
+        loss = self._common_step(batch, batch_idx)
+        self.log('train_loss', loss, on_step=True, on_epoch=True, prog_bar=True)
+        return loss
+
+    def validation_step(self, batch, batch_idx):
+        loss = self._common_step(batch, batch_idx)
+        self.log('val_loss', loss, on_epoch=True, prog_bar=True)
+        return loss
+    
+    def test_step(self, batch, batch_idx):
+        loss = self._common_step(batch, batch_idx)
+        self.log('test_loss', loss, on_epoch=True, prog_bar=True)
+        return loss
+
+    def configure_optimizers(self):
+        return torch.optim.Adam(self.parameters(), lr=self.hparams.learning_rate)
+
+# --- 3. Optuna를 사용한 하이퍼파라미터 최적화 ---
+def objective(trial: optuna.Trial):
+    hparams = {
+        'learning_rate': trial.suggest_float('learning_rate', 1e-4, 1e-2, log=True),
+        'sid_embedding_dim': trial.suggest_categorical('sid_embedding_dim', [8, 16, 32]),
+        'gru_units': trial.suggest_categorical('gru_units', [64, 128]),
+        'num_gru_layers': trial.suggest_int('num_gru_layers', 1, 2),
+        'dropout_rate': trial.suggest_float('dropout_rate', 0.1, 0.4),
+        'num_singers': NUM_SINGERS
+    }
+    
+    wandb_logger = WandbLogger(project="duration_predictor_optuna", name=f"trial-{trial.number}", group="optuna-study")
+    wandb_logger.log_hyperparams(hparams)
+
+    model = DurationPredictor(hparams)
+    datamodule = DurationDataModule(batch_size=BATCH_SIZE)
+
+    trainer = pl.Trainer(
+        logger=wandb_logger, max_epochs=5, accelerator="auto", devices=1,
+        enable_checkpointing=False, callbacks=[EarlyStopping(monitor="val_loss", mode="min", patience=2)]
+    )
+    trainer.fit(model, datamodule)
+    return trainer.callback_metrics["val_loss"].item()
+
+# --- 4. 메인 실행 블록 (업데이트) ---
+if __name__ == '__main__':
+    # --- 1. 하이퍼파라미터 최적화 ---
+    print("--- 1. 하이퍼파라미터 최적화 시작 (Optuna) ---")
+    study = optuna.create_study(direction='minimize')
+    study.optimize(objective, n_trials=10) # 실제로는 50~100회 이상 권장
+
+    print("최적화 완료!")
+    best_hparams = study.best_params
+    print(f"최고의 val_loss: {study.best_value}\n최적 하이퍼파라미터: {best_hparams}")
+    
+    # --- 2. 최적 모델 훈련 및 저장 ---
+    print("\n--- 2. 최적의 하이퍼파라미터로 최종 모델 학습 및 평가 시작 ---")
+    final_hparams = best_hparams
+    final_hparams['num_singers'] = NUM_SINGERS
+    
+    datamodule = DurationDataModule(batch_size=BATCH_SIZE)
+    model = DurationPredictor(final_hparams)
+    
+    wandb_logger = WandbLogger(project="duration_predictor_final", name="final_best_model")
+    wandb_logger.log_hyperparams(final_hparams)
+
+    checkpoint_callback = ModelCheckpoint(
+        dirpath='checkpoints', filename='best-model', save_top_k=1, monitor='val_loss', mode='min'
+    )
+    
+    trainer = pl.Trainer(
+        logger=wandb_logger, max_epochs=20, accelerator="auto", devices=1,
+        callbacks=[checkpoint_callback, EarlyStopping(monitor="val_loss", mode="min", patience=4)]
+    )
+    trainer.fit(model, datamodule)
+    print("최종 모델 학습 완료!")
+    
+    # --- 3. 최종 성능 평가 (테스트) ---
+    print(f"저장된 최고 성능 모델 경로: {checkpoint_callback.best_model_path}")
+    # trainer.test()는 최고의 체크포인트를 자동으로 불러와 평가를 진행합니다.
+    test_results = trainer.test(ckpt_path='best', datamodule=datamodule)
+    print("최종 테스트 결과:", test_results)
+    
+    # --- 4. Gradio 데모 실행 ---
+    print("\n--- 4. Gradio 데모 인터페이스 실행 ---")
+    
+    best_model = DurationPredictor.load_from_checkpoint(checkpoint_callback.best_model_path)
+    best_model.eval()
+
+    def predict_duration(singer_id_str, duration_sequence_str):
+        try:
+            singer_id = int(singer_id_str)
+            durations = [float(d.strip()) for d in duration_sequence_str.split(',')]
+            
+            if len(durations) > MAX_SEQ_LENGTH:
+                durations = durations[:MAX_SEQ_LENGTH]
+            else:
+                durations += [0] * (MAX_SEQ_LENGTH - len(durations))
+
+            duration_tensor = torch.tensor(durations, dtype=torch.float32).unsqueeze(0)
+            sid_tensor = torch.full_like(duration_tensor, singer_id, dtype=torch.long)
+
+            with torch.no_grad():
+                prediction = best_model(duration_tensor, sid_tensor)
+            
+            output_sequence = prediction.squeeze().tolist()
+            return ", ".join([f"{x:.4f}" for x in output_sequence])
+
+        except Exception as e:
+            return f"오류 발생: {e}"
+
+    iface = gr.Interface(
+        fn=predict_duration,
+        inputs=[
+            gr.Textbox(label="가수 ID (Singer ID)", value="10"),
+            gr.Textbox(label="음표 길이 시퀀스 (쉼표로 구분)", 
+                       value="0.1, 0.2, 0.15, 0.5, 0.4, 0.12, 0.1, 0.25")
+        ],
+        outputs=gr.Textbox(label="예측된 음표 길이 시퀀스"),
+        title="🎵 Duration Predictor (리듬 표현 예측기)",
+        description="가수 ID와 정규 음표 길이 시퀀스를 입력하면, 해당 가수의 고유한 리듬 표현이 적용된 음표 길이를 예측합니다."
+    )
+    
+    iface.launch()

src/toy_duration_predictor/_legacy/train_torch_vanilla.py

@@ -0,0 +1,172 @@
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader, TensorDataset, random_split
+import numpy as np
+import os
+import copy
+
+# --- 1. 모델 아키텍처 및 하이퍼파라미터 정의 ---
+# 이 값들은 실제 데이터셋과 실험 목적에 맞게 조정될 수 있습니다.
+
+# 데이터 관련 파라미터
+MAX_SEQ_LENGTH = 32
+NUM_SINGERS = 100
+NUM_SAMPLES = 100000  # 10만개 샘플
+BATCH_SIZE = 256
+
+# 모델 구조 관련 파라미터
+SID_EMBEDDING_DIM = 16
+GRU_UNITS = 128
+NUM_GRU_LAYERS = 2
+DROPOUT_RATE = 0.3
+
+# 훈련 관련 파라미터
+LEARNING_RATE = 0.001
+NUM_EPOCHS = 50       # 최대 훈련 에포크 수
+EARLY_STOPPING_PATIENCE = 5 # 검증 성능이 5 에포크 동안 개선되지 않으면 조기 종료
+
+class DurationPredictorGRU(nn.Module):
+    """
+    가수 ID(SID)와 음표 길이 시퀀스를 입력받아,
+    표현력 있는(expressive) 음표 길이 시퀀스를 예측하는 양방향 GRU 모델입니다.
+    """
+    def __init__(self):
+        super(DurationPredictorGRU, self).__init__()
+        self.sid_embedding = nn.Embedding(NUM_SINGERS, SID_EMBEDDING_DIM)
+        gru_input_dim = 1 + SID_EMBEDDING_DIM
+        self.gru = nn.GRU(
+            input_size=gru_input_dim,
+            hidden_size=GRU_UNITS,
+            num_layers=NUM_GRU_LAYERS,
+            batch_first=True,
+            bidirectional=True,
+            dropout=DROPOUT_RATE if NUM_GRU_LAYERS > 1 else 0
+        )
+        self.fc_out = nn.Linear(GRU_UNITS * 2, 1)
+
+    def forward(self, duration_input, sid_input):
+        sid_embedded = self.sid_embedding(sid_input)
+        duration_reshaped = duration_input.unsqueeze(-1)
+        x = torch.cat([duration_reshaped, sid_embedded], dim=-1)
+        gru_output, _ = self.gru(x)
+        predictions = self.fc_out(gru_output)
+        return predictions
+
+# --- 2. 데이터 준비 ---
+print("--- 데이터셋 준비 중... ---")
+# 가상의 전체 데이터셋 생성
+durations = torch.rand(NUM_SAMPLES, MAX_SEQ_LENGTH)
+sids = torch.randint(0, NUM_SINGERS, (NUM_SAMPLES, MAX_SEQ_LENGTH))
+labels = (durations * torch.rand_like(durations) * 2).unsqueeze(-1)
+full_dataset = TensorDataset(durations, sids, labels)
+
+# 훈련(80%), 검증(10%), 테스트(10%) 데이터셋으로 분할
+train_size = int(0.8 * NUM_SAMPLES)
+val_size = int(0.1 * NUM_SAMPLES)
+test_size = NUM_SAMPLES - train_size - val_size
+train_dataset, val_dataset, test_dataset = random_split(
+    full_dataset, [train_size, val_size, test_size],
+    generator=torch.Generator().manual_seed(42)
+)
+
+# 각 데이터셋을 위한 DataLoader 생성
+train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
+val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE)
+test_loader = DataLoader(test_dataset, batch_size=BATCH_SIZE)
+
+print(f"훈련 데이터 샘플 수: {len(train_dataset)}")
+print(f"검증 데이터 샘플 수: {len(val_dataset)}")
+print(f"테스트 데이터 샘플 수: {len(test_dataset)}")
+
+# --- 3. 훈련 및 검증 루프 ---
+print("\n--- 모델 훈련 시작... ---")
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+model = DurationPredictorGRU().to(device)
+optimizer = torch.optim.Adam(model.parameters(), lr=LEARNING_RATE)
+loss_fn = nn.MSELoss()
+
+# 최적 모델 저장을 위한 변수 초기화
+best_val_loss = float('inf')
+patience_counter = 0
+best_model_state = None
+
+for epoch in range(NUM_EPOCHS):
+    # --- 훈련 단계 ---
+    model.train() # 모델을 훈련 모드로 설정
+    total_train_loss = 0
+    for batch_idx, (duration, sid, label) in enumerate(train_loader):
+        duration, sid, label = duration.to(device), sid.to(device), label.to(device)
+
+        # 순전파
+        predictions = model(duration, sid)
+        loss = loss_fn(predictions, label)
+
+        # 역전파
+        optimizer.zero_grad()
+        loss.backward()
+        optimizer.step()
+
+        total_train_loss += loss.item()
+    
+    avg_train_loss = total_train_loss / len(train_loader)
+
+    # --- 검증 단계 ---
+    model.eval() # 모델을 평가 모드로 설정
+    total_val_loss = 0
+    with torch.no_grad(): # 그래디언트 계산 비활성화
+        for duration, sid, label in val_loader:
+            duration, sid, label = duration.to(device), sid.to(device), label.to(device)
+            predictions = model(duration, sid)
+            loss = loss_fn(predictions, label)
+            total_val_loss += loss.item()
+
+    avg_val_loss = total_val_loss / len(val_loader)
+
+    print(f"Epoch [{epoch+1}/{NUM_EPOCHS}], Train Loss: {avg_train_loss:.6f}, Val Loss: {avg_val_loss:.6f}")
+
+    # --- 체크포인팅 및 조기 종료 로직 ---
+    # 검증 손실이 개선되었는지 확인
+    if avg_val_loss < best_val_loss:
+        best_val_loss = avg_val_loss
+        # 가장 좋은 모델의 가중치를 deepcopy로 저장
+        best_model_state = copy.deepcopy(model.state_dict())
+        patience_counter = 0 # 인내심 카운터 초기화
+        print(f"  -> 검증 성능 개선! 최적 모델 저장됨. (Val Loss: {best_val_loss:.6f})")
+    else:
+        patience_counter += 1
+        print(f"  -> 검증 성능 개선 없음. (Patience: {patience_counter}/{EARLY_STOPPING_PATIENCE})")
+
+    # 조기 종료 조건 확인
+    if patience_counter >= EARLY_STOPPING_PATIENCE:
+        print(f"\n조기 종료: {EARLY_STOPPING_PATIENCE} 에포크 동안 검증 성능 개선이 없어 훈련을 중단합니다.")
+        break
+
+# --- 4. 최종 모델 평가 (테스트 단계) ---
+print("\n--- 최종 모델 평가 시작 (테스트 데이터셋 사용)... ---")
+
+# 저장된 최적의 모델 가중치를 불러오기
+if best_model_state:
+    model.load_state_dict(best_model_state)
+else:
+    print("경고: 저장된 최적 모델이 없습니다. 마지막 에포크 모델로 평가합니다.")
+
+model.eval()
+total_test_loss = 0
+with torch.no_grad():
+    for duration, sid, label in test_loader:
+        duration, sid, label = duration.to(device), sid.to(device), label.to(device)
+        predictions = model(duration, sid)
+        loss = loss_fn(predictions, label)
+        total_test_loss += loss.item()
+
+avg_test_loss = total_test_loss / len(test_loader)
+
+print("-" * 50)
+print(f"최종 테스트 손실 (MSE): {avg_test_loss:.6f}")
+print("이것이 논문에 보고할 최종 모델의 일반화 성능입니다.")
+print("-" * 50)
+
+# (선택) 최적 모델 가중치 파일로 저장
+if best_model_state:
+    torch.save(best_model_state, 'best_duration_predictor.pth')
+    print("최적 모델 가중치가 'best_duration_predictor.pth' 파일로 저장되었습니다.")

src/toy_duration_predictor/preprocess/mssv.py

@@ -0,0 +1,188 @@
+from pathlib import Path
+import re
+from typing import List, Dict, Any
+
+import ray
+import pandas as pd
+import midii
+
+from .utils import (
+    _preprocess_sort_by_start_time,
+    _preprocess_remove_front_back_silence,
+    _preprocess_silence_pitch_zero,
+    _preprocess_merge_silence,
+    _preprocess_remove_short_silence,
+    _preprocess_add_quantized_duration_col,
+)
+
+
+def singer_id_from_filepath(filepath):
+    return int(re.findall(r"s\d\d", filepath)[0][1:])
+
+
+def midi_to_note_list(midi_filepath, quantize=False):
+    try:
+        mid = midii.MidiFile(
+            midi_filepath, convert_1_to_0=True, lyric_encoding="utf-8"
+        )
+        mid.lyrics
+    except:  # noqa: E722
+        mid = midii.MidiFile(
+            midi_filepath, convert_1_to_0=True, lyric_encoding="cp949"
+        )
+    if quantize:
+        mid.quantize(unit="32")
+    data = []
+    total_duration = 0
+    residual_duration = 0
+    active_note = {}
+    silence_note = {}
+    for msg in mid.tracks[0]:
+        msg_end_time = total_duration + msg.time
+        if msg.type == "note_on" and msg.velocity > 0:
+            residual_duration += msg.time
+            if residual_duration > 0:
+                if not silence_note:
+                    silence_note = {
+                        "start_time": total_duration,
+                        "pitch": 0,
+                        "lyric": " ",
+                    }
+                silence_note["end_time"] = msg_end_time
+                silence_note["duration"] = (
+                    msg_end_time - silence_note["start_time"]
+                )
+                data.append(silence_note.copy())
+                silence_note.clear()
+                residual_duration = 0
+            active_note = {
+                "start_time": msg_end_time,
+                "pitch": msg.note,
+            }
+        elif msg.type == "lyrics":
+            active_note["lyric"] = midii.MessageAnalyzer_lyrics(
+                msg=msg, encoding=mid.lyric_encoding
+            ).lyric
+        elif msg.type == "note_off" or (
+            msg.type == "note_on" and msg.velocity == 0
+        ):
+            active_note["end_time"] = msg_end_time
+            active_note["duration"] = msg_end_time - active_note["start_time"]
+            data.append(active_note.copy())
+            active_note.clear()
+        else:
+            if not active_note and not silence_note:
+                silence_note = {
+                    "start_time": total_duration,
+                    "pitch": 0,
+                    "lyric": " ",
+                }
+            if not active_note:
+                residual_duration += msg.time
+        total_duration = msg_end_time
+
+    return data, mid.ticks_per_beat
+
+
+def _preprocess_slice_actual_lyric(df):
+    j_indices = df.index[df["lyric"] == "J"].tolist()
+    idx_j = j_indices[0]
+    h_indices = df.index[df["lyric"] == "H"].tolist()
+    idx_h = h_indices[0]
+    slice_start_index = idx_j + 1
+    slice_end_index = idx_h
+    df = df.iloc[slice_start_index:slice_end_index].reset_index(drop=True)
+    return df
+
+
+def preprocess_notes(notes, ticks_per_beat, unit="32"):
+    df = pd.DataFrame(notes)
+
+    # ["J":"H"]
+    df = _preprocess_slice_actual_lyric(df)
+    # sort by time
+    df = _preprocess_sort_by_start_time(df)
+    # remove front & back silence
+    df = _preprocess_remove_front_back_silence(df)
+    # lyric=" " --> pitch=0
+    df = _preprocess_silence_pitch_zero(df)
+    # merge lyric=" " items
+    df = _preprocess_merge_silence(df)
+    # remove silence < 0.3
+    df = _preprocess_remove_short_silence(df, 0.3)
+    #
+    df = _preprocess_add_quantized_duration_col(df, ticks_per_beat, unit=unit)
+    return df
+
+
+def process_midi_flat_map(row: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """
+    Processes a single file path. Returns a list containing one dictionary on success,
+    or an empty list on failure. Designed for use with flat_map().
+    """
+    file_path_str = row["path"]
+    try:
+        mssv_sample_list, ticks_per_beat = midi_to_note_list(file_path_str)
+        df = preprocess_notes(mssv_sample_list, ticks_per_beat=ticks_per_beat)
+        singer_id = singer_id_from_filepath(file_path_str)
+
+        durations = df["duration"].tolist()
+        quantized_durations = df["quantized_duration"].tolist()
+
+        return [
+            {
+                "durations": durations,
+                "quantized_durations": quantized_durations,
+                "singer_id": singer_id,
+            }
+        ]
+    except Exception as e:
+        print(f"CRITICAL ERROR processing {file_path_str}: {e}")
+        return []
+
+
+def preprocess_dataset(midi_file_directory, output_parquet_path):
+    context = ray.init()
+    print(context.dashboard_url)
+    print(f"Ray cluster started: {ray.cluster_resources()}")
+
+    all_midi_paths = list(Path(midi_file_directory).rglob("*.mid"))
+    print(f"Found {len(all_midi_paths)} MIDI files to process.")
+
+    # --- Step 2: Create a dataset of file paths ---
+    # This is the idiomatic way. We create a dataset where each row is a path.
+    # We pass a list of dictionaries to give the column a name: "path".
+    print("Creating initial dataset of file paths...")
+    ds = ray.data.from_items([{"path": str(p)} for p in all_midi_paths])
+
+    # --- Step 3: Use .map() to process files in parallel ---
+    # .map() applies the function to each row of the dataset in parallel.
+    # Ray Data manages the tasks, memory, and scheduling for you.
+    print("Applying parallel processing function using .map()...")
+    processed_ds = ds.flat_map(process_midi_flat_map)
+
+    # --- Step 4: Filter out any rows that failed ---
+    # Our function returns an empty dict on failure, so we filter those out.
+    # processed_ds = processed_ds.filter(lambda row: row)
+
+    # Now, let's inspect the schema of the PROCESSED dataset
+    print("\nProcessed Dataset schema:")
+    print(processed_ds.schema())  # Corrected print statement
+
+    print("\nFirst 1 rows of processed data:")
+    processed_ds.show(1)
+
+    # You chose to repartition to 1, which is fine if the total dataset is < 1GB.
+    # This will create a single Parquet file in the output directory.
+    print("Repartitioning dataset...")
+    processed_ds = processed_ds.repartition(num_blocks=1)
+
+    print(f"\nWriting dataset to Parquet format at: {output_parquet_path}")
+    processed_ds.write_parquet(output_parquet_path)
+
+    # This will now show the correct number of processed files (rows in the dataset)
+    print(
+        f"\nProcessing complete! {processed_ds.count()} files successfully processed."
+    )
+
+    ray.shutdown()

src/toy_duration_predictor/preprocess/utils.py

@@ -0,0 +1,117 @@
+from pathlib import Path
+
+import pandas as pd
+import midii
+
+
+def get_files(dir_path, type, sort=False):
+    paths = Path(dir_path).rglob(f"*.{type}")
+    if sort:
+        return sorted(paths, key=lambda p: p.stem)
+    else:
+        return paths
+
+
+def _preprocess_remove_front_back_silence(df):
+    is_valid_lyric = df["lyric"] != " "
+    valid_indices = df.index[is_valid_lyric].tolist()
+    first_valid_idx = valid_indices[0]
+    last_valid_idx = valid_indices[-1]
+    df = df.iloc[first_valid_idx : last_valid_idx + 1].reset_index(drop=True)
+    return df
+
+
+def _preprocess_sort_by_start_time(df):
+    df = df.sort_values(by="start_time").reset_index(drop=True)
+    return df
+
+
+def _preprocess_remove_front_back_silence(df):
+    is_valid_lyric = df["lyric"] != " "
+    valid_indices = df.index[is_valid_lyric].tolist()
+    first_valid_idx = valid_indices[0]
+    last_valid_idx = valid_indices[-1]
+    df = df.iloc[first_valid_idx : last_valid_idx + 1].reset_index(drop=True)
+    return df
+
+
+def _preprocess_silence_pitch_zero(df):
+    df.loc[df["lyric"] == " ", "pitch"] = 0
+    return df
+
+
+def _preprocess_merge_silence(df):
+    output_notes = []
+    i = 0
+    n = len(df)
+    while i < n:
+        current_row = df.iloc[i]  # Pandas Series
+        if current_row["lyric"] == " ":
+            merged_start_time = current_row["start_time"]
+            merged_end_time = current_row["end_time"]
+
+            j = i + 1
+            while j < n and df.iloc[j]["lyric"] == " ":
+                merged_end_time = df.iloc[j][
+                    "end_time"
+                ]  # 마지막 공백의 end_time으로 업데이트
+                j += 1
+
+            merged_item = {
+                "start_time": merged_start_time,
+                "end_time": merged_end_time,
+                "pitch": 0,
+                "lyric": " ",
+                "duration": merged_end_time - merged_start_time,
+            }
+            output_notes.append(merged_item)
+            i = j  # 병합된 블록 다음으로 인덱스 이동
+        else:
+            non_space_item = {
+                "start_time": current_row["start_time"],
+                "end_time": current_row["end_time"],
+                "pitch": current_row["pitch"],
+                "lyric": current_row["lyric"],
+                "duration": current_row["duration"],
+            }
+            output_notes.append(non_space_item)
+            i += 1
+    df = pd.DataFrame(output_notes)
+    return df
+
+
+def _preprocess_remove_short_silence(df, threshold=0.3):
+    processed_notes = []
+    absorbed_time = 0.0
+
+    for i in range(len(df)):
+        current_note_s = df.iloc[i]
+        if (
+            current_note_s["lyric"] == " "
+            and current_note_s["duration"] < threshold
+        ):
+            absorbed_time += current_note_s["duration"]
+            continue
+        else:
+            note_to_add = current_note_s.to_dict()
+            if absorbed_time > 0:
+                note_to_add["start_time"] -= absorbed_time
+                note_to_add["duration"] = (
+                    note_to_add["end_time"] - note_to_add["start_time"]
+                )
+                absorbed_time = 0.0
+            processed_notes.append(note_to_add)
+
+    df = pd.DataFrame(processed_notes)
+    return df
+
+
+def _preprocess_add_quantized_duration_col(df, ticks_per_beat, unit="32"):
+    unit_tick = midii.beat2tick(
+        midii.NOTE[f"n/{unit}"].beat, ticks_per_beat=ticks_per_beat
+    )
+    df["quantized_duration"], _ = midii.quantize(
+        df["duration"].values, unit=unit_tick
+    )
+
+    return df

src/toy_duration_predictor/train_fastai.py

@@ -0,0 +1,297 @@
+import torch
+import torch.nn as nn
+from datasets import load_dataset, DatasetDict
+from fastai.vision.all import *  # Import a more general base for DataBlock
+
+# Import Ray and the correct modern modules
+import ray
+from ray import tune
+from ray.air import session  # The new way to report metrics
+from ray.tune.search.optuna import OptunaSearch
+
+# Import tools for logging and demos
+import wandb
+from ray.air.integrations.wandb import WandbLoggerCallback
+import gradio as gr
+
+# --- 1. Configuration & Hyperparameters ---
+# You can adjust these values for your experiments
+
+# Data Parameters
+REPO_ID = "ccss17/note-duration-dataset"
+SEQUENCE_LENGTH = 128
+PAD_TOKEN = 0
+BATCH_SIZE = 64
+
+# Model Parameters
+NUM_SINGERS = 18  # IMPORTANT: Set this to the total number of unique singers in your dataset
+SINGER_EMBEDDING_DIM = 16
+HIDDEN_SIZE = 256
+NUM_LAYERS = 2
+DROPOUT = 0.3
+
+# --- NEW: Configuration for wandb ---
+# Set this to False if you want to run the script without logging to wandb
+WANDB_ENABLED = True
+WANDB_ENTITY = "ccss17"  # Your wandb username
+WANDB_PROJECT = "toy-duration-predictor"
+
+# --- 2. Model Architecture Definition (PyTorch) ---
+# A bi-GRU model that takes a sequence and a singer ID as input.
+
+
+class ToyDurationPredictor(nn.Module):
+    def __init__(
+        self,
+        vocab_size,
+        embedding_dim,
+        hidden_size,
+        num_layers,
+        dropout,
+        num_singers,
+        singer_embedding_dim,
+    ):
+        super().__init__()
+        self.num_singers = num_singers
+
+        # Embedding layer for the singer ID
+        self.singer_embedding = nn.Embedding(num_singers, singer_embedding_dim)
+
+        # Bi-directional GRU layers
+        self.rnn = nn.GRU(
+            input_size=1 + singer_embedding_dim,
+            hidden_size=hidden_size,
+            num_layers=num_layers,
+            bidirectional=True,
+            dropout=dropout,
+            batch_first=True,
+        )
+
+        self.fc = nn.Linear(hidden_size * 2, 1)
+
+    def forward(self, x):
+        # fastai's DataBlock will pass the input as a tuple (inp, sid)
+        x_seq, x_sid = x
+
+        # Add a feature dimension to the input sequence
+        x_seq = x_seq.unsqueeze(-1).float()
+
+        # Get singer embedding
+        sid_emb = self.singer_embedding(x_sid)
+
+        # Repeat the singer embedding for each step in the sequence
+        sid_emb_expanded = sid_emb.unsqueeze(1).expand(-1, x_seq.size(1), -1)
+
+        # Concatenate the duration sequence with the singer embedding
+        combined_input = torch.cat([x_seq, sid_emb_expanded], dim=-1)
+
+        # Pass through the GRU
+        outputs, _ = self.rnn(combined_input)
+
+        # Pass through the final fully connected layer
+        prediction = self.fc(outputs)
+
+        return prediction.squeeze(-1)
+
+
+# --- 3. Data Loading and Preparation ---
+
+
+def get_dataloaders(model_type="B"):
+    """
+    Loads data from the Hub, splits it, processes it, and returns DataLoaders.
+    """
+    print(f"--- Preparing DataLoaders for Model {model_type} ---")
+    dataset = load_dataset(REPO_ID, split="train")
+
+    # Perform the 80/10/10 split
+    train_test_split = dataset.train_test_split(test_size=0.2, seed=42)
+    test_valid_split = train_test_split["test"].train_test_split(
+        test_size=0.5, seed=42
+    )
+    split_dataset = DatasetDict(
+        {
+            "train": train_test_split["train"],
+            "valid": test_valid_split["train"],
+            "test": test_valid_split["test"],
+        }
+    )
+
+    # The chunking function with padding remains the same
+    def chunk_examples_with_padding(examples):
+        chunked = {"durations": [], "quantized_durations": [], "singer_id": []}
+        for i in range(len(examples["durations"])):
+            durs, q_durs, sid = (
+                examples["durations"][i],
+                examples["quantized_durations"][i],
+                examples["singer_id"][i],
+            )
+            for j in range(0, len(durs), SEQUENCE_LENGTH):
+                d_chunk = durs[j : j + SEQUENCE_LENGTH]
+                q_chunk = q_durs[j : j + SEQUENCE_LENGTH]
+                if len(d_chunk) < SEQUENCE_LENGTH:
+                    padding_needed = SEQUENCE_LENGTH - len(d_chunk)
+                    d_chunk.extend([PAD_TOKEN] * padding_needed)
+                    q_chunk.extend([PAD_TOKEN] * padding_needed)
+                chunked["durations"].append(d_chunk)
+                chunked["quantized_durations"].append(q_chunk)
+                chunked["singer_id"].append(sid)
+        return chunked
+
+    processed_splits = split_dataset.map(
+        chunk_examples_with_padding,
+        batched=True,
+        remove_columns=split_dataset["train"].column_names,
+    )
+
+    # --- NEW: Simpler, more robust DataBlock setup ---
+
+    # Define functions to get the inputs (x) and target (y) from a row
+    def get_x(row):
+        # The input is a tuple of the sequence and the singer id
+        if model_type == "A":
+            seq = torch.tensor(row["durations"], dtype=torch.long)
+        else:
+            seq = torch.tensor(row["quantized_durations"], dtype=torch.long)
+        sid = torch.tensor(row["singer_id"], dtype=torch.long)
+        return (seq, sid)
+
+    def get_y(row):
+        # The target is always the original durations, as a float for regression
+        return torch.tensor(row["durations"], dtype=torch.float32)
+
+    # Create the DataBlock
+    dblock = DataBlock(
+        blocks=(
+            TransformBlock,
+            RegressionBlock,
+        ),  # A generic transform block and a regression block
+        get_x=get_x,
+        get_y=get_y,
+        splitter=IndexSplitter(
+            split_dataset["valid"]._indices
+        ),  # Use indices for splitting
+    )
+
+    # Create the DataLoaders from the processed training set
+    dls = dblock.dataloaders(processed_splits["train"], bs=BATCH_SIZE)
+
+    return dls, processed_splits["test"]
+
+
+# --- 4. Custom Callback for Ray Tune + fastai Integration ---
+class TuneReportCallbackForFastAI(Callback):
+    def after_epoch(self):
+        train_loss = self.learn.recorder.smooth_loss.item()
+        valid_loss = self.learn.recorder.val_loss.item()
+        mae_metric = self.learn.recorder.metrics[0].value.item()
+        session.report(
+            {
+                "train_loss": train_loss,
+                "valid_loss": valid_loss,
+                "mae": mae_metric,
+            }
+        )
+
+
+# --- 5. Training Function for Ray Tune ---
+def train_tdp(config):
+    model_type = config.pop("model_type", "B")
+    dls, _ = get_dataloaders(model_type=model_type)
+
+    model = ToyDurationPredictor(
+        vocab_size=0,
+        embedding_dim=0,
+        hidden_size=config["hidden_size"],
+        num_layers=config["num_layers"],
+        dropout=config["dropout"],
+        num_singers=NUM_SINGERS,
+        singer_embedding_dim=SINGER_EMBEDDING_DIM,
+    )
+
+    learn = Learner(dls, model, loss_func=MSELossFlat(), metrics=mae).to_fp16()
+    callbacks = [TuneReportCallbackForFastAI()]
+    learn.fit_one_cycle(config["epochs"], lr_max=config["lr"], cbs=callbacks)
+
+
+# --- 6. Main Execution Block ---
+if __name__ == "__main__":
+    # --- Option 1: Run a single training for quick testing ---
+    print("--- Starting Single Training Run for Model B (Your Method) ---")
+
+    if WANDB_ENABLED:
+        try:
+            wandb.init(
+                project=WANDB_PROJECT,
+                entity=WANDB_ENTITY,
+                name="single_run_model_b",
+            )
+        except Exception as e:
+            print(f"Could not initialize wandb: {e}. Disabling for this run.")
+            WANDB_ENABLED = False
+
+    dls_B, test_ds_B = get_dataloaders(model_type="B")
+
+    model = ToyDurationPredictor(
+        vocab_size=0,
+        embedding_dim=0,
+        hidden_size=HIDDEN_SIZE,
+        num_layers=NUM_LAYERS,
+        dropout=DROPOUT,
+        num_singers=NUM_SINGERS,
+        singer_embedding_dim=SINGER_EMBEDDING_DIM,
+    )
+
+    callbacks = [WandbCallback()] if WANDB_ENABLED else []
+    learn = Learner(
+        dls_B, model, loss_func=MSELossFlat(), metrics=mae, cbs=callbacks
+    )
+
+    print("Training the model...")
+    learn.fit_one_cycle(5, 1e-3)
+
+    print("\n--- Evaluating on the held-out test set ---")
+    test_dl = dls_B.test_dl(test_ds_B)
+    loss, mae_val = learn.validate(dl=test_dl)
+    print(
+        f"\nFinal Test Set Performance: Loss (MSE)={loss:.4f}, MAE={mae_val:.4f} ticks"
+    )
+
+    if WANDB_ENABLED:
+        wandb.finish()
+
+    # --- Gradio Demo Section ---
+    print("\n--- Launching Gradio Demo ---")
+
+    def predict_durations(quantized_durations_str, singer_id):
+        try:
+            durs = [int(x.strip()) for x in quantized_durations_str.split(",")]
+            inp_tensor = torch.tensor(durs, dtype=torch.long)
+            sid_tensor = torch.tensor([int(singer_id)], dtype=torch.long)
+
+            # The input to the learner's test_dl is a list of items
+            # Each item should match what get_x would produce
+            dl = learn.dls.test_dl([(inp_tensor, sid_tensor)])
+            preds, _ = learn.get_preds(dl=dl)
+
+            return ", ".join([str(int(p)) for p in preds[0]])
+        except Exception as e:
+            return f"Error: {e}"
+
+    iface = gr.Interface(
+        fn=predict_durations,
+        inputs=[
+            gr.Textbox(
+                label="Quantized Durations (comma-separated)",
+                placeholder="30, 0, 75, 0, 45, 15, ...",
+            ),
+            gr.Number(label="Singer ID", value=2),
+        ],
+        outputs=gr.Textbox(label="Predicted Original Durations"),
+        title="Toy Duration Predictor",
+        description="Enter a sequence of quantized durations and a singer ID to see the model predict the original, stylistic performance.",
+    )
+    iface.launch()
+
+    # --- Option 2: Run Ray Tune ---
+    # ... (Ray Tune code remains the same) ...

src/toy_duration_predictor/train_lightning.py

@@ -0,0 +1,350 @@
+import torch
+import torch.nn as nn
+from torch.utils.data import DataLoader, Dataset
+from datasets import load_dataset, DatasetDict
+import pytorch_lightning as pl
+from pytorch_lightning.loggers import WandbLogger
+from pytorch_lightning.callbacks import TQDMProgressBar
+
+# Import Ray and Tune for hyperparameter search
+from ray import tune
+from ray.tune.integration.pytorch_lightning import TuneReportCallback
+
+# Import tools for demos
+import wandb
+import gradio as gr
+import pandas as pd
+import numpy as np
+
+# --- 1. Configuration & Hyperparameters ---
+# Data Parameters
+REPO_ID = "ccss17/note-duration-dataset"
+SEQUENCE_LENGTH = 128
+PAD_TOKEN = 0
+BATCH_SIZE = 64
+
+# Model Parameters are now set dynamically in the DataModule
+SINGER_EMBEDDING_DIM = 16
+HIDDEN_SIZE = 256
+NUM_LAYERS = 2
+DROPOUT = 0.3
+LEARNING_RATE = 1e-3
+
+# Configuration for wandb
+WANDB_ENABLED = True
+WANDB_ENTITY = "ccss17"  # Your wandb username
+WANDB_PROJECT = "toy-duration-predictor-lightning"
+
+
+# --- 2. Model Architecture Definition (Vanilla PyTorch) ---
+class ToyDurationPredictor(nn.Module):
+    def __init__(
+        self,
+        hidden_size,
+        num_layers,
+        dropout,
+        num_singers,
+        singer_embedding_dim,
+    ):
+        super().__init__()
+        self.singer_embedding = nn.Embedding(num_singers, singer_embedding_dim)
+        self.rnn = nn.GRU(
+            input_size=1 + singer_embedding_dim,
+            hidden_size=hidden_size,
+            num_layers=num_layers,
+            bidirectional=True,
+            dropout=dropout,
+            batch_first=True,
+        )
+        self.fc = nn.Linear(hidden_size * 2, 1)
+
+    def forward(self, x_seq, x_sid):
+        x_seq = x_seq.unsqueeze(-1).float()
+        sid_emb = self.singer_embedding(x_sid)
+        sid_emb_expanded = sid_emb.unsqueeze(1).expand(-1, x_seq.size(1), -1)
+        combined_input = torch.cat([x_seq, sid_emb_expanded], dim=-1)
+        outputs, _ = self.rnn(combined_input)
+        prediction = self.fc(outputs)
+        return prediction.squeeze(-1)
+
+
+# --- 3. Data Preparation (PyTorch Dataset & Lightning DataModule) ---
+class DurationDataset(Dataset):
+    """A standard PyTorch Dataset for our chunked data."""
+
+    def __init__(self, processed_hf_dataset, model_type="B"):
+        self.data = processed_hf_dataset
+        self.model_type = model_type
+
+    def __len__(self):
+        return len(self.data)
+
+    def __getitem__(self, idx):
+        item = self.data[idx]
+        label = torch.tensor(item["durations"], dtype=torch.float32)
+
+        if self.model_type == "A":
+            input_seq = torch.tensor(item["durations"], dtype=torch.long)
+        else:
+            input_seq = torch.tensor(
+                item["quantized_durations"], dtype=torch.long
+            )
+
+        # Use the NEW re-indexed singer_idx column
+        singer_idx = torch.tensor(item["singer_idx"], dtype=torch.long)
+
+        return {"input_seq": input_seq, "singer_id": singer_idx}, label
+
+
+class DurationDataModule(pl.LightningDataModule):
+    """A LightningDataModule to handle loading, splitting, and batching."""
+
+    def __init__(self, model_type="B", batch_size=32):
+        super().__init__()
+        self.model_type = model_type
+        self.batch_size = batch_size
+        self.singer_id_map = {}
+        self.num_singers = 0
+
+    def setup(self, stage=None):
+        dataset = load_dataset(REPO_ID, split="train")
+
+        # --- FIX: Create a mapping for singer IDs ---
+        unique_singer_ids = sorted(dataset.unique("singer_id"))
+        self.num_singers = len(unique_singer_ids)
+        self.singer_id_map = {
+            sid: i for i, sid in enumerate(unique_singer_ids)
+        }
+
+        print(
+            f"Found {self.num_singers} unique singers. Mapping IDs to [0, {self.num_singers - 1}]"
+        )
+
+        def map_singer_ids(example):
+            example["singer_idx"] = self.singer_id_map[example["singer_id"]]
+            return example
+
+        dataset = dataset.map(map_singer_ids)
+        # --- END FIX ---
+
+        train_test_split = dataset.train_test_split(test_size=0.2, seed=42)
+        test_valid_split = train_test_split["test"].train_test_split(
+            test_size=0.5, seed=42
+        )
+        split_dataset = DatasetDict(
+            {
+                "train": train_test_split["train"],
+                "valid": test_valid_split["train"],
+                "test": test_valid_split["test"],
+            }
+        )
+
+        def chunk_examples_with_padding(examples):
+            chunked = {
+                "durations": [],
+                "quantized_durations": [],
+                "singer_idx": [],
+            }
+            for i in range(len(examples["durations"])):
+                durs, q_durs, s_idx = (
+                    examples["durations"][i],
+                    examples["quantized_durations"][i],
+                    examples["singer_idx"][i],
+                )
+                for j in range(0, len(durs), SEQUENCE_LENGTH):
+                    d_chunk, q_chunk = (
+                        durs[j : j + SEQUENCE_LENGTH],
+                        q_durs[j : j + SEQUENCE_LENGTH],
+                    )
+                    if len(d_chunk) < SEQUENCE_LENGTH:
+                        padding = [PAD_TOKEN] * (
+                            SEQUENCE_LENGTH - len(d_chunk)
+                        )
+                        d_chunk.extend(padding)
+                        q_chunk.extend(padding)
+                    chunked["durations"].append(d_chunk)
+                    chunked["quantized_durations"].append(q_chunk)
+                    chunked["singer_idx"].append(s_idx)
+            return chunked
+
+        processed_splits = split_dataset.map(
+            chunk_examples_with_padding,
+            batched=True,
+            remove_columns=dataset.column_names,
+        )
+
+        self.train_ds = DurationDataset(
+            processed_splits["train"], self.model_type
+        )
+        self.val_ds = DurationDataset(
+            processed_splits["valid"], self.model_type
+        )
+        self.test_ds = DurationDataset(
+            processed_splits["test"], self.model_type
+        )
+
+    def train_dataloader(self):
+        return DataLoader(
+            self.train_ds,
+            batch_size=self.batch_size,
+            shuffle=True,
+            num_workers=4,
+            persistent_workers=True,
+        )
+
+    def val_dataloader(self):
+        return DataLoader(
+            self.val_ds,
+            batch_size=self.batch_size,
+            num_workers=4,
+            persistent_workers=True,
+        )
+
+    def test_dataloader(self):
+        return DataLoader(
+            self.test_ds,
+            batch_size=self.batch_size,
+            num_workers=4,
+            persistent_workers=True,
+        )
+
+
+# --- 4. The LightningModule ---
+class LightningTDP(pl.LightningModule):
+    def __init__(
+        self,
+        num_singers,
+        model_type="B",
+        hidden_size=HIDDEN_SIZE,
+        num_layers=NUM_LAYERS,
+        dropout=DROPOUT,
+        learning_rate=LEARNING_RATE,
+    ):
+        super().__init__()
+        self.save_hyperparameters()
+        self.model = ToyDurationPredictor(
+            hidden_size=hidden_size,
+            num_layers=num_layers,
+            dropout=dropout,
+            num_singers=num_singers,
+            singer_embedding_dim=SINGER_EMBEDDING_DIM,
+        )
+        self.loss_fn = nn.MSELoss()
+
+    def forward(self, batch):
+        return self.model(batch["input_seq"], batch["singer_id"])
+
+    def training_step(self, batch, batch_idx):
+        inputs, labels = batch
+        preds = self(inputs)
+        loss = self.loss_fn(preds, labels)
+        self.log(
+            "train_loss",
+            loss,
+            on_step=True,
+            on_epoch=True,
+            prog_bar=True,
+            logger=True,
+        )
+        return loss
+
+    def validation_step(self, batch, batch_idx):
+        inputs, labels = batch
+        preds = self(inputs)
+        loss = self.loss_fn(preds, labels)
+        self.log("val_loss", loss, on_epoch=True, prog_bar=True, logger=True)
+
+    def test_step(self, batch, batch_idx):
+        inputs, labels = batch
+        preds = self(inputs)
+        loss = self.loss_fn(preds, labels)
+        mae = nn.functional.l1_loss(preds, labels)
+        self.log("test_loss", loss, prog_bar=True)
+        self.log("test_mae", mae, prog_bar=True)
+
+    def configure_optimizers(self):
+        return torch.optim.Adam(
+            self.parameters(), lr=self.hparams.learning_rate
+        )
+
+
+# --- 5. Main Execution Block ---
+if __name__ == "__main__":
+    print("--- Starting Single Training Run with PyTorch Lightning ---")
+    data_module = DurationDataModule(model_type="B", batch_size=BATCH_SIZE)
+
+    # Must run setup() to access the number of singers
+    data_module.setup()
+
+    lightning_model = LightningTDP(
+        num_singers=data_module.num_singers, model_type="B"
+    )
+
+    wandb_logger = None
+    if WANDB_ENABLED:
+        try:
+            wandb_logger = WandbLogger(
+                project=WANDB_PROJECT,
+                # entity=WANDB_ENTITY,
+                name="lightning_single_run",
+            )
+        except Exception as e:
+            print(f"Could not initialize wandb: {e}. Disabling for this run.")
+
+    trainer = pl.Trainer(
+        max_epochs=5,
+        logger=wandb_logger,
+        callbacks=[TQDMProgressBar(refresh_rate=10)],
+        accelerator="auto",
+    )
+
+    print("Training Model B...")
+    trainer.fit(lightning_model, datamodule=data_module)
+
+    print("\n--- Evaluating on the held-out test set ---")
+    trainer.test(lightning_model, datamodule=data_module)
+
+    if WANDB_ENABLED and wandb.run:
+        wandb.finish()
+
+    print("\n--- Launching Gradio Demo ---")
+    model_for_demo = lightning_model.model.cpu()
+
+    # Need the mapping for the demo
+    singer_id_reverse_map = {
+        v: k for k, v in data_module.singer_id_map.items()
+    }
+
+    def predict_durations(quantized_durations_str, singer_id_from_user):
+        try:
+            # Map the user-provided singer ID to the model's internal index
+            if singer_id_from_user not in data_module.singer_id_map:
+                return f"Error: Singer ID {singer_id_from_user} not found in the dataset."
+            singer_idx = data_module.singer_id_map[singer_id_from_user]
+
+            durs = [int(x.strip()) for x in quantized_durations_str.split(",")]
+            inp_tensor = torch.tensor(durs, dtype=torch.long)
+            sid_tensor = torch.tensor([singer_idx], dtype=torch.long)
+
+            with torch.no_grad():
+                model_for_demo.eval()
+                preds = model_for_demo(inp_tensor.unsqueeze(0), sid_tensor)
+
+            return ", ".join([str(int(p)) for p in preds[0]])
+        except Exception as e:
+            return f"Error: {e}"
+
+    iface = gr.Interface(
+        fn=predict_durations,
+        inputs=[
+            gr.Textbox(
+                label="Quantized Durations (comma-separated)",
+                placeholder="30, 0, 75, 0, 45, 15, ...",
+            ),
+            gr.Number(label="Singer ID (Original)", value=2),
+        ],
+        outputs=gr.Textbox(label="Predicted Original Durations"),
+        title="Toy Duration Predictor (Lightning)",
+        description="Enter a sequence of quantized durations and an original singer ID to see the model predict the original, stylistic performance.",
+    )
+    iface.launch()

src/toy_duration_predictor/upload.py

@@ -0,0 +1,142 @@
+import json
+import torch
+from huggingface_hub import HfApi
+from pathlib import Path
+
+# --- IMPORTANT ---
+# This script assumes you have already run the main training script and have
+# the necessary files and class definitions available.
+
+# Import your model class from your training script.
+# Make sure the path is correct. For example, if your training script is in 'src/train.py':
+# from src.train import LightningTDP, ToyDurationPredictor
+#
+# For this example, we will define the classes again to make the script standalone.
+# In your real project, you should import them.
+import torch.nn as nn
+import pytorch_lightning as pl
+
+
+class ToyDurationPredictor(nn.Module):
+    def __init__(
+        self,
+        hidden_size,
+        num_layers,
+        dropout,
+        num_singers,
+        singer_embedding_dim,
+    ):
+        super().__init__()
+        self.singer_embedding = nn.Embedding(num_singers, singer_embedding_dim)
+        self.rnn = nn.GRU(
+            input_size=1 + singer_embedding_dim,
+            hidden_size=hidden_size,
+            num_layers=num_layers,
+            bidirectional=True,
+            dropout=dropout,
+            batch_first=True,
+        )
+        self.fc = nn.Linear(hidden_size * 2, 1)
+
+    def forward(self, x_seq, x_sid):
+        x_seq = x_seq.unsqueeze(-1).float()
+        sid_emb = self.singer_embedding(x_sid)
+        sid_emb_expanded = sid_emb.unsqueeze(1).expand(-1, x_seq.size(1), -1)
+        combined_input = torch.cat([x_seq, sid_emb_expanded], dim=-1)
+        outputs, _ = self.rnn(combined_input)
+        prediction = self.fc(outputs)
+        return prediction.squeeze(-1)
+
+
+class LightningTDP(pl.LightningModule):
+    def __init__(
+        self, num_singers, learning_rate, hidden_size, num_layers, dropout
+    ):
+        super().__init__()
+        self.save_hyperparameters()
+        self.model = ToyDurationPredictor(
+            hidden_size=hidden_size,
+            num_layers=num_layers,
+            dropout=dropout,
+            num_singers=num_singers,
+            singer_embedding_dim=16,  # Assuming a fixed value
+        )
+
+
+# --- 1. Configuration ---
+# The path to the best checkpoint saved by PyTorch Lightning
+BEST_MODEL_PATH = "./checkpoints/best-model-B.ckpt"
+# Your Hugging Face username and the desired model repo name
+REPO_ID = "ccss17/toy-duration-predictor"
+# A local directory to stage your files before uploading
+STAGING_DIR = Path("./hf_upload_staging")
+
+
+def upload_model_to_hub():
+    """
+    Loads a model from a checkpoint, prepares all necessary files,
+    and uploads them to the Hugging Face Hub.
+    """
+    # Create the staging directory if it doesn't exist
+    STAGING_DIR.mkdir(exist_ok=True)
+
+    # --- 2. Load the final model and extract its state ---
+    print(f"Loading best model from: {BEST_MODEL_PATH}")
+    try:
+        lightning_model = LightningTDP.load_from_checkpoint(BEST_MODEL_PATH)
+    except FileNotFoundError:
+        print(f"ERROR: Checkpoint file not found at {BEST_MODEL_PATH}")
+        print("Please make sure you have run the training script first.")
+        return
+
+    # Extract the underlying PyTorch model (the weights)
+    final_pytorch_model = lightning_model.model
+
+    # Save the model's weights in the standard Hugging Face format
+    weights_path = STAGING_DIR / "pytorch_model.bin"
+    torch.save(final_pytorch_model.state_dict(), weights_path)
+    print(f"Model weights saved to {weights_path}")
+
+    # --- 3. Create the configuration file ---
+    # This saves all the hyperparameters needed to recreate the model architecture
+    config = {
+        "hidden_size": lightning_model.hparams.hidden_size,
+        "num_layers": lightning_model.hparams.num_layers,
+        "dropout": lightning_model.hparams.dropout,
+        "num_singers": lightning_model.hparams.num_singers,
+        "singer_embedding_dim": 16,  # Assuming a fixed value
+        "architectures": [
+            "ToyDurationPredictor"
+        ],  # Link to the model class name
+    }
+    config_path = STAGING_DIR / "config.json"
+    with open(config_path, "w") as f:
+        json.dump(config, f, indent=4)
+    print(f"Model configuration saved to {config_path}")
+
+    # --- 4. Upload all files to the Hub ---
+    print(f"\nUploading files to repository: {REPO_ID}")
+
+    # Ensure you are logged in
+    # In your terminal run: huggingface-cli login
+    api = HfApi()
+
+    # Create the repository on the Hub (if it doesn't exist)
+    api.create_repo(repo_id=REPO_ID, repo_type="model", exist_ok=True)
+
+    # Upload the entire staging folder
+    api.upload_folder(
+        folder_path=STAGING_DIR,
+        repo_id=REPO_ID,
+        repo_type="model",
+    )
+
+    print("\nUpload complete! Your model is now on the Hugging Face Hub.")
+    print(
+        f"You can load it elsewhere using: AutoModel.from_pretrained('{REPO_ID}', trust_remote_code=True)"
+    )
+
+
+if __name__ == "__main__":
+    # Run the upload process
+    upload_model_to_hub()

test.py

@@ -0,0 +1,17 @@
+from toy_duration_predictor.preprocess import mssv
+import toy_duration_predictor.train_fastai as train_fastai
+
+
+def preprocessing():
+    mssv_path = "/mnt/d/dataset/004.다화자 가창 데이터"
+    mssv_preprocessed_path = "/mnt/d/dataset/mssv_preprocessed_duration"
+    mssv.preprocess_dataset(mssv_path, mssv_preprocessed_path)
+
+
+def test_train():
+    train_fastai.test_train()
+
+
+if __name__ == "__main__":
+    # preprocessing()
+    test_train()

test_wandb.py

@@ -0,0 +1,31 @@
+import random
+
+import wandb
+
+# Start a new wandb run to track this script.
+run = wandb.init(
+    # Set the wandb entity where your project will be logged (generally your team name).
+    # entity="cccsss17",
+    # Set the wandb project where this run will be logged.
+    project="my-awesome-project",
+    # Track hyperparameters and run metadata.
+    config={
+        "learning_rate": 0.02,
+        "architecture": "CNN",
+        "dataset": "CIFAR-100",
+        "epochs": 10,
+    },
+)
+
+# Simulate training.
+epochs = 10
+offset = random.random() / 5
+for epoch in range(2, epochs):
+    acc = 1 - 2**-epoch - random.random() / epoch - offset
+    loss = 2**-epoch + random.random() / epoch + offset
+
+    # Log metrics to wandb.
+    run.log({"acc": acc, "loss": loss})
+
+# Finish the run and upload any remaining data.
+run.finish()