狄雲＠パーツ難民のブログ一覧

＃プログラムの書き換え
decorators.py

import math
import os
import time
from collections import defaultdict
from functools import wraps

import torch
import torch.distributed as dist
from rich import box
from rich.console import Console
from rich.console import Group
from rich.live import Live
from rich.markdown import Markdown
from rich.padding import Padding
from rich.panel import Panel
from rich.progress import BarColumn
from rich.progress import Progress
from rich.progress import SpinnerColumn
from rich.progress import TimeElapsedColumn
from rich.progress import TimeRemainingColumn
from rich.rule import Rule
from rich.table import Table
from torch.utils.tensorboard import SummaryWriter


# This is here so that the history can be pickled.
def default_list():
return []


class Mean:
"""Keeps track of the running mean, along with the latest
value.
"""

def __init__(self):
self.reset()

def __call__(self):
mean = self.total / max(self.count, 1)
return mean

def reset(self):
self.count = 0
self.total = 0

def update(self, val):
if math.isfinite(val):
self.count += 1
self.total += val


def when(condition):
"""Runs a function only when the condition is met. The condition is
a function that is run.

Parameters
----------
condition : Callable
Function to run to check whether or not to run the decorated
function.

Example
-------
Checkpoint only runs every 100 iterations, and only if the
local rank is 0.

>>> i = 0
>>> rank = 0
>>>
>>> @when(lambda: i % 100 == 0 and rank == 0)
>>> def checkpoint():
>>> print("Saving to /runs/exp1")
>>>
>>> for i in range(1000):
>>> checkpoint()

"""

def decorator(fn):
@wraps(fn)
def decorated(*args, **kwargs):
if condition():
return fn(*args, **kwargs)

return decorated

return decorator


def timer(prefix: str = "time"):
"""Adds execution time to the output dictionary of the decorated
function. The function decorated by this must output a dictionary.
The key added will follow the form "[prefix]/[name_of_function]"

Parameters
----------
prefix : str, optional
The key added will follow the form "[prefix]/[name_of_function]",
by default "time".
"""

def decorator(fn):
@wraps(fn)
def decorated(*args, **kwargs):
s = time.perf_counter()
output = fn(*args, **kwargs)
assert isinstance(output, dict)
e = time.perf_counter()
output[f"{prefix}/{fn.__name__}"] = e - s
return output

return decorated

return decorator


class Tracker:
"""
A tracker class that helps to monitor the progress of training and logging the metrics.

Attributes
----------
metrics : dict
A dictionary containing the metrics for each label.
history : dict
A dictionary containing the history of metrics for each label.
writer : SummaryWriter
A SummaryWriter object for logging the metrics.
rank : int
The rank of the current process.
step : int
The current step of the training.
tasks : dict
A dictionary containing the progress bars and tables for each label.
pbar : Progress
A progress bar object for displaying the progress.
consoles : list
A list of console objects for logging.
live : Live
A Live object for updating the display live.

Methods
-------
print(msg: str)
Prints the given message to all consoles.
update(label: str, fn_name: str)
Updates the progress bar and table for the given label.
done(label: str, title: str)
Resets the progress bar and table for the given label and prints the final result.
track(label: str, length: int, completed: int = 0, op=getattr(dist, "ReduceOp", None) and dist.ReduceOp.AVG or None, ddp_active: bool = "LOCAL_RANK" in os.environ)
A decorator for tracking the progress and metrics of a function.
log(label: str, value_type: str = "value", history: bool = True)
A decorator for logging the metrics of a function.
is_best(label: str, key: str) -> bool
Checks if the latest value of the given key in the label is the best so far.
state_dict() -> dict
Returns a dictionary containing the state of the tracker.
load_state_dict(state_dict: dict) -> Tracker
Loads the state of the tracker from the given state dictionary.
"""

