CSC 8608 – Concepts avancés et applications du deep learning

# À exécuter à la racine du dépôt mkdir -p TP3/assets TP3/outputs

import os import torch import torchaudio import transformers import datasets def main(): print("=== TP3 sanity check ===") print("torch:", torch.__version__) print("torchaudio:", torchaudio.__version__) print("transformers:", transformers.__version__) print("datasets:", datasets.__version__) device = "cuda" if torch.cuda.is_available() else "cpu" print("device:", device) if device == "cuda": # TODO: compléter les informations GPU gpu_name = torch.cuda.get_device_name(________) gpu_mem_gb = torch.cuda.get_device_properties(________).total_memory / (1024**3) print("gpu_name:", gpu_name) print("gpu_mem_gb:", round(gpu_mem_gb, 2)) # Génère un mini signal audio (1 seconde) pour valider torchaudio sr = 16000 t = torch.linspace(0, 1, sr) wav = 0.1 * torch.sin(2 * torch.pi * 440.0 * t) # 440 Hz wav = wav.unsqueeze(0) # [1, time] mel = torchaudio.transforms.MelSpectrogram( sample_rate=sr, n_fft=400, hop_length=160, n_mels=80 )(wav) logmel = (mel + 1e-6).log() print("wav_shape:", tuple(wav.shape)) print("logmel_shape:", tuple(logmel.shape)) if __name__ == "__main__": main()

# Depuis la racine du dépôt python TP3/sanity_check.py

Hello, thank you for calling customer support. My name is Alex, and I will help you today. I’m calling about an order that arrived damaged. The package was delivered yesterday, but the screen is cracked. I would like a refund or a replacement as soon as possible. The order number is A X 1 9 7 3 5. You can reach me at john dot smith at example dot com. Also, my phone number is 555 0199. Thank you.

mkdir -p TP3/data ls -lh TP3/data/call_01.wav # Inspecter l'audio (au choix) ffprobe TP3/data/call_01.wav # ou soxi TP3/data/call_01.wav

# Exemple : convertir un fichier source en WAV mono 16 kHz ffmpeg -i TP3/data/________ -ac 1 -ar ________ TP3/data/call_01.wav

import torch import torchaudio def rms(x: torch.Tensor) -> float: return float(torch.sqrt(torch.mean(x ** 2)).item()) def clipping_rate(x: torch.Tensor, thr: float = 0.99) -> float: return float((x.abs() > thr).float().mean().item()) def main(): path = "TP3/data/call_01.wav" wav, sr = torchaudio.load(path) # wav: [channels, time] wav = wav.mean(dim=0, keepdim=True) # force mono [1, time] num_samples = wav.shape[________] duration_s = num_samples / sr print("path:", path) print("sr:", sr) print("shape:", tuple(wav.shape)) print("duration_s:", round(duration_s, 2)) print("rms:", round(rms(wav), 4)) print("clipping_rate:", round(clipping_rate(wav), 4)) if __name__ == "__main__": main()

python TP3/inspect_audio.py

