mirror of
https://github.com/vladmandic/automatic
synced 2026-09-19 17:24:32 +02:00
Merge remote-tracking branch 'origin/dev' into refactor/remove-face-restoration
# Conflicts: # .pylintrc # .ruff.toml
This commit is contained in:
CalamitousFelicitousness
@@ -1,241 +0,0 @@
|
||||
import time
|
||||
import warnings
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
from typing import Dict, Any
|
||||
|
||||
warnings.filterwarnings("ignore", category=UserWarning)
|
||||
|
||||
warmup = 2
|
||||
repeats = 50
|
||||
dtypes = [torch.bfloat16] # , torch.float16]
|
||||
# if hasattr(torch, "float8_e4m3fn"):
|
||||
# dtypes.append(torch.float8_e4m3fn)
|
||||
|
||||
PROFILES = {
|
||||
"sdxl": {"l_q": 4096, "l_k": 4096, "h": 32, "d": 128},
|
||||
"flux.1": {"l_q": 16717, "l_k": 16717, "h": 24, "d": 128},
|
||||
"sd35": {"l_q": 16538, "l_k": 16538, "h": 24, "d": 128},
|
||||
"qwen-image": {"l_q": 16384, "l_k": 16384, "h": 24, "d": 128},
|
||||
"z-image": {"l_q": 4096, "l_k": 4096, "h": 32, "d": 120},
|
||||
"wan2.1": {"l_q": 16384, "l_k": 16384, "h": 40, "d": 128},
|
||||
}
|
||||
|
||||
def get_stats(reset: bool = False):
|
||||
torch.cuda.synchronize()
|
||||
if reset:
|
||||
with torch.no_grad():
|
||||
torch.cuda.empty_cache()
|
||||
torch.cuda.reset_peak_memory_stats()
|
||||
m = torch.cuda.max_memory_allocated()
|
||||
t = time.perf_counter()
|
||||
return m / (1024 ** 2), t
|
||||
|
||||
def print_gpu_info():
|
||||
if not torch.cuda.is_available():
|
||||
print("GPU: Not available")
|
||||
return
|
||||
|
||||
device = torch.cuda.current_device()
|
||||
props = torch.cuda.get_device_properties(device)
|
||||
total_mem = props.total_memory / (1024**3)
|
||||
free_mem, _ = torch.cuda.mem_get_info(device)
|
||||
free_mem = free_mem / (1024**3)
|
||||
major, minor = torch.cuda.get_device_capability(device)
|
||||
|
||||
print(f"gpu: {torch.cuda.get_device_name(device)}")
|
||||
print(f"vram: total={total_mem:.2f}GB free={free_mem:.2f}GB")
|
||||
print(f"cuda: capability={major}.{minor} version={torch.version.cuda}")
|
||||
print(f"torch: {torch.__version__}")
|
||||
|
||||
def benchmark_attention(
|
||||
backend: str,
|
||||
dtype: torch.dtype,
|
||||
b: int = 1,
|
||||
l_q: int = 4096,
|
||||
l_k: int = 4096,
|
||||
h: int = 32,
|
||||
d: int = 128,
|
||||
warmup: int = 10,
|
||||
repeats: int = 100
|
||||
) -> Dict[str, Any]:
|
||||
device = "cuda" if torch.cuda.is_available() else "cpu"
|
||||
|
||||
# Initialize tensors
|
||||
q = torch.randn(b, h, l_q, d, device=device, dtype=torch.float16 if dtype.is_floating_point and dtype.itemsize == 1 else dtype, requires_grad=False).to(dtype)
|
||||
k = torch.randn(b, h, l_k, d, device=device, dtype=torch.float16 if dtype.is_floating_point and dtype.itemsize == 1 else dtype, requires_grad=False).to(dtype)
|
||||
v = torch.randn(b, h, l_k, d, device=device, dtype=torch.float16 if dtype.is_floating_point and dtype.itemsize == 1 else dtype, requires_grad=False).to(dtype)
|
||||
|
||||
results = {
|
||||
"backend": backend,
|
||||
"dtype": str(dtype),
|
||||
"status": "pass",
|
||||
"latency_ms": 0.0,
|
||||
"memory_mb": 0.0,
|
||||
"version": "N/A",
|
||||
"error": ""
|
||||
}
|
||||
try:
|
||||
if backend.startswith("sdpa_"):
|
||||
from torch.nn.attention import sdpa_kernel, SDPBackend
|
||||
sdp_type = backend[len("sdpa_"):]
|
||||
# Map friendly names to new SDPA backends
|
||||
backend_map = {
|
||||
"math": [SDPBackend.MATH],
|
||||
"flash": [SDPBackend.FLASH_ATTENTION],
|
||||
"mem_efficient": [SDPBackend.EFFICIENT_ATTENTION],
|
||||
"all": [SDPBackend.FLASH_ATTENTION, SDPBackend.EFFICIENT_ATTENTION, SDPBackend.MATH]
|
||||
}
|
||||
if sdp_type not in backend_map:
|
||||
raise ValueError(f"Unknown SDPA type: {sdp_type}")
|
||||
|
||||
results["version"] = torch.__version__
|
||||
|
||||
with sdpa_kernel(backend_map[sdp_type]):
|
||||
# Warmup
|
||||
for _ in range(warmup):
|
||||
_ = F.scaled_dot_product_attention(q, k, v)
|
||||
|
||||
start_mem, start_time = get_stats(True)
|
||||
|
||||
for _ in range(repeats):
|
||||
_ = F.scaled_dot_product_attention(q, k, v)
|
||||
|
||||
end_mem, end_time = get_stats()
|
||||
|
||||
results["latency_ms"] = (end_time - start_time) / repeats * 1000
|
||||
results["memory_mb"] = end_mem - start_mem
|
||||
|
||||
elif backend == "flash_attn":
|
||||
from flash_attn import flash_attn_func, __version__ as fa_version
|
||||
results["version"] = fa_version
|
||||
# Flash attention usually expects (B, L, H, D)
|
||||
q_fa = q.transpose(1, 2)
|
||||
k_fa = k.transpose(1, 2)
|
||||
v_fa = v.transpose(1, 2)
|
||||
|
||||
for _ in range(warmup):
|
||||
_ = flash_attn_func(q_fa, k_fa, v_fa)
|
||||
|
||||
start_mem, start_time = get_stats(True)
|
||||
|
||||
for _ in range(repeats):
|
||||
_ = flash_attn_func(q_fa, k_fa, v_fa)
|
||||
|
||||
end_mem, end_time = get_stats()
|
||||
|
||||
results["latency_ms"] = (end_time - start_time) / repeats * 1000
|
||||
results["memory_mb"] = end_mem - start_mem
|
||||
|
||||
elif backend == "xformers":
|
||||
from xformers.ops import memory_efficient_attention
|
||||
from xformers import __version__ as xf_version
|
||||
results["version"] = xf_version
|
||||
# xformers also usually prefers (B, L, H, D)
|
||||
q_xf = q.transpose(1, 2)
|
||||
k_xf = k.transpose(1, 2)
|
||||
v_xf = v.transpose(1, 2)
|
||||
|
||||
for _ in range(warmup):
|
||||
_ = memory_efficient_attention(q_xf, k_xf, v_xf)
|
||||
|
||||
start_mem, start_time = get_stats(True)
|
||||
|
||||
for _ in range(repeats):
|
||||
_ = memory_efficient_attention(q_xf, k_xf, v_xf)
|
||||
|
||||
end_mem, end_time = get_stats()
|
||||
|
||||
results["latency_ms"] = (end_time - start_time) / repeats * 1000
|
||||
results["memory_mb"] = end_mem - start_mem
|
||||
|
||||
elif backend == "sage_attn":
|
||||
from sageattention import sageattn
|
||||
import sageattention
|
||||
# Attempt to get version from package metadata or a common attribute
|
||||
try:
|
||||
import importlib.metadata
|
||||
results["version"] = importlib.metadata.version("sageattention")
|
||||
except Exception:
|
||||
results["version"] = getattr(sageattention, "__version__", "N/A")
|
||||
|
||||
# SageAttention expects (B, H, L, D) logic
|
||||
for _ in range(warmup):
|
||||
_ = sageattn(q, k, v)
|
||||
|
||||
start_mem, start_time = get_stats(True)
|
||||
|
||||
for _ in range(repeats):
|
||||
_ = sageattn(q, k, v)
|
||||
|
||||
end_mem, end_time = get_stats()
|
||||
|
||||
results["latency_ms"] = (end_time - start_time) / repeats * 1000
|
||||
results["memory_mb"] = end_mem - start_mem
|
||||
|
||||
elif backend == "flex_attention":
|
||||
from torch.nn.attention.flex_attention import flex_attention
|
||||
results["version"] = torch.__version__
|
||||
|
||||
# flex_attention requires torch.compile for performance
|
||||
flex_attention_compiled = torch.compile(flex_attention, dynamic=False)
|
||||
|
||||
# Warmup (important to trigger compilation)
|
||||
for _ in range(warmup):
|
||||
_ = flex_attention_compiled(q, k, v)
|
||||
|
||||
start_mem, start_time = get_stats(True)
|
||||
|
||||
for _ in range(repeats):
|
||||
_ = flex_attention_compiled(q, k, v)
|
||||
|
||||
end_mem, end_time = get_stats()
|
||||
|
||||
results["latency_ms"] = (end_time - start_time) / repeats * 1000
|
||||
results["memory_mb"] = end_mem - start_mem
|
||||
except Exception as e:
|
||||
results["status"] = "fail"
|
||||
results["error"] = str(e)[:49]
|
||||
|
||||
return results
|
||||
|
||||
def main():
|
||||
backends = [
|
||||
"sdpa_math",
|
||||
"sdpa_mem_efficient",
|
||||
"sdpa_flash",
|
||||
"flex_attention",
|
||||
"xformers",
|
||||
"flash_attn",
|
||||
"sage_attn",
|
||||
]
|
||||
|
||||
all_results = []
|
||||
|
||||
print_gpu_info()
|
||||
print(f'config: warmup={warmup} repeats={repeats} dtypes={dtypes}')
|
||||
for name, config in PROFILES.items():
|
||||
print(f"profile: {name} (L_q={config['l_q']}, L_k={config['l_k']}, H={config['h']}, D={config['d']})")
|
||||
for dtype in dtypes:
|
||||
print(f" dtype: {dtype}")
|
||||
print(f" {'backend':<20} | {'version':<12} | {'status':<8} | {'latency':<10} | {'memory':<12} | ")
|
||||
for backend in backends:
|
||||
res = benchmark_attention(
|
||||
backend,
|
||||
dtype,
|
||||
l_q=config["l_q"],
|
||||
l_k=config["l_k"],
|
||||
h=config["h"],
|
||||
d=config["d"],
|
||||
warmup=warmup,
|
||||
repeats=repeats
|
||||
)
|
||||
all_results.append(res)
|
||||
|
||||
latency = f"{res['latency_ms']:.4f} ms"
|
||||
memory = f"{res['memory_mb']:.2f} MB"
|
||||
|
||||
print(f" {res['backend']:<20} | {res['version']:<12} | {res['status']:<8} | {latency:<10} | {memory:<12} | {res['error']}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,32 +0,0 @@
|
||||
#!/usr/bin/env bash
|
||||
|
||||
node cli/api-txt2img.js
|
||||
node cli/api-pulid.js
|
||||
|
||||
source venv/bin/activate
|
||||
echo image-exif
|
||||
python cli/api-info.py --input html/logo-bg-0.jpg
|
||||
echo txt2img
|
||||
python cli/api-txt2img.py --detailer --prompt "girl on a mountain" --seed 42 --sampler DEIS --width 1280 --height 800 --steps 10
|
||||
echo img2img
|
||||
python cli/api-img2img.py --init html/logo-bg-0.jpg --steps 10
|
||||
echo inpaint
|
||||
python cli/api-img2img.py --init html/logo-bg-0.jpg --mask html/logo-dark.png --steps 10
|
||||
echo upscale
|
||||
python cli/api-upscale.py --input html/logo-bg-0.jpg --upscaler "ESRGAN 4x Valar" --scale 4
|
||||
echo vqa
|
||||
python cli/api-vqa.py --input html/logo-bg-0.jpg
|
||||
echo detailer
|
||||
python cli/api-detect.py --image html/invoked.jpg
|
||||
echo faceid
|
||||
python cli/api-faceid.py --face html/simple-dark.jpg
|
||||
echo control-txt2img
|
||||
python cli/api-control.py --prompt "cute robot"
|
||||
echo control-img2img
|
||||
python cli/api-control.py --prompt "cute robot" --input html/logo-bg-0.jpg
|
||||
echo control-ipsadapter
|
||||
python cli/api-control.py --prompt "cute robot" --ipadapter "Base SDXL:html/logo-bg-0.jpg:0.8"
|
||||
echo control-preprocess
|
||||
python cli/api-preprocess.py --input html/logo-bg-0.jpg --model "Zoe Depth"
|
||||
echo control-controlnet
|
||||
python cli/api-control.py --prompt "cute robot" --input html/logo-bg-0.jpg --type controlnet --control "Zoe Depth:Xinsir Union XL:0.5"
|
||||
@@ -1,29 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
# Remove the entries that no longer exist in locale from override.
|
||||
|
||||
import sys
|
||||
import json
|
||||
from rich import print # pylint: disable=redefined-builtin
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.argv.pop(0)
|
||||
if len(sys.argv) == 0:
|
||||
print('Invalid parameters.')
|
||||
sys.exit(1)
|
||||
filename = sys.argv[0]
|
||||
labels = []
|
||||
override = None
|
||||
try:
|
||||
with open('html/locale_en.json', 'r', encoding="utf-8") as f:
|
||||
locale = json.load(f)
|
||||
for v in locale.values():
|
||||
for item in v:
|
||||
labels.append(item['label'])
|
||||
with open(filename, 'r', encoding="utf-8") as f:
|
||||
override = json.load(f)
|
||||
except Exception:
|
||||
print('Invalid file format.')
|
||||
sys.exit(1)
|
||||
with open(filename, 'w', encoding="utf-8") as f:
|
||||
json.dump([item for item in override if item['label'] in labels], f, ensure_ascii=False)
|
||||
@@ -1,67 +0,0 @@
|
||||
#!/usr/bin/env node
|
||||
// script used to localize sdnext ui and hints to multiple languages using google gemini ai
|
||||
|
||||
const fs = require('node:fs');
|
||||
|
||||
const { GoogleGenerativeAI } = require('@google/generative-ai');
|
||||
|
||||
const api_key = process.env.GOOGLE_AI_API_KEY;
|
||||
const model = 'gemini-2.5-flash';
|
||||
const prompt = `Translate attached JSON from English to {language} using following rules: fields id, label and reload should be preserved from original, field localized should be a translated version of field label and field hint should be translated in-place.
|
||||
if field is less than 3 characters, do not translate it and keep it as is.
|
||||
Every JSON entry should have id, label, localized, reload and hint fields.
|
||||
Output should be pure JSON without any additional text. To better match translation, context of the text is related to Stable Diffusion and topic of Generative AI.`;
|
||||
const languages = {
|
||||
hr: 'Croatian',
|
||||
de: 'German',
|
||||
es: 'Spanish',
|
||||
fr: 'French',
|
||||
it: 'Italian',
|
||||
pt: 'Portuguese',
|
||||
zh: 'Chinese',
|
||||
ja: 'Japanese',
|
||||
ko: 'Korean',
|
||||
ru: 'Russian',
|
||||
};
|
||||
const chunkLines = 100;
|
||||
|
||||
async function localize() {
|
||||
if (!api_key || api_key.length < 10) {
|
||||
console.error('localize: set GOOGLE_AI_API_KEY env variable with your API key');
|
||||
process.exit();
|
||||
}
|
||||
const genAI = new GoogleGenerativeAI(api_key);
|
||||
const instance = genAI.getGenerativeModel({ model });
|
||||
const raw = fs.readFileSync('html/locale_en.json');
|
||||
const json = JSON.parse(raw);
|
||||
for (const locale of Object.keys(languages)) {
|
||||
const lang = languages[locale];
|
||||
const target = prompt.replace('{language}', lang).trim();
|
||||
const output = {};
|
||||
const fn = `html/locale_${locale}.json`;
|
||||
for (const section of Object.keys(json)) {
|
||||
const data = json[section];
|
||||
output[section] = [];
|
||||
for (let i = 0; i < data.length; i += chunkLines) {
|
||||
let markdown;
|
||||
try {
|
||||
const chunk = data.slice(i, i + chunkLines);
|
||||
const result = await instance.generateContent([target, JSON.stringify(chunk)]);
|
||||
markdown = result.response.text();
|
||||
const text = markdown.replaceAll('```', '').replace(/^.*\n/, '');
|
||||
const parsed = JSON.parse(text);
|
||||
output[section].push(...parsed);
|
||||
console.log(`localize: locale=${locale} lang=${lang} section=${section} chunk=${chunk.length} output=${output[section].length} fn=${fn}`);
|
||||
} catch (err) {
|
||||
console.error('localize:', err);
|
||||
console.error('localize input:', { target, section, i });
|
||||
console.error('localize output:', { markdown });
|
||||
}
|
||||
}
|
||||
const txt = JSON.stringify(output, null, 2);
|
||||
fs.writeFileSync(fn, txt);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
localize();
|
||||
@@ -1,260 +0,0 @@
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import numpy as np
|
||||
import torch
|
||||
|
||||
# Ensure we can import modules
|
||||
sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../")))
|
||||
|
||||
from modules.errors import log
|
||||
from modules.res4lyf import (
|
||||
BASE, SIMPLE, VARIANTS,
|
||||
RESUnifiedScheduler, RESMultistepScheduler, RESDEISMultistepScheduler,
|
||||
ETDRKScheduler, LawsonScheduler, ABNorsettScheduler, PECScheduler,
|
||||
RiemannianFlowScheduler, RESSinglestepScheduler, RESSinglestepSDEScheduler,
|
||||
RESMultistepSDEScheduler, SimpleExponentialScheduler, LinearRKScheduler,
|
||||
LobattoScheduler, GaussLegendreScheduler, RungeKutta44Scheduler,
|
||||
RungeKutta57Scheduler, RungeKutta67Scheduler, SpecializedRKScheduler,
|
||||
BongTangentScheduler, CommonSigmaScheduler, RadauIIAScheduler,
|
||||
LangevinDynamicsScheduler
|
||||
)
|
||||
from modules.schedulers.scheduler_vdm import VDMScheduler
|
||||
from modules.schedulers.scheduler_unipc_flowmatch import FlowUniPCMultistepScheduler
|
||||
from modules.schedulers.scheduler_ufogen import UFOGenScheduler
|
||||
from modules.schedulers.scheduler_tdd import TDDScheduler
|
||||
from modules.schedulers.scheduler_tcd import TCDScheduler
|
||||
from modules.schedulers.scheduler_flashflow import FlashFlowMatchEulerDiscreteScheduler
|
||||
from modules.schedulers.scheduler_dpm_flowmatch import FlowMatchDPMSolverMultistepScheduler
|
||||
from modules.schedulers.scheduler_dc import DCSolverMultistepScheduler
|
||||
from modules.schedulers.scheduler_bdia import BDIA_DDIMScheduler
|
||||
|
||||
def test_scheduler(name, scheduler_class, config):
|
||||
try:
|
||||
scheduler = scheduler_class(**config)
|
||||
except Exception as e:
|
||||
log.error(f'scheduler="{name}" cls={scheduler_class} config={config} error="Init failed: {e}"')
|
||||
return False
|
||||
|
||||
num_steps = 20
|
||||
scheduler.set_timesteps(num_steps)
|
||||
|
||||
sample = torch.randn((1, 4, 64, 64))
|
||||
has_changed = False
|
||||
t0 = time.time()
|
||||
messages = []
|
||||
|
||||
try:
|
||||
for i, t in enumerate(scheduler.timesteps):
|
||||
# Simulate model output (noise or x0 or v), Using random noise for stability check
|
||||
model_output = torch.randn_like(sample)
|
||||
|
||||
# Scaling Check
|
||||
step_idx = scheduler.step_index if hasattr(scheduler, "step_index") and scheduler.step_index is not None else i
|
||||
# Clamp index
|
||||
if hasattr(scheduler, 'sigmas'):
|
||||
step_idx = min(step_idx, len(scheduler.sigmas) - 1)
|
||||
sigma = scheduler.sigmas[step_idx]
|
||||
else:
|
||||
sigma = torch.tensor(1.0) # Dummy for non-sigma schedulers
|
||||
|
||||
# Re-introduce scaling calculation first
|
||||
scaled_sample = scheduler.scale_model_input(sample, t)
|
||||
|
||||
if config.get("prediction_type") == "flow_prediction" or name in ["UFOGenScheduler", "TDDScheduler", "TCDScheduler", "BDIA_DDIMScheduler", "DCSolverMultistepScheduler"]:
|
||||
# Some new schedulers don't use K-diffusion scaling
|
||||
expected_scale = 1.0
|
||||
else:
|
||||
expected_scale = 1.0 / ((sigma**2 + 1) ** 0.5)
|
||||
|
||||
# Simple check with loose tolerance due to float precision
|
||||
expected_scaled_sample = sample * expected_scale
|
||||
if not torch.allclose(scaled_sample, expected_scaled_sample, atol=1e-4):
|
||||
# If failed, double check if it's just 'sample' (no scaling)
|
||||
if torch.allclose(scaled_sample, sample, atol=1e-4):
|
||||
messages.append('warning="scaling is identity"')
|
||||
else:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} expected={expected_scale} error="scaling mismatch"')
|
||||
return False
|
||||
|
||||
if torch.isnan(scaled_sample).any():
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="NaN in scaled_sample"')
|
||||
return False
|
||||
|
||||
if torch.isinf(scaled_sample).any():
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="Inf in scaled_sample"')
|
||||
return False
|
||||
|
||||
output = scheduler.step(model_output, t, sample)
|
||||
|
||||
# Shape and Dtype check
|
||||
if output.prev_sample.shape != sample.shape:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="Shape mismatch: {output.prev_sample.shape} vs {sample.shape}"')
|
||||
return False
|
||||
if output.prev_sample.dtype != sample.dtype:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="Dtype mismatch: {output.prev_sample.dtype} vs {sample.dtype}"')
|
||||
return False
|
||||
|
||||
# Update check: Did the sample change?
|
||||
if not torch.equal(sample, output.prev_sample):
|
||||
has_changed = True
|
||||
|
||||
# Sample Evolution Check
|
||||
step_diff = (sample - output.prev_sample).abs().mean().item()
|
||||
if step_diff < 1e-6:
|
||||
messages.append(f'warning="minimal sample change: {step_diff}"')
|
||||
|
||||
sample = output.prev_sample
|
||||
|
||||
if torch.isnan(sample).any():
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="NaN in sample"')
|
||||
return False
|
||||
|
||||
if torch.isinf(sample).any():
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="Inf in sample"')
|
||||
return False
|
||||
|
||||
# Divergence check
|
||||
if sample.abs().max() > 1e10:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} step={i} error="divergence detected"')
|
||||
return False
|
||||
|
||||
# External check for Sigma Monotonicity
|
||||
if hasattr(scheduler, 'sigmas'):
|
||||
sigmas = scheduler.sigmas.cpu().numpy()
|
||||
if len(sigmas) > 1:
|
||||
diffs = np.diff(sigmas) # Check if potentially monotonic decreasing (standard) OR increasing (some flow/inverse setups). We allow flat sections (diff=0) hence 1e-6 slack
|
||||
is_monotonic_decreasing = np.all(diffs <= 1e-6)
|
||||
is_monotonic_increasing = np.all(diffs >= -1e-6)
|
||||
if not (is_monotonic_decreasing or is_monotonic_increasing):
|
||||
messages.append('warning="sigmas are not monotonic"')
|
||||
|
||||
except Exception as e:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} exception: {e}')
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
||||
if not has_changed:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} error="sample never changed"')
|
||||
return False
|
||||
|
||||
final_std = sample.std().item()
|
||||
if final_std > 50.0 or final_std < 0.1:
|
||||
log.error(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} std={final_std} error="variance drift"')
|
||||
|
||||
t1 = time.time()
|
||||
messages = list(set(messages))
|
||||
log.info(f'scheduler="{name}" cls={scheduler.__class__.__name__} config={config} time={t1-t0} messages={messages}')
|
||||
return True
|
||||
|
||||
def run_tests():
|
||||
prediction_types = ["epsilon", "v_prediction", "sample"] # flow_prediction is special, usually requires flow sigmas or specific setup, checking standard ones first
|
||||
|
||||
# Test BASE schedulers with their specific parameters
|
||||
log.warning('type="base"')
|
||||
for name, cls in BASE:
|
||||
configs = []
|
||||
|
||||
# prediction_types
|
||||
for pt in prediction_types:
|
||||
configs.append({"prediction_type": pt})
|
||||
|
||||
# Specific params for specific classes
|
||||
if cls == RESUnifiedScheduler:
|
||||
rk_types = ["res_2m", "res_3m", "res_2s", "res_3s", "res_5s", "res_6s", "deis_1s", "deis_2m", "deis_3m"]
|
||||
for rk in rk_types:
|
||||
for pt in prediction_types:
|
||||
configs.append({"rk_type": rk, "prediction_type": pt})
|
||||
|
||||
elif cls == RESMultistepScheduler:
|
||||
variants = ["res_2m", "res_3m", "deis_2m", "deis_3m"]
|
||||
for v in variants:
|
||||
for pt in prediction_types:
|
||||
configs.append({"variant": v, "prediction_type": pt})
|
||||
|
||||
elif cls == RESDEISMultistepScheduler:
|
||||
for order in range(1, 6):
|
||||
for pt in prediction_types:
|
||||
configs.append({"solver_order": order, "prediction_type": pt})
|
||||
|
||||
elif cls == ETDRKScheduler:
|
||||
variants = ["etdrk2_2s", "etdrk3_a_3s", "etdrk3_b_3s", "etdrk4_4s", "etdrk4_4s_alt"]
|
||||
for v in variants:
|
||||
for pt in prediction_types:
|
||||
configs.append({"variant": v, "prediction_type": pt})
|
||||
|
||||
elif cls == LawsonScheduler:
|
||||
variants = ["lawson2a_2s", "lawson2b_2s", "lawson4_4s"]
|
||||
for v in variants:
|
||||
for pt in prediction_types:
|
||||
configs.append({"variant": v, "prediction_type": pt})
|
||||
|
||||
elif cls == ABNorsettScheduler:
|
||||
variants = ["abnorsett_2m", "abnorsett_3m", "abnorsett_4m"]
|
||||
for v in variants:
|
||||
for pt in prediction_types:
|
||||
configs.append({"variant": v, "prediction_type": pt})
|
||||
|
||||
elif cls == PECScheduler:
|
||||
variants = ["pec423_2h2s", "pec433_2h3s"]
|
||||
for v in variants:
|
||||
for pt in prediction_types:
|
||||
configs.append({"variant": v, "prediction_type": pt})
|
||||
|
||||
elif cls == RiemannianFlowScheduler:
|
||||
metrics = ["euclidean", "hyperbolic", "spherical", "lorentzian"]
|
||||
for m in metrics:
|
||||
configs.append({"metric_type": m, "prediction_type": "epsilon"}) # Flow usually uses v or raw, but epsilon check matches others
|
||||
|
||||
if not configs:
|
||||
for pt in prediction_types:
|
||||
configs.append({"prediction_type": pt})
|
||||
|
||||
for conf in configs:
|
||||
test_scheduler(name, cls, conf)
|
||||
|
||||
log.warning('type="simple"')
|
||||
for name, cls in SIMPLE:
|
||||
for pt in prediction_types:
|
||||
test_scheduler(name, cls, {"prediction_type": pt})
|
||||
|
||||
log.warning('type="variants"')
|
||||
for name, cls in VARIANTS:
|
||||
# these classes preset their variants/rk_types in __init__ so we just test prediction types
|
||||
for pt in prediction_types:
|
||||
test_scheduler(name, cls, {"prediction_type": pt})
|
||||
|
||||
# Extra robustness check: Flow Prediction Type
|
||||
log.warning('type="flow"')
|
||||
flow_schedulers = [
|
||||
# res4lyf schedulers
|
||||
RESUnifiedScheduler, RESMultistepScheduler, ABNorsettScheduler,
|
||||
RESSinglestepScheduler, RESSinglestepSDEScheduler, RESDEISMultistepScheduler,
|
||||
RESMultistepSDEScheduler, ETDRKScheduler, LawsonScheduler, PECScheduler,
|
||||
SimpleExponentialScheduler, LinearRKScheduler, LobattoScheduler,
|
||||
GaussLegendreScheduler, RungeKutta44Scheduler, RungeKutta57Scheduler,
|
||||
RungeKutta67Scheduler, SpecializedRKScheduler, BongTangentScheduler,
|
||||
CommonSigmaScheduler, RadauIIAScheduler, LangevinDynamicsScheduler,
|
||||
RiemannianFlowScheduler,
|
||||
# sdnext schedulers
|
||||
FlowUniPCMultistepScheduler, FlashFlowMatchEulerDiscreteScheduler, FlowMatchDPMSolverMultistepScheduler,
|
||||
]
|
||||
for cls in flow_schedulers:
|
||||
test_scheduler(cls.__name__, cls, {"prediction_type": "flow_prediction", "use_flow_sigmas": True})
|
||||
|
||||
log.warning('type="sdnext"')
|
||||
extended_schedulers = [
|
||||
VDMScheduler,
|
||||
UFOGenScheduler,
|
||||
TDDScheduler,
|
||||
TCDScheduler,
|
||||
DCSolverMultistepScheduler,
|
||||
BDIA_DDIMScheduler
|
||||
]
|
||||
for prediction_type in ["epsilon", "v_prediction", "sample"]:
|
||||
for cls in extended_schedulers:
|
||||
test_scheduler(cls.__name__, cls, {"prediction_type": prediction_type})
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_tests()
|
||||
@@ -1,847 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
"""
|
||||
Tagger Settings Test Suite
|
||||
|
||||
Tests all WaifuDiffusion and DeepBooru tagger settings to verify they're properly
|
||||
mapped and affect output correctly.
|
||||
|
||||
Usage:
|
||||
python cli/test-tagger.py [image_path]
|
||||
|
||||
If no image path is provided, uses a built-in test image.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
|
||||
# Add parent directory to path for imports
|
||||
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
sys.path.insert(0, script_dir)
|
||||
os.chdir(script_dir)
|
||||
|
||||
# Suppress installer output during import
|
||||
os.environ['SD_INSTALL_QUIET'] = '1'
|
||||
|
||||
# Initialize cmd_args properly with all argument groups
|
||||
import modules.cmd_args
|
||||
import installer
|
||||
|
||||
# Add installer args to the parser
|
||||
installer.add_args(modules.cmd_args.parser)
|
||||
|
||||
# Parse with empty args to get defaults
|
||||
modules.cmd_args.parsed, _ = modules.cmd_args.parser.parse_known_args([])
|
||||
|
||||
# Now we can safely import modules that depend on cmd_args
|
||||
|
||||
|
||||
# Default test images (in order of preference)
|
||||
DEFAULT_TEST_IMAGES = [
|
||||
'html/sdnext-robot-2k.jpg', # SD.Next robot mascot
|
||||
'venv/lib/python3.13/site-packages/gradio/test_data/lion.jpg',
|
||||
'venv/lib/python3.13/site-packages/gradio/test_data/cheetah1.jpg',
|
||||
'venv/lib/python3.13/site-packages/skimage/data/astronaut.png',
|
||||
'venv/lib/python3.13/site-packages/skimage/data/coffee.png',
|
||||
]
|
||||
|
||||
|
||||
def find_test_image():
|
||||
"""Find a suitable test image from defaults."""
|
||||
for img_path in DEFAULT_TEST_IMAGES:
|
||||
full_path = os.path.join(script_dir, img_path)
|
||||
if os.path.exists(full_path):
|
||||
return full_path
|
||||
return None
|
||||
|
||||
|
||||
def create_test_image():
|
||||
"""Create a simple test image as fallback."""
|
||||
from PIL import Image, ImageDraw
|
||||
img = Image.new('RGB', (512, 512), color=(200, 150, 100))
|
||||
draw = ImageDraw.Draw(img)
|
||||
draw.ellipse([100, 100, 400, 400], fill=(255, 200, 150), outline=(100, 50, 0))
|
||||
draw.rectangle([150, 200, 350, 350], fill=(150, 100, 200))
|
||||
return img
|
||||
|
||||
|
||||
class TaggerTest:
|
||||
"""Test harness for tagger settings."""
|
||||
|
||||
def __init__(self):
|
||||
self.results = {'passed': [], 'failed': [], 'skipped': []}
|
||||
self.test_image = None
|
||||
self.waifudiffusion_loaded = False
|
||||
self.deepbooru_loaded = False
|
||||
|
||||
def log_pass(self, msg):
|
||||
print(f" [PASS] {msg}")
|
||||
self.results['passed'].append(msg)
|
||||
|
||||
def log_fail(self, msg):
|
||||
print(f" [FAIL] {msg}")
|
||||
self.results['failed'].append(msg)
|
||||
|
||||
def log_skip(self, msg):
|
||||
print(f" [SKIP] {msg}")
|
||||
self.results['skipped'].append(msg)
|
||||
|
||||
def log_warn(self, msg):
|
||||
print(f" [WARN] {msg}")
|
||||
self.results['skipped'].append(msg)
|
||||
|
||||
def setup(self):
|
||||
"""Load test image and models."""
|
||||
from PIL import Image
|
||||
|
||||
print("=" * 70)
|
||||
print("TAGGER SETTINGS TEST SUITE")
|
||||
print("=" * 70)
|
||||
|
||||
# Get or create test image
|
||||
if len(sys.argv) > 1 and os.path.exists(sys.argv[1]):
|
||||
img_path = sys.argv[1]
|
||||
print(f"\nUsing provided image: {img_path}")
|
||||
self.test_image = Image.open(img_path).convert('RGB')
|
||||
else:
|
||||
img_path = find_test_image()
|
||||
if img_path:
|
||||
print(f"\nUsing default test image: {img_path}")
|
||||
self.test_image = Image.open(img_path).convert('RGB')
|
||||
else:
|
||||
print("\nNo test image found, creating synthetic image...")
|
||||
self.test_image = create_test_image()
|
||||
|
||||
print(f"Image size: {self.test_image.size}")
|
||||
|
||||
# Load models
|
||||
print("\nLoading models...")
|
||||
from modules.interrogate import waifudiffusion, deepbooru
|
||||
|
||||
t0 = time.time()
|
||||
self.waifudiffusion_loaded = waifudiffusion.load_model()
|
||||
print(f" WaifuDiffusion: {'loaded' if self.waifudiffusion_loaded else 'FAILED'} ({time.time()-t0:.1f}s)")
|
||||
|
||||
t0 = time.time()
|
||||
self.deepbooru_loaded = deepbooru.load_model()
|
||||
print(f" DeepBooru: {'loaded' if self.deepbooru_loaded else 'FAILED'} ({time.time()-t0:.1f}s)")
|
||||
|
||||
def cleanup(self):
|
||||
"""Unload models and free memory."""
|
||||
print("\n" + "=" * 70)
|
||||
print("CLEANUP")
|
||||
print("=" * 70)
|
||||
|
||||
from modules.interrogate import waifudiffusion, deepbooru
|
||||
from modules import devices
|
||||
|
||||
waifudiffusion.unload_model()
|
||||
deepbooru.unload_model()
|
||||
devices.torch_gc(force=True)
|
||||
print(" Models unloaded")
|
||||
|
||||
def print_summary(self):
|
||||
"""Print test summary."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST SUMMARY")
|
||||
print("=" * 70)
|
||||
|
||||
print(f"\n PASSED: {len(self.results['passed'])}")
|
||||
for item in self.results['passed']:
|
||||
print(f" - {item}")
|
||||
|
||||
print(f"\n FAILED: {len(self.results['failed'])}")
|
||||
for item in self.results['failed']:
|
||||
print(f" - {item}")
|
||||
|
||||
print(f"\n SKIPPED: {len(self.results['skipped'])}")
|
||||
for item in self.results['skipped']:
|
||||
print(f" - {item}")
|
||||
|
||||
total = len(self.results['passed']) + len(self.results['failed'])
|
||||
if total > 0:
|
||||
success_rate = len(self.results['passed']) / total * 100
|
||||
print(f"\n SUCCESS RATE: {success_rate:.1f}% ({len(self.results['passed'])}/{total})")
|
||||
|
||||
print("\n" + "=" * 70)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: ONNX Providers Detection
|
||||
# =========================================================================
|
||||
def test_onnx_providers(self):
|
||||
"""Verify ONNX runtime providers are properly detected."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: ONNX Providers Detection")
|
||||
print("=" * 70)
|
||||
|
||||
from modules import devices
|
||||
|
||||
# Test 1: onnxruntime can be imported
|
||||
try:
|
||||
import onnxruntime as ort
|
||||
self.log_pass(f"onnxruntime imported: version={ort.__version__}")
|
||||
except ImportError as e:
|
||||
self.log_fail(f"onnxruntime import failed: {e}")
|
||||
return
|
||||
|
||||
# Test 2: Get available providers
|
||||
available = ort.get_available_providers()
|
||||
if available and len(available) > 0:
|
||||
self.log_pass(f"Available providers: {available}")
|
||||
else:
|
||||
self.log_fail("No ONNX providers available")
|
||||
return
|
||||
|
||||
# Test 3: devices.onnx is properly configured
|
||||
if devices.onnx is not None and len(devices.onnx) > 0:
|
||||
self.log_pass(f"devices.onnx configured: {devices.onnx}")
|
||||
else:
|
||||
self.log_fail(f"devices.onnx not configured: {devices.onnx}")
|
||||
|
||||
# Test 4: Configured providers exist in available providers
|
||||
for provider in devices.onnx:
|
||||
if provider in available:
|
||||
self.log_pass(f"Provider '{provider}' is available")
|
||||
else:
|
||||
self.log_fail(f"Provider '{provider}' configured but not available")
|
||||
|
||||
# Test 5: If WaifuDiffusion loaded, check session providers
|
||||
if self.waifudiffusion_loaded:
|
||||
from modules.interrogate import waifudiffusion
|
||||
if waifudiffusion.tagger.session is not None:
|
||||
session_providers = waifudiffusion.tagger.session.get_providers()
|
||||
self.log_pass(f"WaifuDiffusion session providers: {session_providers}")
|
||||
else:
|
||||
self.log_skip("WaifuDiffusion session not initialized")
|
||||
|
||||
# =========================================================================
|
||||
# TEST: Memory Management (Offload/Reload/Unload)
|
||||
# =========================================================================
|
||||
def get_memory_stats(self):
|
||||
"""Get current GPU and CPU memory usage."""
|
||||
import torch
|
||||
|
||||
stats = {}
|
||||
|
||||
# GPU memory (if CUDA available)
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.synchronize()
|
||||
stats['gpu_allocated'] = torch.cuda.memory_allocated() / 1024 / 1024 # MB
|
||||
stats['gpu_reserved'] = torch.cuda.memory_reserved() / 1024 / 1024 # MB
|
||||
else:
|
||||
stats['gpu_allocated'] = 0
|
||||
stats['gpu_reserved'] = 0
|
||||
|
||||
# CPU/RAM memory (try psutil, fallback to basic)
|
||||
try:
|
||||
import psutil
|
||||
process = psutil.Process()
|
||||
stats['ram_used'] = process.memory_info().rss / 1024 / 1024 # MB
|
||||
except ImportError:
|
||||
stats['ram_used'] = 0
|
||||
|
||||
return stats
|
||||
|
||||
def test_memory_management(self):
|
||||
"""Test model offload to RAM, reload to GPU, and unload with memory monitoring."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: Memory Management (Offload/Reload/Unload)")
|
||||
print("=" * 70)
|
||||
|
||||
import torch
|
||||
import gc
|
||||
from modules import devices
|
||||
from modules.interrogate import waifudiffusion, deepbooru
|
||||
|
||||
# Memory leak tolerance (MB) - some variance is expected
|
||||
GPU_LEAK_TOLERANCE_MB = 50
|
||||
RAM_LEAK_TOLERANCE_MB = 200
|
||||
|
||||
# =====================================================================
|
||||
# DeepBooru: Test GPU/CPU movement with memory monitoring
|
||||
# =====================================================================
|
||||
if self.deepbooru_loaded:
|
||||
print("\n DeepBooru Memory Management:")
|
||||
|
||||
# Baseline memory before any operations
|
||||
gc.collect()
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.empty_cache()
|
||||
baseline = self.get_memory_stats()
|
||||
print(f" Baseline: GPU={baseline['gpu_allocated']:.1f}MB, RAM={baseline['ram_used']:.1f}MB")
|
||||
|
||||
# Test 1: Check initial state (should be on CPU after load)
|
||||
initial_device = next(deepbooru.model.model.parameters()).device
|
||||
print(f" Initial device: {initial_device}")
|
||||
if initial_device.type == 'cpu':
|
||||
self.log_pass("DeepBooru: initial state on CPU")
|
||||
else:
|
||||
self.log_pass(f"DeepBooru: initial state on {initial_device}")
|
||||
|
||||
# Test 2: Move to GPU (start)
|
||||
deepbooru.model.start()
|
||||
gpu_device = next(deepbooru.model.model.parameters()).device
|
||||
after_gpu = self.get_memory_stats()
|
||||
print(f" After start(): {gpu_device} | GPU={after_gpu['gpu_allocated']:.1f}MB (+{after_gpu['gpu_allocated']-baseline['gpu_allocated']:.1f}MB)")
|
||||
if gpu_device.type == devices.device.type:
|
||||
self.log_pass(f"DeepBooru: moved to GPU ({gpu_device})")
|
||||
else:
|
||||
self.log_fail(f"DeepBooru: failed to move to GPU, got {gpu_device}")
|
||||
|
||||
# Test 3: Run inference while on GPU
|
||||
try:
|
||||
tags = deepbooru.model.tag_multi(self.test_image, max_tags=3)
|
||||
after_infer = self.get_memory_stats()
|
||||
print(f" After inference: GPU={after_infer['gpu_allocated']:.1f}MB")
|
||||
if tags:
|
||||
self.log_pass(f"DeepBooru: inference on GPU works ({tags[:30]}...)")
|
||||
else:
|
||||
self.log_fail("DeepBooru: inference on GPU returned empty")
|
||||
except Exception as e:
|
||||
self.log_fail(f"DeepBooru: inference on GPU failed: {e}")
|
||||
|
||||
# Test 4: Offload to CPU (stop)
|
||||
deepbooru.model.stop()
|
||||
gc.collect()
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.empty_cache()
|
||||
after_offload = self.get_memory_stats()
|
||||
cpu_device = next(deepbooru.model.model.parameters()).device
|
||||
print(f" After stop(): {cpu_device} | GPU={after_offload['gpu_allocated']:.1f}MB, RAM={after_offload['ram_used']:.1f}MB")
|
||||
if cpu_device.type == 'cpu':
|
||||
self.log_pass("DeepBooru: offloaded to CPU")
|
||||
else:
|
||||
self.log_fail(f"DeepBooru: failed to offload, still on {cpu_device}")
|
||||
|
||||
# Check GPU memory returned to near baseline after offload
|
||||
gpu_diff = after_offload['gpu_allocated'] - baseline['gpu_allocated']
|
||||
if gpu_diff <= GPU_LEAK_TOLERANCE_MB:
|
||||
self.log_pass(f"DeepBooru: GPU memory cleared after offload (diff={gpu_diff:.1f}MB)")
|
||||
else:
|
||||
self.log_fail(f"DeepBooru: GPU memory leak after offload (diff={gpu_diff:.1f}MB > {GPU_LEAK_TOLERANCE_MB}MB)")
|
||||
|
||||
# Test 5: Full cycle - reload and run again
|
||||
deepbooru.model.start()
|
||||
try:
|
||||
tags = deepbooru.model.tag_multi(self.test_image, max_tags=3)
|
||||
if tags:
|
||||
self.log_pass("DeepBooru: reload cycle works")
|
||||
else:
|
||||
self.log_fail("DeepBooru: reload cycle returned empty")
|
||||
except Exception as e:
|
||||
self.log_fail(f"DeepBooru: reload cycle failed: {e}")
|
||||
deepbooru.model.stop()
|
||||
|
||||
# Test 6: Full unload with memory check
|
||||
deepbooru.unload_model()
|
||||
gc.collect()
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.empty_cache()
|
||||
after_unload = self.get_memory_stats()
|
||||
print(f" After unload: GPU={after_unload['gpu_allocated']:.1f}MB, RAM={after_unload['ram_used']:.1f}MB")
|
||||
|
||||
if deepbooru.model.model is None:
|
||||
self.log_pass("DeepBooru: unload successful")
|
||||
else:
|
||||
self.log_fail("DeepBooru: unload failed, model still exists")
|
||||
|
||||
# Check for memory leaks after full unload
|
||||
gpu_leak = after_unload['gpu_allocated'] - baseline['gpu_allocated']
|
||||
ram_leak = after_unload['ram_used'] - baseline['ram_used']
|
||||
if gpu_leak <= GPU_LEAK_TOLERANCE_MB:
|
||||
self.log_pass(f"DeepBooru: no GPU memory leak after unload (diff={gpu_leak:.1f}MB)")
|
||||
else:
|
||||
self.log_fail(f"DeepBooru: GPU memory leak detected (diff={gpu_leak:.1f}MB > {GPU_LEAK_TOLERANCE_MB}MB)")
|
||||
|
||||
if ram_leak <= RAM_LEAK_TOLERANCE_MB:
|
||||
self.log_pass(f"DeepBooru: no RAM leak after unload (diff={ram_leak:.1f}MB)")
|
||||
else:
|
||||
self.log_warn(f"DeepBooru: RAM increased after unload (diff={ram_leak:.1f}MB) - may be caching")
|
||||
|
||||
# Reload for remaining tests
|
||||
deepbooru.load_model()
|
||||
|
||||
# =====================================================================
|
||||
# WaifuDiffusion: Test session lifecycle with memory monitoring
|
||||
# =====================================================================
|
||||
if self.waifudiffusion_loaded:
|
||||
print("\n WaifuDiffusion Memory Management:")
|
||||
|
||||
# Baseline memory
|
||||
gc.collect()
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.empty_cache()
|
||||
baseline = self.get_memory_stats()
|
||||
print(f" Baseline: GPU={baseline['gpu_allocated']:.1f}MB, RAM={baseline['ram_used']:.1f}MB")
|
||||
|
||||
# Test 1: Session exists
|
||||
if waifudiffusion.tagger.session is not None:
|
||||
self.log_pass("WaifuDiffusion: session loaded")
|
||||
else:
|
||||
self.log_fail("WaifuDiffusion: session not loaded")
|
||||
return
|
||||
|
||||
# Test 2: Get current providers
|
||||
providers = waifudiffusion.tagger.session.get_providers()
|
||||
print(f" Active providers: {providers}")
|
||||
self.log_pass(f"WaifuDiffusion: using providers {providers}")
|
||||
|
||||
# Test 3: Run inference
|
||||
try:
|
||||
tags = waifudiffusion.tagger.predict(self.test_image, max_tags=3)
|
||||
after_infer = self.get_memory_stats()
|
||||
print(f" After inference: GPU={after_infer['gpu_allocated']:.1f}MB, RAM={after_infer['ram_used']:.1f}MB")
|
||||
if tags:
|
||||
self.log_pass(f"WaifuDiffusion: inference works ({tags[:30]}...)")
|
||||
else:
|
||||
self.log_fail("WaifuDiffusion: inference returned empty")
|
||||
except Exception as e:
|
||||
self.log_fail(f"WaifuDiffusion: inference failed: {e}")
|
||||
|
||||
# Test 4: Unload session with memory check
|
||||
model_name = waifudiffusion.tagger.model_name
|
||||
waifudiffusion.unload_model()
|
||||
gc.collect()
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.empty_cache()
|
||||
after_unload = self.get_memory_stats()
|
||||
print(f" After unload: GPU={after_unload['gpu_allocated']:.1f}MB, RAM={after_unload['ram_used']:.1f}MB")
|
||||
|
||||
if waifudiffusion.tagger.session is None:
|
||||
self.log_pass("WaifuDiffusion: unload successful")
|
||||
else:
|
||||
self.log_fail("WaifuDiffusion: unload failed, session still exists")
|
||||
|
||||
# Check for memory leaks after unload
|
||||
gpu_leak = after_unload['gpu_allocated'] - baseline['gpu_allocated']
|
||||
ram_leak = after_unload['ram_used'] - baseline['ram_used']
|
||||
if gpu_leak <= GPU_LEAK_TOLERANCE_MB:
|
||||
self.log_pass(f"WaifuDiffusion: no GPU memory leak after unload (diff={gpu_leak:.1f}MB)")
|
||||
else:
|
||||
self.log_fail(f"WaifuDiffusion: GPU memory leak detected (diff={gpu_leak:.1f}MB > {GPU_LEAK_TOLERANCE_MB}MB)")
|
||||
|
||||
if ram_leak <= RAM_LEAK_TOLERANCE_MB:
|
||||
self.log_pass(f"WaifuDiffusion: no RAM leak after unload (diff={ram_leak:.1f}MB)")
|
||||
else:
|
||||
self.log_warn(f"WaifuDiffusion: RAM increased after unload (diff={ram_leak:.1f}MB) - may be caching")
|
||||
|
||||
# Test 5: Reload session
|
||||
waifudiffusion.load_model(model_name)
|
||||
after_reload = self.get_memory_stats()
|
||||
print(f" After reload: GPU={after_reload['gpu_allocated']:.1f}MB, RAM={after_reload['ram_used']:.1f}MB")
|
||||
if waifudiffusion.tagger.session is not None:
|
||||
self.log_pass("WaifuDiffusion: reload successful")
|
||||
else:
|
||||
self.log_fail("WaifuDiffusion: reload failed")
|
||||
|
||||
# Test 6: Inference after reload
|
||||
try:
|
||||
tags = waifudiffusion.tagger.predict(self.test_image, max_tags=3)
|
||||
if tags:
|
||||
self.log_pass("WaifuDiffusion: inference after reload works")
|
||||
else:
|
||||
self.log_fail("WaifuDiffusion: inference after reload returned empty")
|
||||
except Exception as e:
|
||||
self.log_fail(f"WaifuDiffusion: inference after reload failed: {e}")
|
||||
|
||||
# Final memory check after full cycle
|
||||
gc.collect()
|
||||
if torch.cuda.is_available():
|
||||
torch.cuda.empty_cache()
|
||||
final = self.get_memory_stats()
|
||||
print(f" Final (after full cycle): GPU={final['gpu_allocated']:.1f}MB, RAM={final['ram_used']:.1f}MB")
|
||||
|
||||
# =========================================================================
|
||||
# TEST: Settings Existence
|
||||
# =========================================================================
|
||||
def test_settings_exist(self):
|
||||
"""Verify all tagger settings exist in shared.opts."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: Settings Existence")
|
||||
print("=" * 70)
|
||||
|
||||
from modules import shared
|
||||
|
||||
settings = [
|
||||
('tagger_threshold', float),
|
||||
('tagger_include_rating', bool),
|
||||
('tagger_max_tags', int),
|
||||
('tagger_sort_alpha', bool),
|
||||
('tagger_use_spaces', bool),
|
||||
('tagger_escape_brackets', bool),
|
||||
('tagger_exclude_tags', str),
|
||||
('tagger_show_scores', bool),
|
||||
('waifudiffusion_model', str),
|
||||
('waifudiffusion_character_threshold', float),
|
||||
('interrogate_offload', bool),
|
||||
]
|
||||
|
||||
for setting, _expected_type in settings:
|
||||
if hasattr(shared.opts, setting):
|
||||
value = getattr(shared.opts, setting)
|
||||
self.log_pass(f"{setting} = {value!r}")
|
||||
else:
|
||||
self.log_fail(f"{setting} - NOT FOUND")
|
||||
|
||||
# =========================================================================
|
||||
# TEST: Parameter Effect - Tests a single parameter on both taggers
|
||||
# =========================================================================
|
||||
def test_parameter(self, param_name, test_func, waifudiffusion_supported=True, deepbooru_supported=True):
|
||||
"""Test a parameter on both WaifuDiffusion and DeepBooru."""
|
||||
print(f"\n Testing: {param_name}")
|
||||
|
||||
if waifudiffusion_supported and self.waifudiffusion_loaded:
|
||||
try:
|
||||
result = test_func('waifudiffusion')
|
||||
if result is True:
|
||||
self.log_pass(f"WaifuDiffusion: {param_name}")
|
||||
elif result is False:
|
||||
self.log_fail(f"WaifuDiffusion: {param_name}")
|
||||
else:
|
||||
self.log_skip(f"WaifuDiffusion: {param_name} - {result}")
|
||||
except Exception as e:
|
||||
self.log_fail(f"WaifuDiffusion: {param_name} - {e}")
|
||||
elif waifudiffusion_supported:
|
||||
self.log_skip(f"WaifuDiffusion: {param_name} - model not loaded")
|
||||
|
||||
if deepbooru_supported and self.deepbooru_loaded:
|
||||
try:
|
||||
result = test_func('deepbooru')
|
||||
if result is True:
|
||||
self.log_pass(f"DeepBooru: {param_name}")
|
||||
elif result is False:
|
||||
self.log_fail(f"DeepBooru: {param_name}")
|
||||
else:
|
||||
self.log_skip(f"DeepBooru: {param_name} - {result}")
|
||||
except Exception as e:
|
||||
self.log_fail(f"DeepBooru: {param_name} - {e}")
|
||||
elif deepbooru_supported:
|
||||
self.log_skip(f"DeepBooru: {param_name} - model not loaded")
|
||||
|
||||
def tag(self, tagger, **kwargs):
|
||||
"""Helper to call the appropriate tagger."""
|
||||
if tagger == 'waifudiffusion':
|
||||
from modules.interrogate import waifudiffusion
|
||||
return waifudiffusion.tagger.predict(self.test_image, **kwargs)
|
||||
else:
|
||||
from modules.interrogate import deepbooru
|
||||
return deepbooru.model.tag(self.test_image, **kwargs)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: general_threshold
|
||||
# =========================================================================
|
||||
def test_threshold(self):
|
||||
"""Test that threshold affects tag count."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: general_threshold effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_threshold(tagger):
|
||||
tags_high = self.tag(tagger, general_threshold=0.9)
|
||||
tags_low = self.tag(tagger, general_threshold=0.1)
|
||||
|
||||
count_high = len(tags_high.split(', ')) if tags_high else 0
|
||||
count_low = len(tags_low.split(', ')) if tags_low else 0
|
||||
|
||||
print(f" {tagger}: threshold=0.9 -> {count_high} tags, threshold=0.1 -> {count_low} tags")
|
||||
|
||||
if count_low > count_high:
|
||||
return True
|
||||
elif count_low == count_high == 0:
|
||||
return "no tags returned"
|
||||
else:
|
||||
return "threshold effect unclear"
|
||||
|
||||
self.test_parameter('general_threshold', check_threshold)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: max_tags
|
||||
# =========================================================================
|
||||
def test_max_tags(self):
|
||||
"""Test that max_tags limits output."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: max_tags effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_max_tags(tagger):
|
||||
tags_5 = self.tag(tagger, general_threshold=0.1, max_tags=5)
|
||||
tags_50 = self.tag(tagger, general_threshold=0.1, max_tags=50)
|
||||
|
||||
count_5 = len(tags_5.split(', ')) if tags_5 else 0
|
||||
count_50 = len(tags_50.split(', ')) if tags_50 else 0
|
||||
|
||||
print(f" {tagger}: max_tags=5 -> {count_5} tags, max_tags=50 -> {count_50} tags")
|
||||
|
||||
return count_5 <= 5
|
||||
|
||||
self.test_parameter('max_tags', check_max_tags)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: use_spaces
|
||||
# =========================================================================
|
||||
def test_use_spaces(self):
|
||||
"""Test that use_spaces converts underscores to spaces."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: use_spaces effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_use_spaces(tagger):
|
||||
tags_under = self.tag(tagger, use_spaces=False, max_tags=10)
|
||||
tags_space = self.tag(tagger, use_spaces=True, max_tags=10)
|
||||
|
||||
print(f" {tagger} use_spaces=False: {tags_under[:50]}...")
|
||||
print(f" {tagger} use_spaces=True: {tags_space[:50]}...")
|
||||
|
||||
# Check if underscores are converted to spaces
|
||||
has_underscore_before = '_' in tags_under
|
||||
has_underscore_after = '_' in tags_space.replace(', ', ',') # ignore comma-space
|
||||
|
||||
# If there were underscores before but not after, it worked
|
||||
if has_underscore_before and not has_underscore_after:
|
||||
return True
|
||||
# If there were never underscores, inconclusive
|
||||
elif not has_underscore_before:
|
||||
return "no underscores in tags to convert"
|
||||
else:
|
||||
return False
|
||||
|
||||
self.test_parameter('use_spaces', check_use_spaces)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: escape_brackets
|
||||
# =========================================================================
|
||||
def test_escape_brackets(self):
|
||||
"""Test that escape_brackets escapes special characters."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: escape_brackets effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_escape_brackets(tagger):
|
||||
tags_escaped = self.tag(tagger, escape_brackets=True, max_tags=30, general_threshold=0.1)
|
||||
tags_raw = self.tag(tagger, escape_brackets=False, max_tags=30, general_threshold=0.1)
|
||||
|
||||
print(f" {tagger} escape=True: {tags_escaped[:60]}...")
|
||||
print(f" {tagger} escape=False: {tags_raw[:60]}...")
|
||||
|
||||
# Check for escaped brackets (\\( or \\))
|
||||
has_escaped = '\\(' in tags_escaped or '\\)' in tags_escaped
|
||||
has_unescaped = '(' in tags_raw.replace('\\(', '') or ')' in tags_raw.replace('\\)', '')
|
||||
|
||||
if has_escaped:
|
||||
return True
|
||||
elif has_unescaped:
|
||||
# Has brackets but not escaped - fail
|
||||
return False
|
||||
else:
|
||||
return "no brackets in tags to escape"
|
||||
|
||||
self.test_parameter('escape_brackets', check_escape_brackets)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: sort_alpha
|
||||
# =========================================================================
|
||||
def test_sort_alpha(self):
|
||||
"""Test that sort_alpha sorts tags alphabetically."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: sort_alpha effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_sort_alpha(tagger):
|
||||
tags_conf = self.tag(tagger, sort_alpha=False, max_tags=20, general_threshold=0.1)
|
||||
tags_alpha = self.tag(tagger, sort_alpha=True, max_tags=20, general_threshold=0.1)
|
||||
|
||||
list_conf = [t.strip() for t in tags_conf.split(',')]
|
||||
list_alpha = [t.strip() for t in tags_alpha.split(',')]
|
||||
|
||||
print(f" {tagger} by_confidence: {', '.join(list_conf[:5])}...")
|
||||
print(f" {tagger} alphabetical: {', '.join(list_alpha[:5])}...")
|
||||
|
||||
is_sorted = list_alpha == sorted(list_alpha)
|
||||
return is_sorted
|
||||
|
||||
self.test_parameter('sort_alpha', check_sort_alpha)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: exclude_tags
|
||||
# =========================================================================
|
||||
def test_exclude_tags(self):
|
||||
"""Test that exclude_tags removes specified tags."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: exclude_tags effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_exclude_tags(tagger):
|
||||
tags_all = self.tag(tagger, max_tags=50, general_threshold=0.1, exclude_tags='')
|
||||
tag_list = [t.strip().replace(' ', '_') for t in tags_all.split(',')]
|
||||
|
||||
if len(tag_list) < 2:
|
||||
return "not enough tags to test"
|
||||
|
||||
# Exclude the first tag
|
||||
tag_to_exclude = tag_list[0]
|
||||
tags_filtered = self.tag(tagger, max_tags=50, general_threshold=0.1, exclude_tags=tag_to_exclude)
|
||||
|
||||
print(f" {tagger} without exclusion: {tags_all[:50]}...")
|
||||
print(f" {tagger} excluding '{tag_to_exclude}': {tags_filtered[:50]}...")
|
||||
|
||||
# Check if the exact tag was removed by parsing the filtered list
|
||||
filtered_list = [t.strip().replace(' ', '_') for t in tags_filtered.split(',')]
|
||||
# Also check space variant
|
||||
tag_space_variant = tag_to_exclude.replace('_', ' ')
|
||||
tag_present = tag_to_exclude in filtered_list or tag_space_variant in [t.strip() for t in tags_filtered.split(',')]
|
||||
return not tag_present
|
||||
|
||||
self.test_parameter('exclude_tags', check_exclude_tags)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: tagger_show_scores (via shared.opts)
|
||||
# =========================================================================
|
||||
def test_show_scores(self):
|
||||
"""Test that tagger_show_scores adds confidence scores."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: tagger_show_scores effect")
|
||||
print("=" * 70)
|
||||
|
||||
from modules import shared
|
||||
|
||||
def check_show_scores(tagger):
|
||||
original = shared.opts.tagger_show_scores
|
||||
|
||||
shared.opts.tagger_show_scores = False
|
||||
tags_no_scores = self.tag(tagger, max_tags=5)
|
||||
|
||||
shared.opts.tagger_show_scores = True
|
||||
tags_with_scores = self.tag(tagger, max_tags=5)
|
||||
|
||||
shared.opts.tagger_show_scores = original
|
||||
|
||||
print(f" {tagger} show_scores=False: {tags_no_scores[:50]}...")
|
||||
print(f" {tagger} show_scores=True: {tags_with_scores[:50]}...")
|
||||
|
||||
has_scores = ':' in tags_with_scores and '(' in tags_with_scores
|
||||
no_scores = ':' not in tags_no_scores
|
||||
|
||||
return has_scores and no_scores
|
||||
|
||||
self.test_parameter('tagger_show_scores', check_show_scores)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: include_rating
|
||||
# =========================================================================
|
||||
def test_include_rating(self):
|
||||
"""Test that include_rating includes/excludes rating tags."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: include_rating effect")
|
||||
print("=" * 70)
|
||||
|
||||
def check_include_rating(tagger):
|
||||
tags_no_rating = self.tag(tagger, include_rating=False, max_tags=100, general_threshold=0.01)
|
||||
tags_with_rating = self.tag(tagger, include_rating=True, max_tags=100, general_threshold=0.01)
|
||||
|
||||
print(f" {tagger} include_rating=False: {tags_no_rating[:60]}...")
|
||||
print(f" {tagger} include_rating=True: {tags_with_rating[:60]}...")
|
||||
|
||||
# Rating tags typically start with "rating:" or are like "safe", "questionable", "explicit"
|
||||
rating_keywords = ['rating:', 'safe', 'questionable', 'explicit', 'general', 'sensitive']
|
||||
|
||||
has_rating_before = any(kw in tags_no_rating.lower() for kw in rating_keywords)
|
||||
has_rating_after = any(kw in tags_with_rating.lower() for kw in rating_keywords)
|
||||
|
||||
if has_rating_after and not has_rating_before:
|
||||
return True
|
||||
elif has_rating_after and has_rating_before:
|
||||
return "rating tags appear in both (may need very low threshold)"
|
||||
elif not has_rating_after:
|
||||
return "no rating tags detected"
|
||||
else:
|
||||
return False
|
||||
|
||||
self.test_parameter('include_rating', check_include_rating)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: character_threshold (WaifuDiffusion only)
|
||||
# =========================================================================
|
||||
def test_character_threshold(self):
|
||||
"""Test that character_threshold affects character tag count (WaifuDiffusion only)."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: character_threshold effect (WaifuDiffusion only)")
|
||||
print("=" * 70)
|
||||
|
||||
def check_character_threshold(tagger):
|
||||
if tagger != 'waifudiffusion':
|
||||
return "not supported"
|
||||
|
||||
# Character threshold only affects character tags
|
||||
# We need an image with character tags to properly test this
|
||||
tags_high = self.tag(tagger, character_threshold=0.99, general_threshold=0.5)
|
||||
tags_low = self.tag(tagger, character_threshold=0.1, general_threshold=0.5)
|
||||
|
||||
print(f" {tagger} char_threshold=0.99: {tags_high[:50]}...")
|
||||
print(f" {tagger} char_threshold=0.10: {tags_low[:50]}...")
|
||||
|
||||
# If thresholds are different, the setting is at least being applied
|
||||
# Hard to verify without an image with known character tags
|
||||
return True # Setting exists and is applied (verified by code inspection)
|
||||
|
||||
self.test_parameter('character_threshold', check_character_threshold, deepbooru_supported=False)
|
||||
|
||||
# =========================================================================
|
||||
# TEST: Unified Interface
|
||||
# =========================================================================
|
||||
def test_unified_interface(self):
|
||||
"""Test that the unified tagger interface works for both backends."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST: Unified tagger.tag() interface")
|
||||
print("=" * 70)
|
||||
|
||||
from modules.interrogate import tagger
|
||||
|
||||
# Test WaifuDiffusion through unified interface
|
||||
if self.waifudiffusion_loaded:
|
||||
try:
|
||||
models = tagger.get_models()
|
||||
waifudiffusion_model = next((m for m in models if m != 'DeepBooru'), None)
|
||||
if waifudiffusion_model:
|
||||
tags = tagger.tag(self.test_image, model_name=waifudiffusion_model, max_tags=5)
|
||||
print(f" WaifuDiffusion ({waifudiffusion_model}): {tags[:50]}...")
|
||||
self.log_pass("Unified interface: WaifuDiffusion")
|
||||
except Exception as e:
|
||||
self.log_fail(f"Unified interface: WaifuDiffusion - {e}")
|
||||
|
||||
# Test DeepBooru through unified interface
|
||||
if self.deepbooru_loaded:
|
||||
try:
|
||||
tags = tagger.tag(self.test_image, model_name='DeepBooru', max_tags=5)
|
||||
print(f" DeepBooru: {tags[:50]}...")
|
||||
self.log_pass("Unified interface: DeepBooru")
|
||||
except Exception as e:
|
||||
self.log_fail(f"Unified interface: DeepBooru - {e}")
|
||||
|
||||
def run_all_tests(self):
|
||||
"""Run all tests."""
|
||||
self.setup()
|
||||
|
||||
self.test_onnx_providers()
|
||||
self.test_memory_management()
|
||||
self.test_settings_exist()
|
||||
self.test_threshold()
|
||||
self.test_max_tags()
|
||||
self.test_use_spaces()
|
||||
self.test_escape_brackets()
|
||||
self.test_sort_alpha()
|
||||
self.test_exclude_tags()
|
||||
self.test_show_scores()
|
||||
self.test_include_rating()
|
||||
self.test_character_threshold()
|
||||
self.test_unified_interface()
|
||||
|
||||
self.cleanup()
|
||||
self.print_summary()
|
||||
|
||||
return len(self.results['failed']) == 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test = TaggerTest()
|
||||
success = test.run_all_tests()
|
||||
sys.exit(0 if success else 1)
|
||||
@@ -1,112 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
import sys, os
|
||||
from collections import Counter
|
||||
|
||||
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
os.chdir(script_dir)
|
||||
|
||||
# --- test defition -------------------------------
|
||||
# library
|
||||
fn = r'./modules/styles.py'
|
||||
# tested function
|
||||
funcname = 'select_from_weighted_list'
|
||||
# random needed
|
||||
ns = {'Dict': dict, 'random': __import__('random')}
|
||||
# number of samples to test
|
||||
tries = 2000
|
||||
# allowed deviation in percentage points
|
||||
tolerance_pct = 5
|
||||
# tests
|
||||
tests = [
|
||||
# - empty
|
||||
["", { '': 100 } ],
|
||||
# - no weights
|
||||
[ "red|blonde|black", { 'black': 33, 'red': 33, 'blonde': 33 } ],
|
||||
# - full weights <= 1
|
||||
[ "red:0.1|blonde:0.9", { 'blonde': 90, 'red': 10 } ],
|
||||
# - weights > 1 to test normalization
|
||||
[ "red:1|blonde:2|black:5", { 'blonde': 25, 'red': 12.5, 'black': 62.5 } ],
|
||||
# - disabling 0 weights to force one result
|
||||
[ "red:0|blonde|black:0", { 'blonde': 100 } ],
|
||||
# - weights <= 1 with distribution of the leftover
|
||||
[ "red:0.5|blonde|black:0.3|brown", { 'red': 50, 'black': 30, 'brown': 10, 'blonde': 10 } ],
|
||||
# - weights > 1, unweightes should get default of 1
|
||||
[ "red:2|blonde|black", { 'red': 50, 'blonde': 25, 'black': 25 } ],
|
||||
# - ignore content of ()
|
||||
[ "red:0.5|(blonde:1.3)", { 'red': 50, '(blonde:1.3)': 50 } ],
|
||||
# - ignore content of []
|
||||
[ "red:0.5|[stuff:1.3]", { '[stuff:1.3]': 50, 'red': 50 } ],
|
||||
# - ignore content of <>
|
||||
[ "red:0.5|<lora:1.0>", { '<lora:1.0>': 50, 'red': 50 } ]
|
||||
]
|
||||
|
||||
# -------------------------------------------------
|
||||
|
||||
with open(fn, 'r', encoding='utf-8') as f:
|
||||
src = f.read()
|
||||
start = src.find('def ' + funcname)
|
||||
if start == -1:
|
||||
print('Function not found')
|
||||
sys.exit(1)
|
||||
# find next top-level def or class after start
|
||||
next_def = src.find('\ndef ', start+1)
|
||||
next_class = src.find('\nclass ', start+1)
|
||||
end_candidates = [i for i in (next_def, next_class) if i != -1]
|
||||
end = min(end_candidates) if end_candidates else len(src)
|
||||
func_src = src[start:end]
|
||||
|
||||
exec(func_src, ns)
|
||||
func = ns.get(funcname)
|
||||
if func is None:
|
||||
print('Failed to extract function')
|
||||
sys.exit(1)
|
||||
|
||||
print('Running' , tries, 'isolated quick tests for ' + funcname + ':\n')
|
||||
|
||||
"""Print test summary."""
|
||||
print("\n" + "=" * 70)
|
||||
print("TEST SUMMARY")
|
||||
print("=" * 70)
|
||||
|
||||
for t in tests:
|
||||
print('INPUT:', t)
|
||||
samples = [func(t[0]) for _ in range(tries)]
|
||||
c = Counter(samples)
|
||||
print("SAMPLES: ", dict(c))
|
||||
|
||||
# validation
|
||||
expected_pct = t[1]
|
||||
expected_keys = set(expected_pct.keys())
|
||||
actual_keys = set(c.keys())
|
||||
missing = expected_keys - actual_keys
|
||||
unexpected = actual_keys - expected_keys
|
||||
|
||||
if missing or unexpected:
|
||||
if missing:
|
||||
print("MISSING: ", sorted(missing))
|
||||
if unexpected:
|
||||
print("UNEXPECTED: ", sorted(unexpected))
|
||||
print("RESULT: FAILED (keys)")
|
||||
print('')
|
||||
continue
|
||||
|
||||
failures = []
|
||||
for k, pct in expected_pct.items():
|
||||
expected_count = tries * (pct / 100.0)
|
||||
actual_count = c.get(k, 0)
|
||||
actual_pct = (actual_count / tries) * 100.0
|
||||
if abs(actual_pct - pct) > tolerance_pct:
|
||||
failures.append(
|
||||
f"{k}: expected {pct:.1f}%, got {actual_pct:.1f}% "
|
||||
f"({actual_count}/{tries})"
|
||||
)
|
||||
|
||||
if failures:
|
||||
print("OUT OF RANGE: ")
|
||||
for line in failures:
|
||||
print(" - " + line)
|
||||
print("RESULT: FAILED (distribution)")
|
||||
else:
|
||||
print("RESULT: PASSED")
|
||||
print('')
|
||||
@@ -1,40 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
from rich import print # pylint: disable=redefined-builtin
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.argv.pop(0)
|
||||
fn = sys.argv[0] if len(sys.argv) > 0 else 'html/locale_en.json'
|
||||
if not os.path.isfile(fn):
|
||||
print(f'File not found: {fn}')
|
||||
sys.exit(1)
|
||||
with open(fn, 'r', encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
keys = []
|
||||
t_names = 0
|
||||
t_hints = 0
|
||||
t_localized = 0
|
||||
t_long = 0
|
||||
for k in data.keys():
|
||||
names = len(data[k])
|
||||
t_names += names
|
||||
hints = len([k for k in data[k] if k["hint"] != ""])
|
||||
t_hints += hints
|
||||
localized = len([k for k in data[k] if k["localized"] != ""])
|
||||
t_localized += localized
|
||||
missing = names - hints
|
||||
long = 0
|
||||
for v in data[k]:
|
||||
if v['label'] in keys:
|
||||
print(f' Duplicate: {k}.{v["label"]}')
|
||||
else:
|
||||
if len(v['label']) > 63:
|
||||
long += 1
|
||||
print(f' Long label: {k}.{v["label"]}')
|
||||
keys.append(v['label'])
|
||||
t_long += long
|
||||
print(f'Section: [bold magenta]{k.ljust(20)}[/bold magenta] entries={names} localized={"[bold green]" + str(localized) + "[/bold green]" if localized > 0 else "0"} long={"[bold red]" + str(long) + "[/bold red]" if long > 0 else "0"} hints={hints} missing={"[bold red]" + str(missing) + "[/bold red]" if missing > 0 else "[bold green]0[/bold green]"}')
|
||||
print(f'Totals: entries={t_names} localized={localized} long={t_long} hints={t_hints} missing={t_names - t_hints}')
|
||||
Reference in New Issue
Block a user