def __init__(
self,
writer: SummaryWriter = None,
log_file: str = None,
rank: int = 0,
console_width: int = 100,
step: int = 0,
):
"""
Initializes the Tracker object.

Parameters
----------
writer : SummaryWriter, optional
A SummaryWriter object for logging the metrics, by default None.
log_file : str, optional
The path to the log file, by default None.
rank : int, optional
The rank of the current process, by default 0.
console_width : int, optional
The width of the console, by default 100.
step : int, optional
The current step of the training, by default 0.
"""
self.metrics = {}
self.history = {}
self.writer = writer
self.rank = rank
self.step = step

# Create progress bars etc.
self.tasks = {}
self.pbar = Progress(
SpinnerColumn(),
"[progress.description]{task.description}",
"{task.completed}/{task.total}",
BarColumn(),
TimeElapsedColumn(),
"/",
TimeRemainingColumn(),
)
self.consoles = [Console(width=console_width)]
self.live = Live(console=self.consoles[0], refresh_per_second=10)
if log_file is not None:
self.consoles.append(Console(width=console_width, file=open(log_file, "a")))

def print(self, msg):
"""
Prints the given message to all consoles.

Parameters
----------
msg : str
The message to be printed.
"""
if self.rank == 0:
for c in self.consoles:
c.log(msg)

def update(self, label, fn_name):
"""
Updates the progress bar and table for the given label.

Parameters
----------
label : str
The label of the progress bar and table to be updated.
fn_name : str
The name of the function associated with the label.
"""
if self.rank == 0:
self.pbar.advance(self.tasks[label]["pbar"])

# Create table
table = Table(title=label, expand=True, box=box.MINIMAL)
table.add_column("key", style="cyan")
table.add_column("value", style="bright_blue")
table.add_column("mean", style="bright_green")

keys = self.metrics[label]["value"].keys()
for k in keys:
value = self.metrics[label]["value"][k]
mean = self.metrics[label]["mean"][k]()
table.add_row(k, f"{value:10.6f}", f"{mean:10.6f}")

self.tasks[label]["table"] = table
tables = [t["table"] for t in self.tasks.values()]
group = Group(*tables, self.pbar)
self.live.update(
Group(
Padding("", (0, 0)),
Rule(f"[italic]{fn_name}()", style="white"),
Padding("", (0, 0)),
Panel.fit(
group, padding=(0, 5), title="[b]Progress", border_style="blue"
),
)
)

def done(self, label: str, title: str):
"""
Resets the progress bar and table for the given label and prints the final result.

Parameters
----------
label : str
The label of the progress bar and table to be reset.
title : str
The title to be displayed when printing the final result.
"""
for label in self.metrics:
for v in self.metrics[label]["mean"].values():
v.reset()

if self.rank == 0:
self.pbar.reset(self.tasks[label]["pbar"])
tables = [t["table"] for t in self.tasks.values()]
group = Group(Markdown(f"# {title}"), *tables, self.pbar)
self.print(group)

def track(
self,
label: str,
length: int,
completed: int = 0,
op=getattr(dist, "ReduceOp", None) and dist.ReduceOp.AVG or None,
ddp_active: bool = "LOCAL_RANK" in os.environ,
):
"""
A decorator for tracking the progress and metrics of a function.

Parameters
----------
label : str
The label to be associated with the progress and metrics.
length : int
The total number of iterations to be completed.
completed : int, optional
The number of iterations already completed, by default 0.
op : optional
The reduce operation to be used.
ddp_active : bool, optional
Whether the DistributedDataParallel is active, by default "LOCAL_RANK" in os.environ.
"""
self.tasks[label] = {
"pbar": self.pbar.add_task(
f"[white]Iteration ({label})", total=length, completed=completed
),
"table": Table(),
}
self.metrics[label] = {
"value": defaultdict(),
"mean": defaultdict(lambda: Mean()),
}

