wehub-resource-sync
40 lines
1.2 KiB
Python
40 lines
1.2 KiB
Python
import base64
|
|
import os
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, Optional
|
|
import requests
|
|
import json
|
|
import gradio as gr
|
|
import uuid
|
|
|
|
|
|
def encode_image(img_path):
|
|
if not img_path:
|
|
return None
|
|
with open(img_path, "rb") as fin:
|
|
image_data = base64.b64encode(fin.read()).decode("utf-8")
|
|
return image_data
|
|
|
|
|
|
def get_latest_files(directory: str, file_types: list = ['.webm', '.zip']) -> Dict[str, Optional[str]]:
|
|
"""Get the latest recording and trace files"""
|
|
latest_files: Dict[str, Optional[str]] = {ext: None for ext in file_types}
|
|
|
|
if not os.path.exists(directory):
|
|
os.makedirs(directory, exist_ok=True)
|
|
return latest_files
|
|
|
|
for file_type in file_types:
|
|
try:
|
|
matches = list(Path(directory).rglob(f"*{file_type}"))
|
|
if matches:
|
|
latest = max(matches, key=lambda p: p.stat().st_mtime)
|
|
# Only return files that are complete (not being written)
|
|
if time.time() - latest.stat().st_mtime > 1.0:
|
|
latest_files[file_type] = str(latest)
|
|
except Exception as e:
|
|
print(f"Error getting latest {file_type} file: {e}")
|
|
|
|
return latest_files
|