File size: 3,316 Bytes
007b0fc | 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 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import inspect
import time
from datetime import datetime
from functools import wraps
from typing import Any, Callable
import streamlit as st
def summarize_parameter_value(value: str, max_length: int = 100) -> str:
"""Summarize parameter values that are too long or complex."""
if not value:
return ""
# Handle DataFrames
if "DataFrame" in value and "[" in value and "]" in value:
try:
dims = value[value.find("[") + 1 : value.find("]")]
return f"DataFrame[{dims}]"
except Exception:
return "DataFrame"
# Handle lists, tuples, and other sequences
if value.startswith(("[", "(", "{")):
try:
item_count = value.count(",") + 1
return f"{value[:20]}... ({item_count} items)"
except Exception:
return f"{value[:20]}..."
# Handle long strings
if len(value) > max_length:
return f"{value[:max_length]}..."
return value
def timer(include_params: bool = False) -> Callable:
"""
Decorator to time function execution and store results in session state.
Args:
include_params: Whether to include function parameters in timing logs
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
if not st.session_state.get("ENABLE_TIMING", False):
return func(*args, **kwargs)
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
duration = end - start
# Initialize timing stats if needed
if "timing_stats" not in st.session_state:
st.session_state.timing_stats = {}
st.session_state.timing_logs = []
# Initialize list for this function if needed
if func.__name__ not in st.session_state.timing_stats:
st.session_state.timing_stats[func.__name__] = []
# Append new duration
st.session_state.timing_stats[func.__name__].append(duration)
# Create log entry
log_entry = {
"timestamp": datetime.now().isoformat(),
"function": func.__name__,
"duration": duration,
}
if include_params:
# Get parameter names from function signature
sig = inspect.signature(func)
param_names = list(sig.parameters.keys())
# Combine args and kwargs into a parameter dictionary
param_values = {}
for i, arg in enumerate(args):
if i < len(param_names):
param_values[param_names[i]] = summarize_parameter_value(
str(arg),
max_length=40,
)
param_values.update(
{
k: summarize_parameter_value(str(v), max_length=40)
for k, v in kwargs.items()
}
)
log_entry["parameters"] = param_values
st.session_state.timing_logs.append(log_entry)
return result
return wrapper
return decorator
|