def decorator(fn):
@wraps(fn)
def decorated(*args, **kwargs):
output = fn(*args, **kwargs)
if not isinstance(output, dict):
self.update(label, fn.__name__)
return output
# Collect across all DDP processes
scalar_keys = []
for k, v in output.items():
if isinstance(v, (int, float)):
v = torch.tensor([v])
if not torch.is_tensor(v):
continue
if ddp_active and v.is_cuda and op is not None: # pragma: no cover
dist.all_reduce(v, op=op)
output[k] = v.detach()
if torch.numel(v) == 1:
scalar_keys.append(k)
output[k] = v.item()

# Save the outputs to tracker
for k, v in output.items():
if k not in scalar_keys:
continue
self.metrics[label]["value"][k] = v
# Update the running mean
self.metrics[label]["mean"][k].update(v)

self.update(label, fn.__name__)
return output

return decorated

return decorator

def log(self, label: str, value_type: str = "value", history: bool = True):
"""
A decorator for logging the metrics of a function.

Parameters
----------
label : str
The label to be associated with the logging.
value_type : str, optional
The type of value to be logged, by default "value".
history : bool, optional
Whether to save the history of the metrics, by default True.
"""
assert value_type in ["mean", "value"]
if history:
if label not in self.history:
self.history[label] = defaultdict(default_list)

def decorator(fn):
@wraps(fn)
def decorated(*args, **kwargs):
output = fn(*args, **kwargs)
if self.rank == 0:
nonlocal value_type, label
metrics = self.metrics[label][value_type]
for k, v in metrics.items():
v = v() if isinstance(v, Mean) else v
if self.writer is not None:
self.writer.add_scalar(f"{k}/{label}", v, self.step)
if label in self.history:
self.history[label][k].append(v)

if label in self.history:
self.history[label]["step"].append(self.step)

return output

return decorated

return decorator

def is_best(self, label, key):
"""
Checks if the latest value of the given key in the label is the best so far.

Parameters
----------
label : str
The label of the metrics to be checked.
key : str
The key of the metric to be checked.

Returns
-------
bool
True if the latest value is the best so far, otherwise False.
"""
return self.history[label][key][-1] == min(self.history[label][key])

def state_dict(self):
"""
Returns a dictionary containing the state of the tracker.

Returns
-------
dict
A dictionary containing the history and step of the tracker.
"""
return {"history": self.history, "step": self.step}

def load_state_dict(self, state_dict):
"""
Loads the state of the tracker from the given state dictionary.

Parameters
----------
state_dict : dict
A dictionary containing the history and step of the tracker.

Returns
-------
Tracker
The tracker object with the loaded state.
"""
self.history = state_dict["history"]
self.step = state_dict["step"]
return self

ここまで出来たら生成ができるはず
Irodori-TTSの実行

PowerShellから次のコマンドを実行
uv run --no-sync python gradio_app.py --server-name 0.0.0.0 --server-port 7860

あとはブラウザで、localhost:7860か127.0.0.1:7860へアクセスする。

rocm7.2.1での音声生成にやっと成功。

出来た事
　WEB-UI
　　テキストの音声生成
　　参照音声を利用したテキストの音声生成


問題点
　生成した音声に「電子透かし」を入れられない

今後の予定
　特定人物のサンプル音声からの生成
　※Rocm6.1では成功している

RTX2000シリーズ以降やCPUで実行する人には関係のない苦労です。
TRX5050あたりを買った方が安く簡単に生成できます。
いいんだよ！やってみたかったんだよ！

ここから覚書
Irodori-TTSをインストールしたいフォルダで、右クリックメニューからPowerShellを開く
以下のコマンドを順次実行
＃は何をするかのメモなので実行しなくてもよい

＃PowerShellの一時的なセキュリティ解除
Set-ExecutionPolicy -ExecutionPolicy Bypass -Scope Process

＃Irodori-TTSのインストール
git clone https://github.com/Aratako/Irodori-TTS.git

＃仮想環境のセットアップ
python3.12 -m venv .venv
.venv\Scripts\activate

＃PythonのVersionを確認
＃3.12.nnと表示されていればOK
python --version

