TE Curve Verification Pipeline Curve Payload Diagnostics

Build CVP 1.2 curve-payload diagnostics for screened candidates.

class scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.CurveDiagnosticEntry(candidate_id, candidate_family, candidate_source_label, candidate_surface, direction_label, source_file_path, speed_rpm, torque_nm, oil_temperature_deg, curve_mae_deg, curve_rmse_deg, mean_percentage_error_pct, truth_peak_to_peak_deg, predicted_peak_to_peak_deg, peak_to_peak_error_pct, residual_peak_to_peak_pct, derivative_rmse_deg_per_deg, residual_smoothness_deg_per_deg2, residual_lag1_autocorrelation, closure_mismatch_deg, mean_harmonic_amplitude_error_pct, mean_harmonic_phase_error_deg, max_harmonic_amplitude_error_pct, max_harmonic_phase_error_deg)[source]

Bases: object

One curve-payload diagnostic row.

Parameters:

  • candidate_id (str)

  • candidate_family (str)

  • candidate_source_label (str)

  • candidate_surface (str)

  • direction_label (str)

  • source_file_path (str)

  • speed_rpm (float)

  • torque_nm (float)

  • oil_temperature_deg (float)

  • curve_mae_deg (float)

  • curve_rmse_deg (float)

  • mean_percentage_error_pct (float)

  • truth_peak_to_peak_deg (float)

  • predicted_peak_to_peak_deg (float)

  • peak_to_peak_error_pct (float)

  • residual_peak_to_peak_pct (float)

  • derivative_rmse_deg_per_deg (float)

  • residual_smoothness_deg_per_deg2 (float)

  • residual_lag1_autocorrelation (float)

  • closure_mismatch_deg (float)

  • mean_harmonic_amplitude_error_pct (float)

  • mean_harmonic_phase_error_deg (float)

  • max_harmonic_amplitude_error_pct (float)

  • max_harmonic_phase_error_deg (float)

candidate_id: str
candidate_family: str
candidate_source_label: str
candidate_surface: str
direction_label: str
source_file_path: str
speed_rpm: float
torque_nm: float
oil_temperature_deg: float
curve_mae_deg: float
curve_rmse_deg: float
mean_percentage_error_pct: float
truth_peak_to_peak_deg: float
predicted_peak_to_peak_deg: float
peak_to_peak_error_pct: float
residual_peak_to_peak_pct: float
derivative_rmse_deg_per_deg: float
residual_smoothness_deg_per_deg2: float
residual_lag1_autocorrelation: float
closure_mismatch_deg: float
mean_harmonic_amplitude_error_pct: float
mean_harmonic_phase_error_deg: float
max_harmonic_amplitude_error_pct: float
max_harmonic_phase_error_deg: float
to_csv_row()[source]

Return a stable CSV row.

Return type:

dict[str, Any]

class scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.CandidateDiagnosticSummary(rank, candidate_id, candidate_family, candidate_source_label, candidate_surface, valid_direction_list, curve_count, mean_percentage_error_pct, mean_curve_mae_deg, mean_peak_to_peak_error_pct, mean_residual_peak_to_peak_pct, mean_derivative_rmse_deg_per_deg, mean_residual_smoothness_deg_per_deg2, mean_residual_lag1_autocorrelation, mean_closure_mismatch_deg, mean_stitched_boundary_mismatch_deg, mean_harmonic_amplitude_error_pct, mean_harmonic_phase_error_deg, max_harmonic_amplitude_error_pct, max_harmonic_phase_error_deg, diagnostic_score)[source]

Bases: object

Aggregate diagnostics for one candidate.

Parameters:

  • rank (int)

  • candidate_id (str)

  • candidate_family (str)

  • candidate_source_label (str)

  • candidate_surface (str)

  • valid_direction_list (tuple[str, ...])

  • curve_count (int)

  • mean_percentage_error_pct (float)

  • mean_curve_mae_deg (float)

  • mean_peak_to_peak_error_pct (float)

  • mean_residual_peak_to_peak_pct (float)

  • mean_derivative_rmse_deg_per_deg (float)

  • mean_residual_smoothness_deg_per_deg2 (float)

  • mean_residual_lag1_autocorrelation (float)

  • mean_closure_mismatch_deg (float)

  • mean_stitched_boundary_mismatch_deg (float)

  • mean_harmonic_amplitude_error_pct (float)

  • mean_harmonic_phase_error_deg (float)

  • max_harmonic_amplitude_error_pct (float)

  • max_harmonic_phase_error_deg (float)

  • diagnostic_score (float)

rank: int
candidate_id: str
candidate_family: str
candidate_source_label: str
candidate_surface: str
valid_direction_list: tuple[str, ...]
curve_count: int
mean_percentage_error_pct: float
mean_curve_mae_deg: float
mean_peak_to_peak_error_pct: float
mean_residual_peak_to_peak_pct: float
mean_derivative_rmse_deg_per_deg: float
mean_residual_smoothness_deg_per_deg2: float
mean_residual_lag1_autocorrelation: float
mean_closure_mismatch_deg: float
mean_stitched_boundary_mismatch_deg: float
mean_harmonic_amplitude_error_pct: float
mean_harmonic_phase_error_deg: float
max_harmonic_amplitude_error_pct: float
max_harmonic_phase_error_deg: float
diagnostic_score: float
to_csv_row()[source]

