Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 5,482 Bytes
4d8a789 0972775 4d8a789 2ca459e 0972775 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 2ca459e 4d8a789 | 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 | """
Utility functions for Hugging Face tools
Ported from: hf-mcp-server/packages/mcp/src/jobs/formatters.ts
Includes GPU memory validation for job submissions
"""
import json
from datetime import datetime
from typing import Any, Dict, List, Optional
def truncate(text: str, max_length: int) -> str:
"""Truncate a string to a maximum length with ellipsis"""
if len(text) <= max_length:
return text
return text[: max_length - 3] + "..."
def format_date(date_str: Optional[str]) -> str:
"""Format a date string to a readable format"""
if not date_str:
return "N/A"
try:
date = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
return date.strftime("%Y-%m-%d %H:%M:%S")
except Exception:
return date_str
def format_command(command: Optional[List[str]]) -> str:
"""Format command array as a single string"""
if not command or len(command) == 0:
return "N/A"
return " ".join(command)
def get_image_or_space(job: Dict[str, Any]) -> str:
"""Get image/space identifier from job"""
if job.get("spaceId"):
return job["spaceId"]
if job.get("dockerImage"):
return job["dockerImage"]
return "N/A"
def format_jobs_table(jobs: List[Dict[str, Any]]) -> str:
"""Format jobs as a markdown table"""
if len(jobs) == 0:
return "No jobs found."
# Calculate dynamic ID column width
longest_id_length = max(len(job["id"]) for job in jobs)
id_column_width = max(longest_id_length, len("JOB ID"))
# Define column widths
col_widths = {
"id": id_column_width,
"image": 20,
"command": 30,
"created": 19,
"status": 12,
}
# Build header
header = f"| {'JOB ID'.ljust(col_widths['id'])} | {'IMAGE/SPACE'.ljust(col_widths['image'])} | {'COMMAND'.ljust(col_widths['command'])} | {'CREATED'.ljust(col_widths['created'])} | {'STATUS'.ljust(col_widths['status'])} |"
separator = f"|{'-' * (col_widths['id'] + 2)}|{'-' * (col_widths['image'] + 2)}|{'-' * (col_widths['command'] + 2)}|{'-' * (col_widths['created'] + 2)}|{'-' * (col_widths['status'] + 2)}|"
# Build rows
rows = []
for job in jobs:
job_id = job["id"]
image = truncate(get_image_or_space(job), col_widths["image"])
command = truncate(format_command(job.get("command")), col_widths["command"])
created = truncate(format_date(job.get("createdAt")), col_widths["created"])
status = truncate(job["status"]["stage"], col_widths["status"])
rows.append(
f"| {job_id.ljust(col_widths['id'])} | {image.ljust(col_widths['image'])} | {command.ljust(col_widths['command'])} | {created.ljust(col_widths['created'])} | {status.ljust(col_widths['status'])} |"
)
return "\n".join([header, separator] + rows)
def format_scheduled_jobs_table(jobs: List[Dict[str, Any]]) -> str:
"""Format scheduled jobs as a markdown table"""
if len(jobs) == 0:
return "No scheduled jobs found."
# Calculate dynamic ID column width
longest_id_length = max(len(job["id"]) for job in jobs)
id_column_width = max(longest_id_length, len("ID"))
# Define column widths
col_widths = {
"id": id_column_width,
"schedule": 12,
"image": 18,
"command": 25,
"lastRun": 19,
"nextRun": 19,
"suspend": 9,
}
# Build header
header = f"| {'ID'.ljust(col_widths['id'])} | {'SCHEDULE'.ljust(col_widths['schedule'])} | {'IMAGE/SPACE'.ljust(col_widths['image'])} | {'COMMAND'.ljust(col_widths['command'])} | {'LAST RUN'.ljust(col_widths['lastRun'])} | {'NEXT RUN'.ljust(col_widths['nextRun'])} | {'SUSPENDED'.ljust(col_widths['suspend'])} |"
separator = f"|{'-' * (col_widths['id'] + 2)}|{'-' * (col_widths['schedule'] + 2)}|{'-' * (col_widths['image'] + 2)}|{'-' * (col_widths['command'] + 2)}|{'-' * (col_widths['lastRun'] + 2)}|{'-' * (col_widths['nextRun'] + 2)}|{'-' * (col_widths['suspend'] + 2)}|"
# Build rows
rows = []
for job in jobs:
job_id = job["id"]
schedule = truncate(job["schedule"], col_widths["schedule"])
image = truncate(get_image_or_space(job["jobSpec"]), col_widths["image"])
command = truncate(
format_command(job["jobSpec"].get("command")), col_widths["command"]
)
last_run = truncate(format_date(job.get("lastRun")), col_widths["lastRun"])
next_run = truncate(format_date(job.get("nextRun")), col_widths["nextRun"])
suspend = "Yes" if job.get("suspend") else "No"
rows.append(
f"| {job_id.ljust(col_widths['id'])} | {schedule.ljust(col_widths['schedule'])} | {image.ljust(col_widths['image'])} | {command.ljust(col_widths['command'])} | {last_run.ljust(col_widths['lastRun'])} | {next_run.ljust(col_widths['nextRun'])} | {suspend.ljust(col_widths['suspend'])} |"
)
return "\n".join([header, separator] + rows)
def format_job_details(jobs: Any) -> str:
"""Format job details as JSON in a markdown code block"""
job_array = jobs if isinstance(jobs, list) else [jobs]
json_str = json.dumps(job_array, indent=2)
return f"```json\n{json_str}\n```"
def format_scheduled_job_details(jobs: Any) -> str:
"""Format scheduled job details as JSON in a markdown code block"""
job_array = jobs if isinstance(jobs, list) else [jobs]
json_str = json.dumps(job_array, indent=2)
return f"```json\n{json_str}\n```"
|