Spaces:
Running
Running
File size: 2,689 Bytes
1d8403e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 | # ==================================================================================================
# DEEPFAKE AUDIO - utils/profiler.py (Diagnostic Performance Telemetry)
# ==================================================================================================
#
# 📝 DESCRIPTION
# This module provides low-overhead execution timing and performance telemetry
# utilities. It is used throughout the pipeline to monitor the average latency
# of neural inference stages, enabling researchers to identify computational
# bottlenecks in real-time.
#
# 👤 AUTHORS
# - Amey Thakur (https://github.com/Amey-Thakur)
# - Mega Satish (https://github.com/msatmod)
#
# 🤝🏻 CREDITS
# Original Real-Time Voice Cloning methodology by CorentinJ
# Repository: https://github.com/CorentinJ/Real-Time-Voice-Cloning
#
# 🔗 PROJECT LINKS
# Repository: https://github.com/Amey-Thakur/DEEPFAKE-AUDIO
# Video Demo: https://youtu.be/i3wnBcbHDbs
# Research: https://github.com/Amey-Thakur/DEEPFAKE-AUDIO/blob/main/DEEPFAKE-AUDIO.ipynb
#
# 📜 LICENSE
# Released under the MIT License
# Release Date: 2021-02-06
# ==================================================================================================
from time import perf_counter as timer
from collections import OrderedDict
import numpy as np
class Profiler:
def __init__(self, summarize_every=5, disabled=False):
self.last_tick = timer()
self.logs = OrderedDict()
self.summarize_every = summarize_every
self.disabled = disabled
def tick(self, name):
if self.disabled:
return
# Log the time needed to execute that function
if not name in self.logs:
self.logs[name] = []
if len(self.logs[name]) >= self.summarize_every:
self.summarize()
self.purge_logs()
self.logs[name].append(timer() - self.last_tick)
self.reset_timer()
def purge_logs(self):
for name in self.logs:
self.logs[name].clear()
def reset_timer(self):
self.last_tick = timer()
def summarize(self):
n = max(map(len, self.logs.values()))
assert n == self.summarize_every
print("\nAverage execution time over %d steps:" % n)
name_msgs = ["%s (%d/%d):" % (name, len(deltas), n) for name, deltas in self.logs.items()]
pad = max(map(len, name_msgs))
for name_msg, deltas in zip(name_msgs, self.logs.values()):
print(" %s mean: %4.0fms std: %4.0fms" %
(name_msg.ljust(pad), np.mean(deltas) * 1000, np.std(deltas) * 1000))
print("", flush=True)
|