＃ROCm環境をセットアップする
pip install --no-cache-dir `
https://repo.radeon.com/rocm/windows/rocm-rel-7.2.1/rocm_sdk_core-7.2.1-py3-none-win_amd64.whl `
https://repo.radeon.com/rocm/windows/rocm-rel-7.2.1/rocm_sdk_devel-7.2.1-py3-none-win_amd64.whl `
https://repo.radeon.com/rocm/windows/rocm-rel-7.2.1/rocm_sdk_libraries_custom-7.2.1-py3-none-win_amd64.whl `
https://repo.radeon.com/rocm/windows/rocm-rel-7.2.1/rocm-7.2.1.tar.gz

＃ROCm AMD GPU用のtorch、torchaudioをインストール
pip install --no-cache-dir `
https://repo.radeon.com/rocm/windows/rocm-rel-7.2.1/torch-2.9.1%2Brocm7.2.1-cp312-cp312-win_amd64.whl `
https://repo.radeon.com/rocm/windows/rocm-rel-7.2.1/torchaudio-2.9.1%2Brocm7.2.1-cp312-cp312-win_amd64.whl

＃ROCmホイールパッケージのインストール
python -m pip install --index-url https://repo.amd.com/rocm/whl/gfx110X-dgpu/ "rocm[libraries,devel]"

＃HIPがsystemを認識できているかの確認
＃自分のPC環境が表示されればOK
hipinfo

＃ROCｍとＨＩＰがRX7800XTを認識できているか確認
＃RX7800XT表示があればOK
python -c "import torch; print(torch.cuda.get_device_name(0) if torch.cuda.is_available() else 'None')"

＃プログラムの書き換え
watermark.py
from __future__ import annotations

import logging
from collections.abc import Iterable

import torch

logger = logging.getLogger(__name__)

IRODORI_WATERMARK_PAYLOAD = (73, 82, 68, 84, 83) # "IRDTS"


def _as_single_channel_vector(audio: torch.Tensor) -> torch.Tensor | None:
squeezed = audio.detach().float().squeeze()
if squeezed.ndim == 0 or squeezed.numel() == 0:
return None
if squeezed.ndim == 1:
return squeezed
return squeezed.reshape(-1)


def _match_original_rank(audio: torch.Tensor, *, reference: torch.Tensor) -> torch.Tensor:
if reference.ndim == 2:
return audio.reshape(1, -1)
return audio.reshape(-1)


class SilentCipherWatermarker:
def __init__(self, *, device: str, model_type: str = "44.1k") -> None:
# MIOpenエラー回避のためウォーターマーク機能を無効化
self.model = None

@staticmethod
def _load_backend(*, device: str, model_type: str):
return None

@property
def ready(self) -> bool:
return False

def encode_one(
self,
audio: torch.Tensor,
*,
sample_rate: int,
payload: Iterable[int] = IRODORI_WATERMARK_PAYLOAD,
) -> torch.Tensor:
return audio

def encode_batch(self, audios: list[torch.Tensor], *, sample_rate: int) -> list[torch.Tensor]:
return audios

＃プログラムの書き換え
gradio_app.py
#!/usr/bin/env python3
from __future__ import annotations

import argparse
from datetime import datetime
from pathlib import Path

import gradio as gr
from huggingface_hub import hf_hub_download

from irodori_tts.inference_runtime import (
RuntimeKey,
SamplingRequest,
clear_cached_runtime,
default_runtime_device,
get_cached_runtime,
list_available_runtime_devices,
list_available_runtime_precisions,
save_wav,
)
#生成秒数を30･0　から　None　へ変更
FIXED_SECONDS = None
MAX_GRADIO_CANDIDATES = 32
GRADIO_AUDIO_COLS_PER_ROW = 8


def _default_checkpoint() -> str:
candidates = sorted(
[
*Path(".").glob("**/checkpoint_*.pt"),
*Path(".").glob("**/checkpoint_*.safetensors"),
]
)
if not candidates:
return "Aratako/Irodori-TTS-500M-v2"
return str(candidates[-1])


