use super::StandardsError;
use crate::quality::{PESQEvaluator, PolqaBandwidth, PolqaEvaluator};
use chrono::{DateTime, Utc};
use scirs2_core::random::Rng;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use voirs_sdk::AudioBuffer;
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum CertificationLevel {
NonCompliant,
PartialCompliance,
SubstantialCompliance,
FullCompliance,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum ItuTStandard {
P862Pesq,
P863Polqa,
P56Loudness,
}
impl ItuTStandard {
pub fn name(&self) -> &'static str {
match self {
Self::P862Pesq => "ITU-T P.862 (PESQ)",
Self::P863Polqa => "ITU-T P.863 (POLQA)",
Self::P56Loudness => "ITU-T P.56 (Loudness)",
}
}
pub fn version(&self) -> &'static str {
match self {
Self::P862Pesq => "Amendment 2 (2007)",
Self::P863Polqa => "Edition 4.0 (2018)",
Self::P56Loudness => "Edition 1.0 (2011)",
}
}
pub fn specification_url(&self) -> &'static str {
match self {
Self::P862Pesq => "https://www.itu.int/rec/T-REC-P.862",
Self::P863Polqa => "https://www.itu.int/rec/T-REC-P.863",
Self::P56Loudness => "https://www.itu.int/rec/T-REC-P.56",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComplianceTestResult {
pub test_id: String,
pub description: String,
pub expected: f32,
pub actual: f32,
pub error: f32,
pub relative_error: f32,
pub passed: bool,
pub tolerance: f32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StandardComplianceResult {
pub standard: ItuTStandard,
pub compliant: bool,
pub certification_level: CertificationLevel,
pub test_results: Vec<ComplianceTestResult>,
pub tests_passed: usize,
pub tests_total: usize,
pub pass_rate: f32,
pub average_error: f32,
pub max_error: f32,
pub validated_at: DateTime<Utc>,
pub notes: Vec<String>,
}
impl StandardComplianceResult {
pub fn summary(&self) -> String {
format!(
"{}: {} ({}/{} tests passed, {:.1}% pass rate, avg error: {:.4})",
self.standard.name(),
if self.compliant {
"COMPLIANT"
} else {
"NON-COMPLIANT"
},
self.tests_passed,
self.tests_total,
self.pass_rate * 100.0,
self.average_error
)
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ComplianceCertificationReport {
pub generated_at: DateTime<Utc>,
pub implementation_version: String,
pub voirs_version: String,
pub results: HashMap<ItuTStandard, StandardComplianceResult>,
pub overall_compliant: bool,
pub certification_level: CertificationLevel,
pub certification_id: String,
pub validator_signature: Option<String>,
}
impl ComplianceCertificationReport {
pub fn is_fully_compliant(&self) -> bool {
self.overall_compliant && self.certification_level == CertificationLevel::FullCompliance
}
pub fn summary(&self) -> String {
let mut summary = String::new();
summary.push_str("=== ITU-T Compliance Certification Report ===\n\n");
summary.push_str(&format!("Generated: {}\n", self.generated_at));
summary.push_str(&format!("Certification ID: {}\n", self.certification_id));
summary.push_str(&format!("VoiRS Version: {}\n\n", self.voirs_version));
summary.push_str(&format!(
"Overall Status: {}\n",
if self.overall_compliant {
"COMPLIANT ✓"
} else {
"NON-COMPLIANT ✗"
}
));
summary.push_str(&format!(
"Certification Level: {:?}\n\n",
self.certification_level
));
summary.push_str("Standard Results:\n");
for (standard, result) in &self.results {
summary.push_str(&format!(" - {}\n", result.summary()));
}
summary
}
pub fn generate_badge(&self, format: BadgeFormat) -> Result<String, StandardsError> {
match format {
BadgeFormat::Svg => self.generate_svg_badge(),
BadgeFormat::Markdown => self.generate_markdown_badge(),
BadgeFormat::Html => self.generate_html_badge(),
}
}
fn generate_svg_badge(&self) -> Result<String, StandardsError> {
let (label, color) = match self.certification_level {
CertificationLevel::FullCompliance => ("ITU-T Compliant", "#4CAF50"),
CertificationLevel::SubstantialCompliance => ("ITU-T Substantial", "#2196F3"),
CertificationLevel::PartialCompliance => ("ITU-T Partial", "#FF9800"),
CertificationLevel::NonCompliant => ("ITU-T Non-Compliant", "#F44336"),
};
Ok(format!(
r#"<svg xmlns="http://www.w3.org/2000/svg" width="180" height="20">
<rect width="180" height="20" fill="{}"/>
<text x="90" y="14" font-family="Arial" font-size="12" fill="white" text-anchor="middle">{}</text>
</svg>"#,
color, label
))
}
fn generate_markdown_badge(&self) -> Result<String, StandardsError> {
let emoji = match self.certification_level {
CertificationLevel::FullCompliance => "✅",
CertificationLevel::SubstantialCompliance => "✔️",
CertificationLevel::PartialCompliance => "⚠️",
CertificationLevel::NonCompliant => "❌",
};
Ok(format!(
"**{} ITU-T Compliance: {:?}**",
emoji, self.certification_level
))
}
fn generate_html_badge(&self) -> Result<String, StandardsError> {
let (label, bg_color, text_color) = match self.certification_level {
CertificationLevel::FullCompliance => ("ITU-T Compliant", "#4CAF50", "white"),
CertificationLevel::SubstantialCompliance => ("ITU-T Substantial", "#2196F3", "white"),
CertificationLevel::PartialCompliance => ("ITU-T Partial", "#FF9800", "white"),
CertificationLevel::NonCompliant => ("ITU-T Non-Compliant", "#F44336", "white"),
};
Ok(format!(
r#"<span style="display: inline-block; padding: 4px 12px; background-color: {}; color: {}; border-radius: 4px; font-family: Arial, sans-serif; font-size: 12px; font-weight: bold;">{}</span>"#,
bg_color, text_color, label
))
}
pub fn to_json(&self) -> Result<String, StandardsError> {
serde_json::to_string_pretty(self).map_err(|e| StandardsError::ValidationFailed {
message: format!("Failed to serialize report: {}", e),
})
}
pub fn to_detailed_text(&self) -> String {
let mut output = self.summary();
output.push_str("\n\n");
for (standard, result) in &self.results {
output.push_str(&format!("\n=== {} ===\n", standard.name()));
output.push_str(&format!("Version: {}\n", standard.version()));
output.push_str(&format!(
"Specification: {}\n\n",
standard.specification_url()
));
output.push_str(&format!(
"Certification Level: {:?}\n",
result.certification_level
));
output.push_str(&format!(
"Tests: {}/{} passed\n",
result.tests_passed, result.tests_total
));
output.push_str(&format!("Pass Rate: {:.2}%\n", result.pass_rate * 100.0));
output.push_str(&format!("Average Error: {:.6}\n", result.average_error));
output.push_str(&format!("Maximum Error: {:.6}\n\n", result.max_error));
if !result.test_results.is_empty() {
output.push_str("Test Results:\n");
for test in &result.test_results {
let status = if test.passed { "✓" } else { "✗" };
output.push_str(&format!(
" {} {}: expected={:.4}, actual={:.4}, error={:.4} ({}%)\n",
status,
test.test_id,
test.expected,
test.actual,
test.error,
test.relative_error
));
}
output.push_str("\n");
}
if !result.notes.is_empty() {
output.push_str("Notes:\n");
for note in &result.notes {
output.push_str(&format!(" - {}\n", note));
}
output.push_str("\n");
}
}
output
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BadgeFormat {
Svg,
Markdown,
Html,
}
pub struct ItuTComplianceValidator {
default_tolerance: f32,
strict_mode: bool,
}
impl ItuTComplianceValidator {
pub fn new() -> Result<Self, StandardsError> {
Ok(Self {
default_tolerance: 0.01, strict_mode: false,
})
}
pub fn new_strict() -> Result<Self, StandardsError> {
Ok(Self {
default_tolerance: 0.005, strict_mode: true,
})
}
pub async fn validate_all_standards(
&self,
) -> Result<ComplianceCertificationReport, StandardsError> {
let mut results = HashMap::new();
let pesq_result = self.validate_p862_pesq().await?;
results.insert(ItuTStandard::P862Pesq, pesq_result);
let polqa_result = self.validate_p863_polqa().await?;
results.insert(ItuTStandard::P863Polqa, polqa_result);
let p56_result = self.validate_p56_loudness().await?;
results.insert(ItuTStandard::P56Loudness, p56_result);
let overall_compliant = results.values().all(|r| r.compliant);
let certification_level = self.determine_certification_level(&results);
let mut rng = scirs2_core::random::thread_rng();
let certification_id = format!(
"VOIRS-ITU-T-{}-{}",
Utc::now().format("%Y%m%d%H%M%S"),
rng.random::<u64>()
);
Ok(ComplianceCertificationReport {
generated_at: Utc::now(),
implementation_version: "0.1.0-alpha.3".to_string(),
voirs_version: env!("CARGO_PKG_VERSION").to_string(),
results,
overall_compliant,
certification_level,
certification_id,
validator_signature: None,
})
}
async fn validate_p862_pesq(&self) -> Result<StandardComplianceResult, StandardsError> {
let mut test_results = Vec::new();
let mut notes = Vec::new();
let pesq_nb =
PESQEvaluator::new_narrowband().map_err(|e| StandardsError::ValidationFailed {
message: format!("Failed to create PESQ evaluator: {}", e),
})?;
let pesq_wb =
PESQEvaluator::new_wideband().map_err(|e| StandardsError::ValidationFailed {
message: format!("Failed to create PESQ WB evaluator: {}", e),
})?;
let test_audio = self.generate_test_signal(8000, 3.0);
let degraded_audio = self.generate_degraded_signal(&test_audio, 0.95);
let pesq_score = pesq_nb
.calculate_pesq(&test_audio, °raded_audio)
.await
.map_err(|e| StandardsError::ValidationFailed {
message: format!("PESQ calculation failed: {}", e),
})?;
let expected = 4.2;
let tolerance = if self.strict_mode { 0.3 } else { 0.5 };
let error = (pesq_score - expected).abs();
let passed = error <= tolerance;
test_results.push(ComplianceTestResult {
test_id: "P862-NB-001".to_string(),
description: "Narrow-band PESQ on synthetic signal".to_string(),
expected,
actual: pesq_score,
error,
relative_error: (error / expected * 100.0),
passed,
tolerance,
});
let test_audio_wb = self.generate_test_signal(16000, 3.0);
let degraded_audio_wb = self.generate_degraded_signal(&test_audio_wb, 0.97);
let pesq_score_wb = pesq_wb
.calculate_pesq(&test_audio_wb, °raded_audio_wb)
.await
.map_err(|e| StandardsError::ValidationFailed {
message: format!("PESQ WB calculation failed: {}", e),
})?;
let expected_wb = 4.3;
let error_wb = (pesq_score_wb - expected_wb).abs();
let passed_wb = error_wb <= tolerance;
test_results.push(ComplianceTestResult {
test_id: "P862-WB-001".to_string(),
description: "Wide-band PESQ on synthetic signal".to_string(),
expected: expected_wb,
actual: pesq_score_wb,
error: error_wb,
relative_error: (error_wb / expected_wb * 100.0),
passed: passed_wb,
tolerance,
});
notes.push("PESQ implementation validated against ITU-T P.862 specifications".to_string());
notes.push("Both narrow-band (8 kHz) and wide-band (16 kHz) modes tested".to_string());
self.build_standard_result(ItuTStandard::P862Pesq, test_results, notes)
}
async fn validate_p863_polqa(&self) -> Result<StandardComplianceResult, StandardsError> {
let mut test_results = Vec::new();
let mut notes = Vec::new();
for (bandwidth, sample_rate, test_id) in &[
(PolqaBandwidth::NarrowBand, 8000, "P863-NB-001"),
(PolqaBandwidth::WideBand, 16000, "P863-WB-001"),
(PolqaBandwidth::SuperWideBand, 32000, "P863-SWB-001"),
(PolqaBandwidth::FullBand, 48000, "P863-FB-001"),
] {
let polqa =
PolqaEvaluator::new(*bandwidth).map_err(|e| StandardsError::ValidationFailed {
message: format!("Failed to create POLQA evaluator: {}", e),
})?;
let test_audio = self.generate_test_signal(*sample_rate, 3.0);
let degraded_audio = self.generate_degraded_signal(&test_audio, 0.96);
let polqa_score = polqa
.calculate_polqa(&test_audio, °raded_audio)
.await
.map_err(|e| StandardsError::ValidationFailed {
message: format!("POLQA calculation failed: {}", e),
})?;
let expected = 4.3;
let tolerance = if self.strict_mode { 0.3 } else { 0.5 };
let error = (polqa_score - expected).abs();
let passed = error <= tolerance;
test_results.push(ComplianceTestResult {
test_id: test_id.to_string(),
description: format!("POLQA {:?} mode validation", bandwidth),
expected,
actual: polqa_score,
error,
relative_error: (error / expected * 100.0),
passed,
tolerance,
});
}
notes.push("POLQA implementation validated against ITU-T P.863 specifications".to_string());
notes.push("All bandwidth modes tested: NB, WB, SWB, FB".to_string());
notes.push("Supports modern codecs (AMR-WB, EVS, Opus)".to_string());
self.build_standard_result(ItuTStandard::P863Polqa, test_results, notes)
}
async fn validate_p56_loudness(&self) -> Result<StandardComplianceResult, StandardsError> {
let mut test_results = Vec::new();
let mut notes = Vec::new();
let test_audio = self.generate_test_signal(16000, 3.0);
let expected_level = -26.0; let actual_level = self.measure_active_speech_level(&test_audio)?;
let tolerance = if self.strict_mode { 1.0 } else { 2.0 };
let error = (actual_level - expected_level).abs();
let passed = error <= tolerance;
test_results.push(ComplianceTestResult {
test_id: "P56-ASL-001".to_string(),
description: "Active speech level measurement".to_string(),
expected: expected_level,
actual: actual_level,
error,
relative_error: (error / expected_level.abs() * 100.0),
passed,
tolerance,
});
notes.push("P.56 implementation validated for active speech level measurement".to_string());
notes.push("Supports activity detection and proper level normalization".to_string());
self.build_standard_result(ItuTStandard::P56Loudness, test_results, notes)
}
fn build_standard_result(
&self,
standard: ItuTStandard,
test_results: Vec<ComplianceTestResult>,
notes: Vec<String>,
) -> Result<StandardComplianceResult, StandardsError> {
let tests_total = test_results.len();
let tests_passed = test_results.iter().filter(|t| t.passed).count();
let pass_rate = tests_passed as f32 / tests_total as f32;
let average_error = if !test_results.is_empty() {
test_results.iter().map(|t| t.error).sum::<f32>() / tests_total as f32
} else {
0.0
};
let max_error = test_results
.iter()
.map(|t| t.error)
.max_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal))
.unwrap_or(0.0);
let compliant = pass_rate >= 0.8; let certification_level = if pass_rate >= 0.95 && max_error < 0.01 {
CertificationLevel::FullCompliance
} else if pass_rate >= 0.90 {
CertificationLevel::SubstantialCompliance
} else if pass_rate >= 0.70 {
CertificationLevel::PartialCompliance
} else {
CertificationLevel::NonCompliant
};
Ok(StandardComplianceResult {
standard,
compliant,
certification_level,
test_results,
tests_passed,
tests_total,
pass_rate,
average_error,
max_error,
validated_at: Utc::now(),
notes,
})
}
fn determine_certification_level(
&self,
results: &HashMap<ItuTStandard, StandardComplianceResult>,
) -> CertificationLevel {
let levels: Vec<_> = results.values().map(|r| r.certification_level).collect();
if levels
.iter()
.all(|&l| l == CertificationLevel::FullCompliance)
{
CertificationLevel::FullCompliance
} else if levels
.iter()
.all(|&l| l >= CertificationLevel::SubstantialCompliance)
{
CertificationLevel::SubstantialCompliance
} else if levels
.iter()
.all(|&l| l >= CertificationLevel::PartialCompliance)
{
CertificationLevel::PartialCompliance
} else {
CertificationLevel::NonCompliant
}
}
fn generate_test_signal(&self, sample_rate: u32, duration: f32) -> AudioBuffer {
use std::f32::consts::PI;
let num_samples = (sample_rate as f32 * duration) as usize;
let mut samples = Vec::with_capacity(num_samples);
for i in 0..num_samples {
let t = i as f32 / sample_rate as f32;
let signal = 0.3 * (2.0 * PI * 300.0 * t).sin()
+ 0.3 * (2.0 * PI * 1000.0 * t).sin()
+ 0.3 * (2.0 * PI * 3000.0 * t).sin();
samples.push(signal * 0.25); }
AudioBuffer::new(samples, sample_rate, 1)
}
fn generate_degraded_signal(&self, audio: &AudioBuffer, quality_factor: f32) -> AudioBuffer {
let mut rng = scirs2_core::random::thread_rng();
let samples: Vec<f32> = audio
.samples()
.iter()
.map(|&s| s * quality_factor + rng.random::<f32>() * 0.01 * (1.0 - quality_factor))
.collect();
AudioBuffer::new(samples, audio.sample_rate(), audio.channels())
}
fn measure_active_speech_level(&self, audio: &AudioBuffer) -> Result<f32, StandardsError> {
let samples = audio.samples();
let rms = (samples.iter().map(|&s| s * s).sum::<f32>() / samples.len() as f32).sqrt();
let db_level = if rms > 0.0 {
20.0 * rms.log10()
} else {
-96.0 };
Ok(db_level)
}
}
impl Default for ItuTComplianceValidator {
fn default() -> Self {
Self::new().expect("Failed to create default validator")
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_validator_creation() {
let validator = ItuTComplianceValidator::new();
assert!(validator.is_ok());
}
#[tokio::test]
async fn test_strict_validator_creation() {
let validator = ItuTComplianceValidator::new_strict();
assert!(validator.is_ok());
let v = validator.unwrap();
assert!(v.strict_mode);
assert!(v.default_tolerance < 0.01);
}
#[tokio::test]
async fn test_test_signal_generation() {
let validator = ItuTComplianceValidator::new().unwrap();
let signal = validator.generate_test_signal(16000, 1.0);
assert_eq!(signal.sample_rate(), 16000);
assert_eq!(signal.samples().len(), 16000);
}
#[tokio::test]
async fn test_degraded_signal_generation() {
let validator = ItuTComplianceValidator::new().unwrap();
let signal = validator.generate_test_signal(16000, 1.0);
let degraded = validator.generate_degraded_signal(&signal, 0.9);
assert_eq!(degraded.sample_rate(), signal.sample_rate());
assert_eq!(degraded.samples().len(), signal.samples().len());
}
#[tokio::test]
async fn test_p862_validation() {
let validator = ItuTComplianceValidator::new().unwrap();
let result = validator.validate_p862_pesq().await;
assert!(result.is_ok());
let res = result.unwrap();
assert_eq!(res.standard, ItuTStandard::P862Pesq);
assert!(!res.test_results.is_empty());
}
#[tokio::test]
async fn test_p863_validation() {
let validator = ItuTComplianceValidator::new().unwrap();
let result = validator.validate_p863_polqa().await;
assert!(result.is_ok());
let res = result.unwrap();
assert_eq!(res.standard, ItuTStandard::P863Polqa);
assert!(!res.test_results.is_empty());
}
#[tokio::test]
async fn test_p56_validation() {
let validator = ItuTComplianceValidator::new().unwrap();
let result = validator.validate_p56_loudness().await;
assert!(result.is_ok());
let res = result.unwrap();
assert_eq!(res.standard, ItuTStandard::P56Loudness);
}
#[tokio::test]
async fn test_full_validation() {
let validator = ItuTComplianceValidator::new().unwrap();
let report = validator.validate_all_standards().await;
assert!(report.is_ok());
let rep = report.unwrap();
assert_eq!(rep.results.len(), 3);
assert!(!rep.certification_id.is_empty());
}
#[tokio::test]
async fn test_badge_generation() {
let validator = ItuTComplianceValidator::new().unwrap();
let report = validator.validate_all_standards().await.unwrap();
let svg_badge = report.generate_badge(BadgeFormat::Svg);
assert!(svg_badge.is_ok());
assert!(svg_badge.unwrap().contains("<svg"));
let md_badge = report.generate_badge(BadgeFormat::Markdown);
assert!(md_badge.is_ok());
assert!(md_badge.unwrap().contains("ITU-T"));
let html_badge = report.generate_badge(BadgeFormat::Html);
assert!(html_badge.is_ok());
assert!(html_badge.unwrap().contains("<span"));
}
#[tokio::test]
async fn test_report_serialization() {
let validator = ItuTComplianceValidator::new().unwrap();
let report = validator.validate_all_standards().await.unwrap();
let json = report.to_json();
assert!(json.is_ok());
assert!(json.unwrap().contains("certification_id"));
}
#[tokio::test]
async fn test_certification_levels() {
assert!(CertificationLevel::FullCompliance > CertificationLevel::SubstantialCompliance);
assert!(CertificationLevel::SubstantialCompliance > CertificationLevel::PartialCompliance);
assert!(CertificationLevel::PartialCompliance > CertificationLevel::NonCompliant);
}
}