import os import json from dataclasses import dataclass from typing import List, Tuple import torch import torchaudio # silero-vad (modèle VAD léger prêt à l’emploi) # Référence: https://github.com/snakers4/silero-vad # Installation (si besoin): pip install silero-vad from silero_vad import get_speech_timestamps @dataclass class Segment: start_s: float end_s: float def load_wav_mono_16k(path: str) -> Tuple[torch.Tensor, int]: wav, sr = torchaudio.load(path) # [C, T] wav = wav.mean(dim=0, keepdim=True) # mono [1, T] if sr != 16000: wav = torchaudio.functional.resample(wav, sr, 16000) sr = 16000 return wav.squeeze(0), sr # [T], sr def main(): in_path = "TP3/data/call_01.wav" out_path = "TP3/outputs/vad_segments_call_01.json" os.makedirs("TP3/outputs", exist_ok=True) wav, sr = load_wav_mono_16k(in_path) # wav: [T] duration_s = wav.numel() / sr model, utils = torch.hub.load( repo_or_dir="snakers4/silero-vad", model="silero_vad", trust_repo=True ) model.to("cpu").eval() # TODO: VAD -> timestamps (en indices samples) # Astuce: get_speech_timestamps attend un tenseur 1D float32 en 16 kHz speech_ts = get_speech_timestamps( wav.to(torch.float32), model, sampling_rate=________ ) # Convertir en segments en secondes segments: List[Segment] = [] for seg in speech_ts: start_s = seg["start"] / sr end_s = seg["end"] / sr segments.append(Segment(start_s=start_s, end_s=end_s)) # Filtrage simple : supprimer segments trop courts min_dur_s = 0.30 segments = [s for s in segments if (s.end_s - s.start_s) >= min_dur_s] # Stats total_speech_s = sum((s.end_s - s.start_s) for s in segments) speech_ratio = total_speech_s / max(duration_s, 1e-9) print("duration_s:", round(duration_s, 2)) print("num_segments:", len(segments)) print("total_speech_s:", round(total_speech_s, 2)) print("speech_ratio:", round(speech_ratio, 3)) # Sauvegarde JSON payload = { "audio_path": in_path, "sample_rate": sr, "duration_s": duration_s, "min_segment_s": min_dur_s, "segments": [{"start_s": s.start_s, "end_s": s.end_s} for s in segments], "stats": { "num_segments": len(segments), "total_speech_s": total_speech_s, "speech_ratio": speech_ratio } } with open(out_path, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2) print("saved:", out_path) if __name__ == "__main__": main()

python TP3/vad_segment.py cat TP3/outputs/vad_segments_call_01.json | head -n 60

# TODO: tester un filtrage plus strict min_dur_s = ________

import os import json import time from typing import Dict, Any, List import torch import torchaudio from transformers import pipeline def load_wav_mono_16k(path: str): wav, sr = torchaudio.load(path) # [C, T] wav = wav.mean(dim=0, keepdim=True) # mono [1, T] if sr != 16000: wav = torchaudio.functional.resample(wav, sr, 16000) sr = 16000 return wav.squeeze(0), sr # [T], sr def main(): audio_path = "TP3/data/call_01.wav" vad_path = "TP3/outputs/vad_segments_call_01.json" out_path = "TP3/outputs/asr_call_01.json" os.makedirs("TP3/outputs", exist_ok=True) wav, sr = load_wav_mono_16k(audio_path) audio_duration_s = wav.numel() / sr with open(vad_path, "r", encoding="utf-8") as f: vad_payload = json.load(f) segments = vad_payload["segments"] # list of {start_s, end_s} # Choix modèle : petit si CPU, plus gros si GPU (mais rester raisonnable) model_id = "openai/________" # trouver un modèle de whisper sur HuggingFace device = 0 if torch.cuda.is_available() else -1 asr = pipeline( task="automatic-speech-recognition", model=model_id, device=device ) t0 = time.time() results: List[Dict[str, Any]] = [] for i, seg in enumerate(segments): start_s = float(seg["start_s"]) end_s = float(seg["end_s"]) start = int(start_s * sr) end = int(end_s * sr) seg_wav = wav[start:end] # HF pipeline attend soit un chemin, soit un dict {"array": ..., "sampling_rate": ...} inp = {"array": seg_wav.numpy(), "sampling_rate": sr} generate_kwargs = { "language": "english" } out = asr(inp, generate_kwargs=generate_kwargs) # out: {"text": "...", ...} text = out.get("text", "").strip() results.append({ "segment_id": i, "start_s": start_s, "end_s": end_s, "text": text }) t1 = time.time() elapsed_s = t1 - t0 rtf = elapsed_s / max(audio_duration_s, 1e-9) # Reconstruction transcript complet (simple) full_text = " ".join([r["text"] for r in results]).strip() payload = { "audio_path": audio_path, "model_id": model_id, "device": "cuda" if device == 0 else "cpu", "audio_duration_s": audio_duration_s, "elapsed_s": elapsed_s, "rtf": rtf, "segments": results, "full_text": full_text } with open(out_path, "w", encoding="utf-8") as f: json.dump(payload, f, indent=2) print("model_id:", model_id) print("device:", payload["device"]) print("audio_duration_s:", round(audio_duration_s, 2)) print("elapsed_s:", round(elapsed_s, 2)) print("rtf:", round(rtf, 3)) print("saved:", out_path) if __name__ == "__main__": main()