def _default_model_device() -> str:
return default_runtime_device()


def _default_codec_device() -> str:
return default_runtime_device()


def _precision_choices_for_device(device: str) -> list[str]:
return list_available_runtime_precisions(device)


def _on_model_device_change(device: str) -> gr.Dropdown:
choices = _precision_choices_for_device(device)
return gr.Dropdown(choices=choices, value=choices[0])


def _on_codec_device_change(device: str) -> gr.Dropdown:
choices = _precision_choices_for_device(device)
return gr.Dropdown(choices=choices, value=choices[0])


def _parse_optional_float(raw: str | None, label: str) -> float | None:
if raw is None:
return None
text = str(raw).strip()
if text == "" or text.lower() == "none":
return None
try:
return float(text)
except ValueError as exc:
raise ValueError(f"{label} must be a float or blank.") from exc


def _parse_optional_int(raw: str | None, label: str) -> int | None:
if raw is None:
return None
text = str(raw).strip()
if text == "" or text.lower() == "none":
return None
try:
return int(text)
except ValueError as exc:
raise ValueError(f"{label} must be an int or blank.") from exc


def _format_timings(stage_timings: list[tuple[str, float]], total_to_decode: float) -> str:
lines = [
"[timing] ---- request ----",
*[f"[timing] {name}: {sec * 1000.0:.1f} ms" for name, sec in stage_timings],
f"[timing] total_to_decode: {total_to_decode:.3f} s",
]
return "\n".join(lines)


def _resolve_ref_wav(uploaded_audio: str | None) -> str | None:
if uploaded_audio is not None and str(uploaded_audio).strip() != "":
return str(uploaded_audio)
return None


def _resolve_checkpoint_path(raw_checkpoint: str) -> str:
checkpoint = str(raw_checkpoint).strip()
if checkpoint == "":
raise ValueError("checkpoint is required.")

suffix = Path(checkpoint).suffix.lower()
if suffix in {".pt", ".safetensors"}:
return checkpoint

resolved = hf_hub_download(repo_id=checkpoint, filename="model.safetensors")
print(f"[gradio] checkpoint: hf://{checkpoint} -> {resolved}", flush=True)
return str(resolved)


def _build_runtime_key(
checkpoint: str,
model_device: str,
model_precision: str,
codec_device: str,
codec_precision: str,
#enable_watermark: bool,
) -> RuntimeKey:
checkpoint_path = _resolve_checkpoint_path(checkpoint)
return RuntimeKey(
checkpoint=checkpoint_path,
model_device=str(model_device),
codec_repo="Aratako/Semantic-DACVAE-Japanese-32dim",
model_precision=str(model_precision),
codec_device=str(codec_device),
codec_precision=str(codec_precision),
#enable_watermark=bool(enable_watermark),
compile_model=False,
compile_dynamic=False,
)


def _load_model(
checkpoint: str,
model_device: str,
model_precision: str,
codec_device: str,
codec_precision: str,
enable_watermark: bool,
) -> str:
runtime_key = _build_runtime_key(
checkpoint=checkpoint,
model_device=model_device,
model_precision=model_precision,
codec_device=codec_device,
codec_precision=codec_precision,
#enable_watermark=enable_watermark,
)
_, reloaded = get_cached_runtime(runtime_key)
if reloaded:
status = "loaded model into memory"
else:
status = "model already loaded; reused existing runtime"
return (
f"{status}\n"
f"checkpoint: {runtime_key.checkpoint}\n"
f"model_device: {runtime_key.model_device}\n"
f"model_precision: {runtime_key.model_precision}\n"
f"codec_device: {runtime_key.codec_device}\n"
f"codec_precision: {runtime_key.codec_precision}"
)


