Spaces:
Running
Running
| # ================================================================================================== | |
| # 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) | |