python TP3/asr_whisper.py

cat TP3/outputs/asr_call_01.json | head -n 120

import os import re import json from typing import Dict, Any, List, Tuple from collections import Counter EMAIL_RE = re.compile(r"([A-Za-z0-9._%+-]+)@([A-Za-z0-9.-]+)\.([A-Za-z]{2,})") # Téléphone US simplifié (ex: 555 0199, 555-0199, 5550199). On masque large. PHONE_RE = re.compile(r"\b(\d[\d\-\s]{5,}\d)\b") INTENTS = { "refund_or_replacement": ["refund", "replacement", "damaged", "cracked", "broken"], "delivery_issue": ["delivered", "package", "arrived", "yesterday", "order"], "general_support": ["help", "support", "thank you", "calling"], } STOPWORDS = set([ "the","a","an","and","or","to","for","of","in","on","is","it","i","you","we","my","your", "was","were","be","as","at","but","this","that","with","about","today" ]) def redact_pii(text: str) -> Tuple[str, Dict[str, int]]: stats = {"emails": 0, "phones": 0} # TODO: masquer emails def _email_sub(m): stats["emails"] += 1 return "[REDACTED_EMAIL]" text = EMAIL_RE.sub(_email_sub, text) # TODO: masquer téléphones def _phone_sub(m): stats["phones"] += 1 return "[REDACTED_PHONE]" text = PHONE_RE.sub(_phone_sub, text) return text, stats def normalize(text: str) -> str: # minuscule + espaces t = text.lower() t = re.sub(r"\s+", " ", t).strip() return t def tokenize(text: str) -> List[str]: # tokens alphabétiques simples toks = re.findall(r"[a-z]+", text.lower()) return [w for w in toks if w not in STOPWORDS and len(w) > 2] def score_intents(text: str) -> Dict[str, int]: t = normalize(text) scores: Dict[str, int] = {} for intent, kws in INTENTS.items(): s = 0 for kw in kws: # TODO: compter occurrences naïvement s += t.count(________) scores[intent] = s return scores def pick_intent(scores: Dict[str, int]) -> str: # intention avec meilleur score ; fallback si tous à 0 best_intent = max(scores.items(), key=lambda kv: kv[1])[0] if scores[best_intent] == 0: return "unknown" return best_intent def main(): in_path = "TP3/outputs/asr_call_01.json" out_path = "TP3/outputs/call_summary_call_01.json" os.makedirs("TP3/outputs", exist_ok=True) with open(in_path, "r", encoding="utf-8") as f: asr = json.load(f) full_text = asr["full_text"] redacted_text, pii_stats = redact_pii(full_text) tokens = tokenize(redacted_text) top_terms = Counter(tokens).most_common(10) intent_scores = score_intents(redacted_text) intent = pick_intent(intent_scores) summary = { "audio_path": asr["audio_path"], "model_id": asr["model_id"], "device": asr["device"], "audio_duration_s": asr["audio_duration_s"], "elapsed_s": asr["elapsed_s"], "rtf": asr["rtf"], "pii_stats": pii_stats, "intent_scores": intent_scores, "intent": intent, "top_terms": top_terms, "redacted_text": redacted_text } with open(out_path, "w", encoding="utf-8") as f: json.dump(summary, f, indent=2) print("intent:", intent) print("pii_stats:", pii_stats) print("top_terms:", top_terms[:5]) print("saved:", out_path) if __name__ == "__main__": main()

python TP3/callcenter_analytics.py

cat TP3/outputs/call_summary_call_01.json | head -n 120