def _run_generation(
checkpoint: str,
model_device: str,
model_precision: str,
codec_device: str,
codec_precision: str,
enable_watermark: bool,
text: str,
uploaded_audio: str | None,
num_steps: int,
num_candidates: int,
seed_raw: str,
cfg_guidance_mode: str,
cfg_scale_text: float,
cfg_scale_speaker: float,
cfg_scale_raw: str,
cfg_min_t: float,
cfg_max_t: float,
context_kv_cache: bool,
truncation_factor_raw: str,
rescale_k_raw: str,
rescale_sigma_raw: str,
speaker_kv_scale_raw: str,
speaker_kv_min_t_raw: str,
speaker_kv_max_layers_raw: str,
) -> tuple[object, ...]:
def stdout_log(msg: str) -> None:
print(msg, flush=True)

runtime_key = _build_runtime_key(
checkpoint=checkpoint,
model_device=model_device,
model_precision=model_precision,
codec_device=codec_device,
codec_precision=codec_precision,
#enable_watermark=enable_watermark,
)

if str(text).strip() == "":
raise ValueError("text is required.")
requested_candidates = int(num_candidates)
if requested_candidates <= 0:
raise ValueError("num_candidates must be >= 1.")
if requested_candidates > MAX_GRADIO_CANDIDATES:
raise ValueError(f"num_candidates must be <= {MAX_GRADIO_CANDIDATES}.")

cfg_scale = _parse_optional_float(cfg_scale_raw, "cfg_scale")
truncation_factor = _parse_optional_float(truncation_factor_raw, "truncation_factor")
rescale_k = _parse_optional_float(rescale_k_raw, "rescale_k")
rescale_sigma = _parse_optional_float(rescale_sigma_raw, "rescale_sigma")
speaker_kv_scale = _parse_optional_float(speaker_kv_scale_raw, "speaker_kv_scale")
speaker_kv_min_t = _parse_optional_float(speaker_kv_min_t_raw, "speaker_kv_min_t")
speaker_kv_max_layers = _parse_optional_int(speaker_kv_max_layers_raw, "speaker_kv_max_layers")
seed = _parse_optional_int(seed_raw, "seed")

ref_wav = _resolve_ref_wav(uploaded_audio=uploaded_audio)
no_ref = ref_wav is None
ref_normalize_db = -16.0
ref_ensure_max = True

runtime, reloaded = get_cached_runtime(runtime_key)
stdout_log(f"[gradio] runtime: {'reloaded' if reloaded else 'reused'}")
stdout_log(
(
"[gradio] request: model_device={} model_precision={} codec_device={} codec_precision={} "
"watermark={} mode={} seconds={} steps={} seed={} no_ref={} candidates={}"
).format(
model_device,
model_precision,
codec_device,
codec_precision,
enable_watermark,
cfg_guidance_mode,
FIXED_SECONDS,
num_steps,
"random" if seed is None else seed,
no_ref,
requested_candidates,
)
)

result = runtime.synthesize(
SamplingRequest(
text=str(text),
ref_wav=ref_wav,
ref_latent=None,
no_ref=bool(no_ref),
ref_normalize_db=ref_normalize_db,
ref_ensure_max=bool(ref_ensure_max),
num_candidates=requested_candidates,
decode_mode="sequential",
seconds=FIXED_SECONDS,
max_ref_seconds=30.0,
max_text_len=None,
num_steps=int(num_steps),
seed=None if seed is None else int(seed),
cfg_guidance_mode=str(cfg_guidance_mode),
cfg_scale_text=float(cfg_scale_text),
cfg_scale_speaker=float(cfg_scale_speaker),
cfg_scale=cfg_scale,
cfg_min_t=float(cfg_min_t),
cfg_max_t=float(cfg_max_t),
truncation_factor=truncation_factor,
rescale_k=rescale_k,
rescale_sigma=rescale_sigma,
context_kv_cache=bool(context_kv_cache),
speaker_kv_scale=speaker_kv_scale,
speaker_kv_min_t=speaker_kv_min_t,
speaker_kv_max_layers=speaker_kv_max_layers,
trim_tail=True,
),
log_fn=stdout_log,
)

