Create utils/HFManager.py

utils/HFManager.py

@@ -0,0 +1,91 @@
+import json
+from huggingface_hub import HfApi, Repository, hf_hub_download
+from datetime import datetime
+from typing import List, Dict, Optional
+import os
+import logging
+
+def fetch_training_metrics_commits(repo_id: str, token: Optional[str] = None) -> List[Dict]:
+    """
+    Fetch training metrics from a Hugging Face repository.
+    
+    Args:
+        repo_id (str): The repository ID
+        token (Optional[str]): Hugging Face API token
+    """
+    try:
+        api = HfApi(token=token)
+        commits = api.list_repo_commits(repo_id=repo_id)
+        
+        training_metrics = []
+        processed_commits = 0
+        failed_commits = 0
+
+        print(f"Found {len(commits)} total commits in repository")
+
+        # Create a clean cache directory
+        cache_dir = os.path.expanduser("~/.cache/dashboard_metrics")
+        os.makedirs(cache_dir, exist_ok=True)
+
+        for commit in commits:
+            try:
+                files = api.list_repo_tree(
+                    repo_id=repo_id, 
+                    revision=commit.commit_id
+                )
+                json_files = [f for f in files if f.path.endswith('.json')]
+
+                for json_file in json_files:
+                    try:
+                        # Use custom cache directory
+                        local_path = hf_hub_download(
+                            repo_id=repo_id,
+                            filename=json_file.path,
+                            revision=commit.commit_id,
+                            cache_dir=cache_dir,
+                            force_download=True  # Always get fresh copy
+                        )
+
+                        with open(local_path, 'r') as f:
+                            metrics_data = json.load(f)
+
+                        if isinstance(metrics_data, dict) and "metrics" in metrics_data:
+                            miner_uid = metrics_data.get("miner_uid")
+                            job_id = metrics_data["metrics"].get("job_id")
+
+                            if miner_uid and job_id:
+                                metrics_entry = {
+                                    "model_repo": metrics_data.get("model_repo", "unknown"),
+                                    "metrics": metrics_data["metrics"],
+                                    "miner_uid": miner_uid,
+                                    "job_id": job_id,
+                                    "timestamp": metrics_data.get("timestamp", "unknown")
+                                }
+                                training_metrics.append(metrics_entry)
+                                processed_commits += 1
+
+                    except Exception as e:
+                        failed_commits += 1
+                        logging.warning(f"Error processing file {json_file.path}: {str(e)}")
+                        continue
+
+            except Exception as e:
+                failed_commits += 1
+                logging.warning(f"Error processing commit {commit.commit_id}: {str(e)}")
+                continue
+
+        filtered_metrics = [
+            entry for entry in training_metrics 
+            if entry.get('miner_uid') and entry['metrics'].get('job_id')
+        ]
+
+        logging.info(f"Successfully processed {processed_commits} commits with valid metrics")
+        if failed_commits > 0:
+            logging.warning(f"Failed to process {failed_commits} commits")
+            
+        return filtered_metrics
+
+    except Exception as e:
+        logging.error(f"Error fetching commits: {str(e)}")
+        return []
+