# Regex "strict" email (fonctionne après normalisation) EMAIL_RE = re.compile(r"\b[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}\b") # Téléphone : 7+ digits au total, séparateurs optionnels (espaces, -, .) PHONE_RE = re.compile(r"\b(?:\d[\s\-\.]*){7,}\d\b") DIGIT_WORDS = { "zero":"0","oh":"0","o":"0", "one":"1","won":"1", "two":"2","too":"2","to":"2", "three":"3","free":"3","tree":"3", "four":"4","for":"4", "five":"5","fife":"5","hi":"5", "six":"6", "seven":"7", "eight":"8","ate":"8", "nine":"9", } def preclean(text: str) -> str: t = text.lower() # Séparer chiffres collés à des mots : "5550199thank" -> "5550199 thank" t = re.sub(r"(\d)([a-z])", r"\1 \2", t) t = re.sub(r"([a-z])(\d)", r"\1 \2", t) # Ajouter un espace après ponctuation collée entre deux tokens : "hello.thank" -> "hello. thank" t = re.sub(r"([a-z0-9])([.,!?])([a-z0-9])", r"\1\2 \3", t) # Apostrophes gênantes pour les heuristiques (ex: "don't" -> "dont") t = t.replace("'", "").replace("’", "").replace("...", " ") # Compacter espaces t = re.sub(r"\s+", " ", t).strip() return t def normalize_spelled_tokens(text: str) -> str: """ Normalisation pragmatique: - 'dot' -> '.', 'at' -> '@' (utile email) - mots-chiffres -> digits - collage des séquences de digits séparés (>= 6 digits) """ t = preclean(text) # Normalisation email parlée t = re.sub(r"\bdot\b", ".", t) t = re.sub(r"\bat\b", "@", t) t = re.sub(r"\s*([.@])\s*", r"\1", t) # supprime espaces autour de . et @ # Remplacer mots->digits (token-level) def _tok_sub(m): w = m.group(0) return DIGIT_WORDS.get(w, w) t = re.sub(r"\b[a-z]+\b", _tok_sub, t) # Coller les digits isolés : "5 5 5 0 1 9 9" -> "5550199" def _collapse(m): digits = re.findall(r"\d", m.group(0)) return "".join(digits) t = re.sub(r"(?:\b\d\b[\s,\-\.]*){6,}\b", _collapse, t) # Re-séparer digits/lettres au cas où après collapse t = re.sub(r"(\d)([a-z])", r"\1 \2", t) return t def redact_order_id(text: str) -> Tuple[str, int]: """ Masque un identifiant après 'order number is' même s’il est épelé (a.x.1.9.7.3.5). """ count = 0 pattern = re.compile(r"\border number is\b\s+([a-z0-9\s\.\-]{3,80})", re.IGNORECASE) def _sub(m): nonlocal count span = m.group(1) cleaned = re.findall(r"[A-Za-z0-9]", span) if len(cleaned) >= 5: count += 1 return "order number is [REDACTED_ORDER]" return m.group(0) return pattern.sub(_sub, text), count def redact_spoken_email(text: str) -> Tuple[str, int]: """ 1) masque les vrais emails détectables (après normalisation) 2) sinon masque par contexte : "reach me ..." jusqu'à un marqueur (also/phone/order/thank) ou fin """ count = 0 # (1) email standard def _email_sub(m): nonlocal count count += 1 return "[REDACTED_EMAIL]" t = EMAIL_RE.sub(_email_sub, text) if count > 0: return t, count # (2) fallback par contexte (robuste même si ASR “massacre” le local-part) ctx = re.compile( r"(\byou can reach me\b|\breach me\b)\s*(?:@)?\s*([a-z0-9.\s]{3,80})" r"(?=\b(?:also|my phone|phone number|order number|thank)\b|$)", re.IGNORECASE ) def _ctx_sub(m): nonlocal count count += 1 return m.group(1) + " [REDACTED_EMAIL]" return ctx.sub(_ctx_sub, t), count def redact_phone(text: str) -> Tuple[str, int]: count = 0 def _sub(m): nonlocal count count += 1 return "[REDACTED_PHONE]" return PHONE_RE.sub(_sub, text), count def redact_pii(text: str) -> Tuple[str, Dict[str, int]]: """ Post-traitement + redaction PII: - normalise tokens épelés (digits, dot/at) - masque order id (contexte) - masque email (standard ou contexte) - masque téléphone (séquences de digits) """ stats = {"emails": 0, "phones": 0, "orders": 0} t = normalize_spelled_tokens(text) t, n_orders = redact_order_id(t) stats["orders"] += n_orders t, n_emails = redact_spoken_email(t) stats["emails"] += n_emails t, n_phones = redact_phone(t) stats["phones"] += n_phones return t, stats

