File size: 5,513 Bytes
e38b1c0 |
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 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 |
import requests
import time
from typing import Dict, Any, Optional
import os
class MLTracker:
"""Python client for ML Tracker - Free W&B Alternative"""
def __init__(self, api_key: str, base_url: str = "https://your-space-url.hf.space"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.current_experiment = None
self.step = 0
def init(self, experiment_name: str, config: Optional[Dict[str, Any]] = None):
self.current_experiment = experiment_name
self.step = 0
if config:
self.log_config(experiment_name, config)
def log_config(self, experiment_name: str, config: Dict[str, Any]):
try:
response = requests.post(
f"{self.base_url}/api/log",
json={
"api_key": self.api_key,
"experiment": experiment_name,
"step": 0,
"metrics": {},
"config": config,
"timestamp": time.time()
}
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"Warning: Failed to log config: {e}")
return None
def log(self, metrics: Dict[str, Any], step: Optional[int] = None):
if not self.current_experiment:
raise ValueError("No experiment initialized. Call init() first.")
if step is None:
self.step += 1
step = self.step
else:
self.step = max(self.step, step)
try:
response = requests.post(
f"{self.base_url}/api/log",
json={
"api_key": self.api_key,
"experiment": self.current_experiment,
"step": step,
"metrics": metrics,
"timestamp": time.time()
}
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"Warning: Failed to log metrics: {e}")
return None
def delete_experiment(self, experiment_name: str):
try:
response = requests.delete(
f"{self.base_url}/api/experiment/{experiment_name}",
params={"api_key": self.api_key}
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"Error deleting experiment: {e}")
return None
def get_experiments(self):
try:
response = requests.get(
f"{self.base_url}/api/experiments",
params={"api_key": self.api_key}
)
response.raise_for_status()
return response.json().get("experiments", [])
except Exception as e:
print(f"Error fetching experiments: {e}")
return []
def get_experiment(self, experiment_name: str):
try:
response = requests.get(
f"{self.base_url}/api/experiment/{experiment_name}",
params={"api_key": self.api_key}
)
response.raise_for_status()
return response.json()
except Exception as e:
print(f"Error fetching experiment: {e}")
return None
def finish(self):
self.current_experiment = None
self.step = 0
# Global tracker and convenience functions
_global_tracker = None
def init(experiment_name: str, config: Optional[Dict[str, Any]] = None,
api_key: Optional[str] = None, base_url: Optional[str] = None):
global _global_tracker
if api_key is None:
api_key = os.environ.get("ML_TRACKER_API_KEY")
if not api_key:
raise ValueError("API key not provided and ML_TRACKER_API_KEY not set")
if base_url is None:
base_url = os.environ.get("ML_TRACKER_BASE_URL", "https://your-space-url.hf.space")
_global_tracker = MLTracker(api_key, base_url)
_global_tracker.init(experiment_name, config)
def log(metrics: Dict[str, Any], step: Optional[int] = None):
if _global_tracker is None:
raise ValueError("No experiment initialized. Call init() first.")
return _global_tracker.log(metrics, step)
def finish():
global _global_tracker
if _global_tracker:
_global_tracker.finish()
_global_tracker = None
if __name__ == "__main__":
# Example usage
tracker = MLTracker(
api_key="your-api-key-here",
base_url="https://your-space-url.hf.space"
)
tracker.init("my_experiment", config={
"model": "ResNet50",
"dataset": "CIFAR-10",
"learning_rate": 0.001,
"batch_size": 32
})
for epoch in range(10):
loss = 2.0 * (0.9 ** epoch) + 0.1
accuracy = 1.0 - (0.9 ** epoch)
tracker.log({
"loss": loss,
"accuracy": accuracy,
"epoch": epoch
})
tracker.finish()
init("another_experiment", config={"model": "BERT"})
for step in range(5):
log({"loss": 1.0 / (step + 1), "step": step})
finish()
experiments = tracker.get_experiments()
print(f"Found {len(experiments)} experiments")
if experiments:
exp_data = tracker.get_experiment(experiments[0]["experiment"])
if exp_data:
print(f"Experiment has {len(exp_data.get('metrics', []))} metric entries")
|