out_dir = Path("gradio_outputs")
out_dir.mkdir(parents=True, exist_ok=True)
stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
out_paths: list[str] = []
for i, audio in enumerate(result.audios, start=1):
out_path = save_wav(
out_dir / f"sample_{stamp}_{i:03d}.wav",
audio.float(),
result.sample_rate,
)
out_paths.append(str(out_path))

runtime_msg = "runtime: reloaded" if reloaded else "runtime: reused"
detail_lines = [
runtime_msg,
f"seed_used: {result.used_seed}",
f"candidates: {len(result.audios)}",
*[f"saved[{i}]: {path}" for i, path in enumerate(out_paths, start=1)],
*result.messages,
]
detail_text = "\n".join(detail_lines)
timing_text = _format_timings(result.stage_timings, result.total_to_decode)
stdout_log(f"[gradio] saved {len(out_paths)} candidates")

audio_updates: list[object] = []
for i in range(MAX_GRADIO_CANDIDATES):
if i < len(out_paths):
audio_updates.append(gr.update(value=out_paths[i], visible=True))
else:
audio_updates.append(gr.update(value=None, visible=False))
return (*audio_updates, detail_text, timing_text)


def _clear_runtime_cache() -> str:
clear_cached_runtime()
return "cleared loaded model from memory"


def build_ui() -> gr.Blocks:
default_checkpoint = _default_checkpoint()
default_model_device = _default_model_device()
default_codec_device = _default_codec_device()
# 修正後
device_choices = ["cuda", "cpu"]
model_precision_choices = _precision_choices_for_device(default_model_device)
codec_precision_choices = _precision_choices_for_device(default_codec_device)

with gr.Blocks(title="Irodori-TTS Gradio") as demo:
gr.Markdown("# Irodori-TTS Inference (Cached Runtime)")
gr.Markdown(
"When settings are unchanged, runtime is reused and only sampling/decoding runs."
)

with gr.Row():
checkpoint = gr.Textbox(
label="Checkpoint (.pt/.safetensors or HF repo id)",
value=default_checkpoint,
scale=4,
)
# model_device = gr.Dropdown(
# label="Model Device",
# choices=device_choices,
# value=default_model_device,
# scale=1,
# )

model_device = gr.Dropdown(
label="Model Device",
choices=device_choices,
value="cuda", # 強制的にcudaを選択状態にする
scale=1,
)

model_precision = gr.Dropdown(
label="Model Precision",
choices=model_precision_choices,
value=model_precision_choices[0],
scale=1,
)
codec_device = gr.Dropdown(
label="Codec Device",
choices=device_choices,
value=default_codec_device,
scale=1,
)
codec_precision = gr.Dropdown(
label="Codec Precision",
choices=codec_precision_choices,
value=codec_precision_choices[0],
scale=1,
)
enable_watermark = gr.State(False)

with gr.Row():
load_model_btn = gr.Button("Load Model")
clear_cache_btn = gr.Button("Unload Model")
clear_cache_msg = gr.Textbox(label="Model Status", interactive=False)

text = gr.Textbox(label="Text", lines=4)
uploaded_audio = gr.Audio(
label="Reference Audio Upload (optional, blank = no-reference mode)",
type="filepath",
)

with gr.Accordion("Sampling", open=True):
with gr.Row():
num_steps = gr.Slider(label="Num Steps", minimum=1, maximum=120, value=40, step=1)
num_candidates = gr.Slider(
label="Num Candidates",
minimum=1,
maximum=MAX_GRADIO_CANDIDATES,
value=1,
step=1,
)
seed_raw = gr.Textbox(label="Seed (blank=random)", value="")

with gr.Row():
cfg_guidance_mode = gr.Dropdown(
label="CFG Guidance Mode",
choices=["independent", "joint", "alternating"],
value="independent",
)
cfg_scale_text = gr.Slider(
label="CFG Scale Text",
minimum=0.0,
maximum=10.0,
value=3.0,
step=0.1,
)
cfg_scale_speaker = gr.Slider(
label="CFG Scale Speaker",
minimum=0.0,
maximum=10.0,
value=5.0,
step=0.1,
)

