Visualizer for tracer JSON files.
| 11 | |
| 12 | |
| 13 | class TracerVisualizer: |
| 14 | """Visualizer for tracer JSON files.""" |
| 15 | |
| 16 | def __init__(self, tracer_json_path: str, port: int = 5000): |
| 17 | """Initialize the visualizer. |
| 18 | |
| 19 | Args: |
| 20 | tracer_json_path: Path to the tracer.json file |
| 21 | port: Port number for the web server (default: 5000) |
| 22 | """ |
| 23 | self.tracer_json_path = Path(tracer_json_path) |
| 24 | self.port = port |
| 25 | self.app = Flask(__name__, |
| 26 | template_folder=str(Path(__file__).parent / "templates"), |
| 27 | static_folder=str(Path(__file__).parent / "static")) |
| 28 | self.records: List[Dict[str, Any]] = [] |
| 29 | self._data_lock = threading.Lock() |
| 30 | self._stop_reload_thread = threading.Event() |
| 31 | self._setup_routes() |
| 32 | self._load_data() |
| 33 | |
| 34 | # Start auto-reload thread (reload every 60 seconds) |
| 35 | self._reload_thread = threading.Thread(target=self._auto_reload_data, daemon=True) |
| 36 | self._reload_thread.start() |
| 37 | |
| 38 | def _load_data(self): |
| 39 | """Load data from tracer.json file.""" |
| 40 | if not self.tracer_json_path.exists(): |
| 41 | print(f"Warning: Tracer JSON file not found: {self.tracer_json_path}") |
| 42 | return |
| 43 | |
| 44 | try: |
| 45 | with open(self.tracer_json_path, 'r', encoding='utf-8') as f: |
| 46 | new_records = json.load(f) |
| 47 | |
| 48 | # Update records with thread lock |
| 49 | with self._data_lock: |
| 50 | self.records = new_records |
| 51 | |
| 52 | print(f"Data loaded: {len(self.records)} records from {self.tracer_json_path}") |
| 53 | except Exception as e: |
| 54 | print(f"Error loading data: {e}") |
| 55 | |
| 56 | def _auto_reload_data(self): |
| 57 | """Auto-reload data every 60 seconds in background thread.""" |
| 58 | while not self._stop_reload_thread.is_set(): |
| 59 | time.sleep(60) # Wait 60 seconds |
| 60 | if not self._stop_reload_thread.is_set(): |
| 61 | print("Auto-reloading data...") |
| 62 | self._load_data() |
| 63 | |
| 64 | def _extract_account_value(self, record: Dict[str, Any]) -> Optional[float]: |
| 65 | """Extract account value from a record. |
| 66 | """ |
| 67 | try: |
| 68 | observation = record.get("observation", {}) |
| 69 | if "online_hyperliquid" in observation: |
| 70 | hyperliquid = observation.get("online_hyperliquid", {}) |