import os import time import numpy as np import torch from transformers import pipeline import torchaudio def main(): os.makedirs("TP3/outputs", exist_ok=True) text = ( "Thanks for calling. I am sorry your order arrived damaged. " "I can offer a replacement or a refund. " "Please confirm your preferred option." ) # Modèle TTS léger (anglais) tts_model_id = "facebook/________" # Trouvez un modèle de TTS (ici, donné par Facebook par exemple) device = 0 if torch.cuda.is_available() else -1 tts = pipeline( task="text-to-speech", model=tts_model_id, device=device ) t0 = time.time() out = tts(text) t1 = time.time() audio = np.asarray(out["audio"], dtype=np.float32) # numpy array sr = int(out["sampling_rate"]) elapsed_s = t1 - t0 audio_dur_s = float(audio.shape[1] / float(sr)) rtf = elapsed_s / max(audio_dur_s, 1e-9) # normaliser la forme vers [1, T] if audio.ndim == 1: # [T] audio = audio[None, :] # [1, T] elif audio.ndim == 2: # cas [T, 1] -> [1, T] if audio.shape[1] == 1: audio = audio.T # [1, T] # cas [1, T] déjà OK elif audio.shape[0] == 1: pass else: # cas multi-canaux [T, C] -> [C, T] audio = audio.T else: raise ValueError(f"Unexpected audio shape: {audio.shape}") out_wav = "outputs/tts_reply_call_01.wav" # ex: tts_reply_call_01 wav_t = torch.from_numpy(audio.astype(np.float32)) # [C, T] torchaudio.save(out_wav, wav_t, sr) print("tts_model_id:", tts_model_id) print("device:", "cuda" if device == 0 else "cpu") print("audio_dur_s:", round(audio_dur_s, 2)) print("elapsed_s:", round(elapsed_s, 2)) print("rtf:", round(rtf, 3)) print("saved:", out_wav) if __name__ == "__main__": main()

python TP3/tts_reply.py

ffprobe TP3/outputs/tts_reply_call_01.wav # ou soxi TP3/outputs/tts_reply_call_01.wav

import time import torch from transformers import pipeline def main(): wav_path = "TP3/outputs/________.wav" model_id = "openai/________" device = 0 if torch.cuda.is_available() else -1 asr = pipeline( task="automatic-speech-recognition", model=model_id, device=device ) t0 = time.time() generate_kwargs = { "language": "english" } out = asr(wav_path, generate_kwargs=generate_kwargs) # out: {"text": "...", ...} t1 = time.time() print("model_id:", model_id) print("elapsed_s:", round(t1 - t0, 2)) print("text:", out.get("text","").strip()) if __name__ == "__main__": main()

import os import json import subprocess from pathlib import Path def run(cmd: str): print(">>", cmd) subprocess.run(cmd, shell=True, check=True) def load_json(path: str): with open(path, "r", encoding="utf-8") as f: return json.load(f) def main(): os.makedirs("TP3/outputs", exist_ok=True) # 1) VAD run("python TP3/________") vad = load_json("TP3/outputs/vad_segments_call_01.json") # 2) ASR run("python TP3/________") asr = load_json("TP3/outputs/asr_call_01.json") # 3) Analytics run("python TP3/________") summ = load_json("TP3/outputs/call_summary_call_01.json") # 4) TTS (optionnel) : si le script existe, on lance tts_path = Path("TP3/tts_reply.py") tts_done = False if tts_path.exists(): run("python TP3/tts_reply.py") tts_done = True # Résumé final (léger) summary = { "audio_path": vad.get("audio_path"), "duration_s": vad.get("duration_s"), "num_segments": vad.get("stats", {}).get("num_segments"), "speech_ratio": vad.get("stats", {}).get("speech_ratio"), "asr_model": asr.get("model_id"), "asr_device": asr.get("device"), "asr_rtf": asr.get("rtf"), "intent": summ.get("intent"), "pii_stats": summ.get("pii_stats"), "tts_generated": tts_done } out_path = "TP3/outputs/pipeline_summary_call_01.json" with open(out_path, "w", encoding="utf-8") as f: json.dump(summary, f, indent=2) print("=== PIPELINE SUMMARY ===") for k, v in summary.items(): print(f"{k}: {v}") print("saved:", out_path) if __name__ == "__main__": main()

python TP3/run_pipeline.py

cat TP3/outputs/pipeline_summary_call_01.json