3d40e88e35
modified: evaluation_studio.py modified: scripts/dataset.py new file: scripts/inference_utils.py modified: scripts/predict_single.py modified: scripts_2/__pycache__/task3_identify.cpython-314.pyc modified: scripts_2/task3_identify.py
120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import pickle
|
|
from pathlib import Path
|
|
|
|
import matplotlib.pyplot as plt
|
|
import numpy as np
|
|
|
|
try:
|
|
from .config import CORE_FEATURE_NAMES, evaluation_dir, make_experiment_config
|
|
from .dataset import build_record_from_file
|
|
from .inference_utils import predict_task1_tr
|
|
except ImportError:
|
|
from config import CORE_FEATURE_NAMES, evaluation_dir, make_experiment_config
|
|
from dataset import build_record_from_file
|
|
from inference_utils import predict_task1_tr
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Predict task1 RMS from a single harmonic waveform CSV.")
|
|
parser.add_argument("--file", type=str, required=True)
|
|
parser.add_argument("--checkpoint", type=str, default=None)
|
|
return parser.parse_args()
|
|
|
|
|
|
def resolve_checkpoint_path(project_root: Path, checkpoint_arg: str | None) -> Path:
|
|
if checkpoint_arg:
|
|
return Path(checkpoint_arg).resolve()
|
|
return project_root / "checkpoints_final" / "task1_final" / "task1_final_model.pkl"
|
|
|
|
|
|
def save_prediction_figure(file_path: Path, record, pred_rms: float, output_dir: Path) -> Path:
|
|
time_middle = record.time_middle.reshape(-1)
|
|
x_middle = record.x_middle.reshape(-1)
|
|
y_middle = record.y_middle.reshape(-1)
|
|
sampling_rate = record.sampling_rate
|
|
fft_values = np.fft.rfft(x_middle - np.mean(x_middle))
|
|
freqs = np.fft.rfftfreq(x_middle.size, d=1.0 / sampling_rate)
|
|
amplitudes = np.abs(fft_values)
|
|
|
|
fig, axes = plt.subplots(4, 1, figsize=(12, 12))
|
|
fig.suptitle(f"Task1 Final Single-File Prediction | {file_path.name}")
|
|
|
|
axes[0].plot(time_middle, x_middle, color="tab:blue")
|
|
axes[0].set_title("Input Base Excitation (Middle Segment)")
|
|
axes[0].set_xlabel("Time")
|
|
axes[0].set_ylabel("Acceleration")
|
|
axes[0].grid(True, alpha=0.3)
|
|
|
|
axes[1].plot(time_middle, y_middle, color="tab:green")
|
|
axes[1].set_title("True Top Response (Middle Segment)")
|
|
axes[1].set_xlabel("Time")
|
|
axes[1].set_ylabel("Acceleration")
|
|
axes[1].grid(True, alpha=0.3)
|
|
|
|
axes[2].plot(freqs, amplitudes, color="tab:purple")
|
|
axes[2].axvline(record.frequency_hz, color="tab:red", linestyle="--", label=f"Frequency = {record.frequency_hz:.4f} Hz")
|
|
axes[2].set_xlim(0.0, 5.0)
|
|
axes[2].set_title("Input Spectrum")
|
|
axes[2].set_xlabel("Frequency (Hz)")
|
|
axes[2].set_ylabel("Amplitude")
|
|
axes[2].grid(True, alpha=0.3)
|
|
axes[2].legend()
|
|
|
|
axes[3].bar(["True RMS", "Pred RMS"], [record.y_rms, pred_rms], color=["tab:green", "tab:orange"])
|
|
error_percent = abs(pred_rms - record.y_rms) / max(abs(record.y_rms), 1e-12) * 100.0
|
|
axes[3].set_title(
|
|
f"True RMS = {record.y_rms:.6f} | Pred RMS = {pred_rms:.6f} | Error = {error_percent:.2f}%"
|
|
)
|
|
axes[3].set_ylabel("RMS")
|
|
axes[3].grid(True, axis="y", alpha=0.3)
|
|
|
|
plt.tight_layout()
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
figure_path = output_dir / f"{file_path.stem}_prediction.png"
|
|
plt.savefig(figure_path, dpi=180, bbox_inches="tight")
|
|
plt.close(fig)
|
|
return figure_path
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
config = make_experiment_config()
|
|
ckpt_path = resolve_checkpoint_path(config.data.project_root, args.checkpoint)
|
|
with ckpt_path.open("rb") as handle:
|
|
payload = pickle.load(handle)
|
|
model = payload["model"]
|
|
|
|
file_path = Path(args.file).resolve()
|
|
record = build_record_from_file(file_path, config.data)
|
|
pred_tr, tr_source = predict_task1_tr(
|
|
model,
|
|
feature_vector=record.features,
|
|
frequency_hz=float(record.frequency_hz),
|
|
normalization_eps=float(config.data.normalization_eps),
|
|
project_root=config.data.project_root,
|
|
prefer_curve=True,
|
|
)
|
|
pred_rms = pred_tr * record.x_rms
|
|
|
|
out_dir = evaluation_dir(config)
|
|
fig_path = save_prediction_figure(file_path, record, pred_rms, out_dir)
|
|
|
|
print(f"Checkpoint: {ckpt_path}")
|
|
print(f"Model: {payload['model_name']}")
|
|
print(f"Input file: {file_path}")
|
|
for feature_name, feature_value in zip(CORE_FEATURE_NAMES, record.features.tolist()):
|
|
print(f"{feature_name}: {feature_value:.6f}")
|
|
print(f"Predicted TR: {pred_tr:.6f}")
|
|
print(f"TR Source: {tr_source}")
|
|
print(f"Predicted RMS: {pred_rms:.6f}")
|
|
print(f"True RMS: {record.y_rms:.6f}")
|
|
print(f"Relative RMS Error (%): {abs(pred_rms - record.y_rms) / max(abs(record.y_rms), 1e-12) * 100.0:.4f}")
|
|
print(f"Figure saved to: {fig_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|