Spaces:
Runtime error
Runtime error
# Copyright (c) Facebook, Inc. and its affiliates. All rights reserved. | |
# | |
# This source code is licensed under the BSD license found in the | |
# LICENSE file in the root directory of this source tree. | |
import time | |
from dataclasses import dataclass | |
from typing import Optional | |
class PhaseStats: | |
name: str | |
tokens: int | |
time: float | |
def show(self) -> str: | |
tps = self.tokens / self.time | |
return ( | |
f"[{self.name}] " | |
f"generated tokens: {self.tokens}" | |
f" - total time: {self.time:.3f}s" | |
f" - {tps:.1f} tokens per second" | |
) | |
class Stats: | |
""" | |
Generation stats, split by phases. | |
""" | |
def __init__(self): | |
self.phases = [] | |
self.current = None | |
def end_phase(self, tokens: int, now: Optional[float] = None): | |
"""Terminate the current phase.""" | |
if self.current is None: | |
return | |
if now is None: | |
now = time.time() | |
cname, ctokens, ctime = self.current | |
stats = PhaseStats( | |
name=cname, | |
tokens=tokens - ctokens, | |
time=now - ctime, | |
) | |
self.phases.append(stats) | |
def phase(self, name: str, tokens: int = 0): | |
""" | |
Start a new phase, and terminate the current one, | |
if one is ongoing. | |
""" | |
now = time.time() | |
self.end_phase(tokens, now) | |
self.current = (name, tokens, now) | |