Return a stable CSV row.

Return type:

dict[str, Any]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.build_argument_parser()[source]

Build the command-line argument parser.

Return type:

ArgumentParser

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.parse_command_line_arguments()[source]

Parse command-line arguments.

Return type:

Namespace

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.format_float(value)[source]

Format a numeric value for stable text output.

Parameters:

value (float)

Return type:

str

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.safe_mean(value_list)[source]

Return the mean of finite values.

Parameters:

value_list (list[float])

Return type:

float

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.resolve_timestamped_output_paths(output_root, report_topic_root, report_date)[source]

Resolve timestamped output and report directories.

Parameters:

  • output_root (Path)

  • report_topic_root (Path)

  • report_date (str | None)

Return type:

tuple[str, Path, Path]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.load_training_config(config_path, output_suffix)[source]

Load and prepare the TE Curve Verification Pipeline runtime config.

Parameters:

  • config_path (Path)

  • output_suffix (str)

Return type:

dict[str, Any]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.filter_candidate_configuration_list(training_config, candidate_id_list)[source]

Resolve and filter TE Curve Verification Pipeline candidate configurations.

Parameters:

  • training_config (dict[str, Any])

  • candidate_id_list (list[str])

Return type:

list[dict[str, Any]]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_wrapped_phase_error_deg(predicted_phase_rad, truth_phase_rad)[source]

Compute absolute wrapped phase error in degrees.

Parameters:

  • predicted_phase_rad (float)

  • truth_phase_rad (float)

Return type:

float

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_harmonic_component(signal_array, angle_deg_array, harmonic_order)[source]

Estimate one harmonic coefficient using angular-position projection.

Parameters:

  • signal_array (ndarray)

  • angle_deg_array (ndarray)

  • harmonic_order (int)

Return type:

complex

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_harmonic_diagnostics(truth_curve_deg, predicted_curve_deg, angle_deg_array, harmonic_order_list)[source]

Compute selected-harmonic amplitude and phase diagnostics.

Parameters:

  • truth_curve_deg (ndarray)

  • predicted_curve_deg (ndarray)

  • angle_deg_array (ndarray)

  • harmonic_order_list (list[int])

Return type:

tuple[float, float, float, float, list[dict[str, float]]]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_curve_diagnostic_entry(candidate_entry, harmonic_order_list)[source]

Compute diagnostics for one candidate curve payload.

Parameters:

  • candidate_entry (dict[str, Any])

  • harmonic_order_list (list[int])

Return type:

tuple[CurveDiagnosticEntry, list[dict[str, float]]]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_lag1_autocorrelation(value_array)[source]

Compute lag-one autocorrelation for a residual curve.

Parameters:

value_array (ndarray)

Return type:

float

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_stitched_boundary_mismatch(entry_list)[source]

Compute a deterministic surrogate stitched-curve boundary mismatch.

Parameters:

entry_list (list[CurveDiagnosticEntry])

Return type:

float

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.compute_candidate_summary_list(curve_diagnostic_entry_list)[source]

Aggregate curve diagnostics by candidate.

Parameters:

curve_diagnostic_entry_list (list[CurveDiagnosticEntry])

Return type:

list[CandidateDiagnosticSummary]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.write_csv(csv_path, row_list)[source]

Write a CSV file with stable newline behavior.

Parameters:

  • csv_path (Path)

  • row_list (list[dict[str, Any]])

Return type:

None

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.write_payload_sample_jsonl(payload_path, candidate_entry_list, harmonic_diagnostic_map, max_payload_curves_per_candidate, payload_point_stride)[source]

Write limited downsampled curve payload samples for inspection.

Parameters:

  • payload_path (Path)

  • candidate_entry_list (list[dict[str, Any]])

  • harmonic_diagnostic_map (dict[tuple[str, str], list[dict[str, float]]])

  • max_payload_curves_per_candidate (int)

  • payload_point_stride (int)

Return type:

None

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.markdown_table(header_list, row_list)[source]

Build a Markdown table.

Parameters:

  • header_list (list[str])

  • row_list (list[list[str]])

Return type:

list[str]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.build_report_lines(run_instance_id, config_path, output_directory, candidate_summary_list, harmonic_order_list, curve_payload_count)[source]

Build the Markdown report body.

Parameters:

  • run_instance_id (str)

  • config_path (Path)

  • output_directory (Path)

  • candidate_summary_list (list[CandidateDiagnosticSummary])

  • harmonic_order_list (list[int])

  • curve_payload_count (int)

Return type:

list[str]

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.write_summary_yaml(summary_path, run_instance_id, config_path, output_directory, report_path, candidate_summary_list, harmonic_order_list)[source]

Write a machine-readable diagnostics summary.

Parameters:

  • summary_path (Path)

  • run_instance_id (str)

  • config_path (Path)

  • output_directory (Path)

  • report_path (Path)

  • candidate_summary_list (list[CandidateDiagnosticSummary])

  • harmonic_order_list (list[int])

Return type:

None

scripts.reports.analysis.build_track2_curve_payload_diagnostics_report.main()[source]

Run the CVP 1.2 diagnostics workflow.

Return type:

None