with gr.Accordion("Advanced (Optional)", open=False):
cfg_scale_raw = gr.Textbox(label="CFG Scale Override (optional)", value="")
with gr.Row():
cfg_min_t = gr.Number(label="CFG Min t", value=0.5)
cfg_max_t = gr.Number(label="CFG Max t", value=1.0)
context_kv_cache = gr.Checkbox(label="Context KV Cache", value=True)
with gr.Row():
truncation_factor_raw = gr.Textbox(label="Truncation Factor (optional)", value="")
rescale_k_raw = gr.Textbox(label="Rescale k (optional)", value="")
rescale_sigma_raw = gr.Textbox(label="Rescale sigma (optional)", value="")
with gr.Row():
speaker_kv_scale_raw = gr.Textbox(label="Speaker KV Scale (optional)", value="")
speaker_kv_min_t_raw = gr.Textbox(label="Speaker KV Min t (optional)", value="0.9")
speaker_kv_max_layers_raw = gr.Textbox(
label="Speaker KV Max Layers (optional)", value=""
)

generate_btn = gr.Button("Generate", variant="primary")

out_audios: list[gr.Audio] = []
num_rows = (
MAX_GRADIO_CANDIDATES + GRADIO_AUDIO_COLS_PER_ROW - 1
) // GRADIO_AUDIO_COLS_PER_ROW
with gr.Column():
for row_idx in range(num_rows):
with gr.Row():
for col_idx in range(GRADIO_AUDIO_COLS_PER_ROW):
i = row_idx * GRADIO_AUDIO_COLS_PER_ROW + col_idx
if i >= MAX_GRADIO_CANDIDATES:
break
out_audios.append(
gr.Audio(
label=f"Generated Audio {i + 1}",
type="filepath",
interactive=False,
visible=(i == 0),
min_width=160,
)
)
out_log = gr.Textbox(label="Run Log", lines=8)
out_timing = gr.Textbox(label="Timing", lines=8)

generate_btn.click(
_run_generation,
inputs=[
checkpoint,
model_device,
model_precision,
codec_device,
codec_precision,
enable_watermark,
text,
uploaded_audio,
num_steps,
num_candidates,
seed_raw,
cfg_guidance_mode,
cfg_scale_text,
cfg_scale_speaker,
cfg_scale_raw,
cfg_min_t,
cfg_max_t,
context_kv_cache,
truncation_factor_raw,
rescale_k_raw,
rescale_sigma_raw,
speaker_kv_scale_raw,
speaker_kv_min_t_raw,
speaker_kv_max_layers_raw,
],
outputs=[*out_audios, out_log, out_timing],
)
model_device.change(
_on_model_device_change, inputs=[model_device], outputs=[model_precision]
)
codec_device.change(
_on_codec_device_change, inputs=[codec_device], outputs=[codec_precision]
)

load_model_btn.click(
_load_model,
inputs=[
checkpoint,
model_device,
model_precision,
codec_device,
codec_precision,
enable_watermark,
],
outputs=[clear_cache_msg],
)
clear_cache_btn.click(_clear_runtime_cache, outputs=[clear_cache_msg])

return demo


def main() -> None:
parser = argparse.ArgumentParser(description="Gradio app for Irodori-TTS with cached runtime.")
parser.add_argument("--server-name", default="127.0.0.1")
parser.add_argument("--server-port", type=int, default=7860)
parser.add_argument("--share", action="store_true")
parser.add_argument("--debug", action="store_true")
args = parser.parse_args()

demo = build_ui()
demo.queue(default_concurrency_limit=1)
demo.launch(
server_name=args.server_name,
server_port=args.server_port,
share=bool(args.share),
debug=bool(args.debug),
)


if __name__ == "__main__":
main()


文字数制限でこれ以上書き込めないので、続きます。