Coverage for apps/streamwise_app.py: 61%

1""" 

2StreamWise HTTP application. 

3""" 

4 

5import sys 

6import os 

7import json 

8import errno 

9import argparse 

10import logging 

11import traceback 

12import asyncio 

13import aiofiles 

14import aiofiles.os 

15import mimetypes 

16 

17from io import BytesIO 

18 

19from abc import ABC 

20from abc import abstractmethod 

21 

22from typing import Optional 

23from typing import Dict 

24from typing import Any 

25from typing import List 

26from typing import Tuple 

27from typing import Type 

28from typing import Union 

29from typing import cast 

30 

31from http import HTTPStatus 

32 

33from datetime import datetime 

34 

35from quart import Quart 

36from quart import request 

37from quart import jsonify 

38from quart import render_template 

39from quart import send_file 

40from quart import send_from_directory 

41from quart import Response 

42 

43from jinja2 import ChoiceLoader 

44from jinja2 import FileSystemLoader 

45 

46from hypercorn.config import Config 

47from hypercorn.asyncio import serve 

48 

49from streamwise_job import StreamWiseJob 

50from streamwise_job import JobStatus 

51from streamwise_job import get_job_id 

52from streamwise_job import is_job_id 

53from streamwise_job import is_status_terminal 

54from streamwise_job import is_status_expired 

55from lmm_service_manager import LMMServiceManager 

56 

57from resolutions import ASPECT_RATIO 

58from resolutions import RESOLUTIONS 

59 

60# Local relative imports 

61sys.path.append("..") # noqa: E402 

62sys.path.append("../..") # noqa: E402 

63 

64from console_utils import setup_logging 

65 

66from media_utils import get_video_file_info 

67from media_utils import get_audio_file_info 

68from media_utils import get_tensor_file_info 

69from media_utils import get_image_file_info 

70from media_utils import get_text_file_info 

71 

72import quart_utils 

73import k8s_utils 

74 

75from quart_utils import QuartReturn 

76 

77from quart_utils import format_string 

78from quart_utils import format_duration 

79from quart_utils import format_duration_short 

80from quart_utils import parse_request_id 

81from quart_utils import get_content_type_emoji 

82from quart_utils import get_file_type_emoji 

83from quart_utils import json_pretty_filter 

84from quart_utils import get_file_type 

85from quart_utils import get_mime_type 

86from quart_utils import get_aspect_ratio 

87from quart_utils import format_bytes 

88from quart_utils import get_friendly_container_name 

89from quart_utils import get_class_emoji 

90 

91from tts_utils import generate_waveform_plt 

92 

93HOST = "0.0.0.0" 

94PORT = 18080 

95 

96K8S_CLUSTER = "incluster" 

97 

98 

99def status_history_to_times(status_history: Dict[float, str]) -> Dict[str, datetime]: 

100 """Convert status history timestamps to datetime objects.""" 

101 times: Dict[str, datetime] = {} 

102 for timestamp_float, status_str in status_history.items(): 

103 if status_str == "COMPLETED" or status_str == "EXPIRED": 

104 times[status_str] = datetime.fromtimestamp(timestamp_float) 

105 elif status_str not in times: 

106 times[status_str] = datetime.fromtimestamp(timestamp_float) 

107 return times 

108 

109 

110class StreamWiseApp(ABC): 

111 """Generic StreamWise Quart application.""" 

112 

113 def __init__( 

114 self, 

115 app_name: str = "streamwise" 

116 ) -> None: 

117 self.app_name = app_name 

118 

119 self.tmp_dir = f"/tmp/{self.app_name}" 

120 self.log_file_name = f"{self.app_name}.log" 

121 

122 self.file_manager = StreamWiseAppFileManager(self.tmp_dir) 

123 

124 self.app = Quart(__name__) 

125 self.register_templates() 

126 self.register_filters() 

127 self.register_routes() 

128 

129 self.args: Optional[argparse.Namespace] = None 

130 self.jobs: Dict[str, StreamWiseJob] = {} 

131 self.service_manager: Optional[LMMServiceManager] = None 

132 

133 @abstractmethod 

134 def create_job( 

135 self, 

136 job_id: str, 

137 job_config: Dict[str, Any] 

138 ) -> StreamWiseJob: 

139 raise NotImplementedError("Subclasses must implement create_job.") 

140 

141 async def submit_job_handler(self) -> QuartReturn: 

142 """Handle job submission.""" 

143 try: 

144 job_id, job_dir, job_config = await self.prepare_submit_job() 

145 

146 # Create the job 

147 job = self.create_job(job_id, job_config) 

148 self.jobs[job_id] = job 

149 

150 # Create an async task to handle the job processing in the background 

151 task = asyncio.create_task(job.generate(job_config)) 

152 job.task = task 

153 

154 # Wait a bit to catch immediate exceptions 

155 await asyncio.sleep(0.1) 

156 if task.done() and task.exception(): 

157 ex = task.exception() 

158 assert ex is not None 

159 app_ex = cast(Exception, ex) 

160 return { 

161 "status": "error", 

162 "error": str(app_ex), 

163 }, self.get_http_status_from_exception(app_ex) 

164 

165 return { 

166 "status": "success", 

167 "job_id": job_id, 

168 } 

169 except ValueError as value_err: 

170 logging.error(f"Value error: {value_err}") 

171 return jsonify({ 

172 "status": "error", 

173 "error": str(value_err), 

174 "traceback": traceback.format_exc() 

175 }), HTTPStatus.BAD_REQUEST 

176 except Exception as ex: 

177 logging.error(f"Error: {ex}") 

178 logging.error(f"Traceback: {traceback.format_exc()}") 

179 return jsonify({ 

180 "status": "error", 

181 "error": str(ex), 

182 "traceback": traceback.format_exc() 

183 }), self.get_http_status_from_exception(ex) 

184 

185 def register_templates(self) -> None: 

186 self.app.jinja_env.loader = ChoiceLoader([ 

187 FileSystemLoader("templates"), 

188 FileSystemLoader("apps/templates"), 

189 FileSystemLoader(f"{self.app_name}/templates"), 

190 FileSystemLoader(f"apps/{self.app_name}/templates"), 

191 FileSystemLoader("../templates"), 

192 ]) 

193 

194 def parse_arguments( 

195 self, 

196 description: str = "StreamWise" 

197 ) -> argparse.Namespace: 

198 parser = argparse.ArgumentParser(description=description) 

199 parser.add_argument("--k8s_cluster", type=str, default=K8S_CLUSTER, help="Kubernetes cluster context name") 

200 parser.add_argument("--host", type=str, default=HOST, help="Host to bind the server to") 

201 parser.add_argument("--port", type=int, default=PORT, help="Port to bind the server to") 

202 parser.add_argument("--certfile", type=str, default=None, help="Path to SSL certificate file for HTTPS") 

203 parser.add_argument("--keyfile", type=str, default=None, help="Path to SSL private key file for HTTPS") 

204 parser.add_argument("--use-https", action="store_true", default=False, 

205 help="Use HTTPS for outbound service connections (health checks, job submissions)") 

206 return parser.parse_args() 

207 

208 def get_http_status_from_exception( 

209 self, 

210 ex: Exception 

211 ) -> HTTPStatus: 

212 """Map exceptions to HTTP status codes.""" 

213 if isinstance(ex, ValueError): 

214 return HTTPStatus.BAD_REQUEST 

215 return HTTPStatus.INTERNAL_SERVER_ERROR 

216 

217 async def main( 

218 self, 

219 args: Any 

220 ) -> None: 

221 """Main entry point for StreamWise application.""" 

222 if args.use_https: 

223 k8s_utils.set_service_scheme("https") 

224 k8s_utils.set_verify_ssl(False) 

225 

226 scheme = "https" if args.certfile else "http" 

227 logging.info( 

228 f"Starting {self.app_name} app on {scheme}://{args.host}:{args.port} " 

229 f"with K8S cluster '{args.k8s_cluster}'" 

230 ) 

231 

232 self.service_manager = LMMServiceManager( 

233 app_name=self.app_name, 

234 k8s_cluster=args.k8s_cluster 

235 ) 

236 await self.service_manager.init_k8s_services() 

237 service_manager_task = asyncio.create_task( 

238 self.service_manager.start_updater()) 

239 

240 await self.service_manager.warmup_services() 

241 

242 try: 

243 http_task = asyncio.create_task(self.run_httpserver( 

244 host=args.host, 

245 port=args.port, 

246 certfile=args.certfile, 

247 keyfile=args.keyfile, 

248 )) 

249 await http_task 

250 except OSError as os_err: 

251 if os_err.errno == errno.EADDRINUSE: 

252 logging.error(f"{args.host}:{args.port} already in use.") 

253 else: 

254 logging.error(f"OS error: {os_err}") 

255 except Exception as ex: 

256 logging.error(f"Error: {ex}") 

257 logging.error(traceback.format_exc()) 

258 finally: 

259 await self.service_manager.stop() 

260 if service_manager_task and not service_manager_task.done(): 

261 service_manager_task.cancel() 

262 await service_manager_task 

263 

264 async def run_httpserver( 

265 self, 

266 host: str = HOST, 

267 port: int = PORT, 

268 certfile: Optional[str] = None, 

269 keyfile: Optional[str] = None, 

270 ) -> None: 

271 """HTTP/HTTPS server runs in the main process.""" 

272 config = Config() 

273 config.bind = [f"{host}:{port}"] 

274 

275 # Display the access logs for debugging 

276 config.accesslog = "-" 

277 

278 # Increase max request body size to 128 MB (default is 16 MB) 

279 config.wsgi_max_body_size = 128 * 1024 * 1024 

280 config.limit_max_request_size = 128 * 1024 * 1024 # type: ignore[attr-defined] 

281 self.app.config["MAX_CONTENT_LENGTH"] = 128 * 1024 * 1024 

282 

283 if certfile: 

284 config.certfile = certfile 

285 if keyfile: 

286 config.keyfile = keyfile 

287 

288 await serve(self.app, config) 

289 

290 def register_filters(self) -> None: 

291 """Register template filters.""" 

292 app = self.app 

293 app.template_filter("format_string")(lambda s: format_string(s)) 

294 app.template_filter("format_duration")(lambda s: format_duration(s)) 

295 app.template_filter("format_duration_short")(lambda s: format_duration_short(s)) 

296 app.template_filter("parse_request_id")(parse_request_id) 

297 app.template_filter("get_content_type_emoji")(get_content_type_emoji) 

298 app.template_filter("get_file_type_emoji")(lambda f: get_file_type_emoji(get_file_type(f))) 

299 app.template_filter("json_pretty")(json_pretty_filter) 

300 app.template_filter("get_file_type")(get_file_type) 

301 app.template_filter("get_mime_type")(get_mime_type) 

302 app.template_filter("get_aspect_ratio")(get_aspect_ratio) 

303 app.template_filter("format_bytes")(format_bytes) 

304 

305 app.template_filter("get_friendly_container_name")(get_friendly_container_name) 

306 app.template_filter("get_class_emoji")(get_class_emoji) 

307 

308 async def prepare_submit_job(self) -> Tuple[str, str, Dict[str, Any]]: 

309 """Prepare for job submission.""" 

310 request_json = await request.get_json() 

311 if not request_json: 

312 raise ValueError("No JSON body received") 

313 

314 job_id = request_json.get("job_id") or get_job_id() 

315 logging.info(f"Received job_id {job_id}.") 

316 

317 job_dir = f"{self.tmp_dir}/{job_id}" 

318 await aiofiles.os.makedirs(job_dir, exist_ok=True) 

319 

320 async with aiofiles.open(f"{job_dir}/request.json", "w") as file: 

321 await file.write(json.dumps(request_json, indent=4)) 

322 

323 job_config = self.get_job_config_from_request(job_id, request_json) 

324 

325 async with aiofiles.open(f"{job_dir}/config.json", "w") as file: 

326 await file.write(json.dumps(job_config, indent=4)) 

327 

328 if self.service_manager is None: 

329 raise ValueError("Service manager not initialized") 

330 

331 return job_id, job_dir, job_config 

332 

333 def get_job_config_from_request( 

334 self, 

335 job_id: str, 

336 request_json: Dict[str, Any] 

337 ) -> Dict[str, Any]: 

338 """Process and return the job configuration from the request JSON.""" 

339 ret = request_json.copy() 

340 ret["job_id"] = job_id 

341 

342 if "resolution" in ret: 

343 resolution_str = ret["resolution"] 

344 resolution_str = resolution_str.lower() 

345 aspect_ratio = ASPECT_RATIO 

346 width, height = RESOLUTIONS[aspect_ratio][resolution_str] 

347 ret["width"] = width 

348 ret["height"] = height 

349 

350 return ret 

351 

352 async def get_logs( 

353 self, 

354 path: Optional[str] = None, 

355 file_name: Optional[str] = None 

356 ) -> str: 

357 """Get the application logs.""" 

358 if path is None: 

359 path = self.tmp_dir 

360 if file_name is None: 

361 file_name = self.log_file_name 

362 logs = "" 

363 if not await aiofiles.os.path.exists(f"{path}/{file_name}"): 

364 logging.warning(f"Log file {path}/{file_name} does not exist.") 

365 return logs 

366 async with aiofiles.open(f"{path}/{file_name}", "r") as file: 

367 logs = await file.read() 

368 return logs 

369 

370 async def get_job_config(self, job_id: str) -> Dict[str, Any]: 

371 """Get the job configuration from the job directory.""" 

372 job_dir = f"{self.tmp_dir}/{job_id}" 

373 config_file = f"{job_dir}/config.json" 

374 if not await aiofiles.os.path.exists(config_file): 

375 logging.warning(f"Job config file {config_file} does not exist.") 

376 return {} 

377 try: 

378 async with aiofiles.open(config_file, "r") as file: 

379 content = await file.read() 

380 config_json = json.loads(content) 

381 return config_json 

382 except Exception as ex: 

383 logging.error(f"Error reading job config: {ex}") 

384 return {} 

385 

386 async def get_job_status( 

387 self, 

388 job_id: str 

389 ) -> Dict[str, JobStatus]: 

390 """Get the status of a job asynchronously.""" 

391 job_dir = f"{self.tmp_dir}/{job_id}" 

392 status_file = f"{job_dir}/status.txt" 

393 if not await aiofiles.os.path.exists(status_file): 

394 return {"status": JobStatus.UNKNOWN.name} 

395 

396 # Get last modified time of the status file 

397 last_modified_time = await aiofiles.os.path.getmtime(status_file) 

398 async with aiofiles.open(status_file, "r") as file: 

399 content = await file.read() 

400 contet_strip = content.strip() 

401 if not contet_strip: 

402 return {"status": JobStatus.UNKNOWN.name} 

403 line = contet_strip.splitlines()[-1].strip() 

404 line_split = line.split(",") 

405 if len(line_split) == 2: 

406 timestamp_str, status_str = line_split 

407 else: 

408 status_str = line_split[0] 

409 status_str = status_str.strip() 

410 if not status_str.isdigit(): 

411 return {"status": JobStatus.UNKNOWN.name} 

412 status_val = int(status_str) 

413 status = JobStatus(status_val) 

414 if not is_status_terminal(status) and is_status_expired(last_modified_time): 

415 return {"status": JobStatus.EXPIRED.name} 

416 return {"status": status.name} 

417 return {"status": JobStatus.UNKNOWN.name} 

418 

419 async def get_services(self) -> Dict[str, List[Dict[str, Any]]]: 

420 """Get the list of available services.""" 

421 ret: Dict[str, List[Dict[str, Any]]] = {"services": []} 

422 if self.service_manager is None: 

423 logging.error("Service manager not initialized") 

424 return ret 

425 for service_name, service in self.service_manager.services.items(): 

426 for container in service.containers: 

427 container_status = { 

428 "service_name": service_name, 

429 "pod_name": service_name, 

430 "container_name": container.name, 

431 "ip": container.ip, 

432 "port": container.port, 

433 "status": container.status, 

434 "busy": container.busy, 

435 } 

436 ret["services"].append(container_status) 

437 return ret 

438 

439 async def get_jobs(self) -> Dict[str, List[Dict[str, Any]]]: 

440 """Get the list of jobs.""" 

441 ret: Dict[str, List[Dict[str, Any]]] = {"jobs": []} 

442 

443 if not await aiofiles.os.path.exists(self.tmp_dir): 

444 logging.warning(f"{self.tmp_dir} directory does not exist") 

445 return ret 

446 

447 file_names = await aiofiles.os.listdir(self.tmp_dir) 

448 job_ids = [] 

449 for file_name in file_names: 

450 if is_job_id(file_name): 

451 job_ids.append(file_name) 

452 

453 async def get_job_details(job_id: str) -> Optional[Dict[str, Any]]: 

454 status = await self.get_job_status(job_id) 

455 status_str = status.get("status", "unknown") 

456 job_details = { 

457 "job_id": job_id, 

458 "status": status_str, 

459 } 

460 

461 job_config = await self.get_job_config(job_id) 

462 if job_config: 

463 job_details.update(job_config) 

464 

465 file_path = f"{self.tmp_dir}/{job_id}/{job_id}.mp4" 

466 if await aiofiles.os.path.exists(file_path): 

467 loop = asyncio.get_running_loop() 

468 file_video_info = await loop.run_in_executor(None, get_video_file_info, file_path) 

469 job_details.update(file_video_info) 

470 return job_details 

471 

472 jobs = await asyncio.gather(*(get_job_details(job_id) for job_id in job_ids)) 

473 ret["jobs"] = [job for job in jobs if job is not None] 

474 return ret 

475 

476 def register_routes(self) -> None: 

477 """Register HTTP routes.""" 

478 route = self.app.route 

479 

480 @route("/", methods=["GET"]) 

481 async def index() -> str: 

482 """Render the index HTML page.""" 

483 services = await self.get_services() 

484 jobs = await self.get_jobs() 

485 logs = { 

486 "app": await self.get_logs(), 

487 "service_manager": await self.get_logs(file_name="service_manager.log"), 

488 } 

489 return await render_template( 

490 "index.html", 

491 services=services["services"], 

492 jobs=jobs["jobs"], 

493 logs=logs, 

494 ) 

495 

496 @route("/health", methods=["GET"]) 

497 async def health() -> QuartReturn: 

498 """Get health status.""" 

499 health: Dict[str, Any] = { 

500 "status": "ok", 

501 "host": self.args.host if self.args else None, 

502 "port": self.args.port if self.args else None, 

503 "k8s_cluster": self.args.k8s_cluster if self.args else None, 

504 "jobs": {}, 

505 "services": {} 

506 } 

507 if self.jobs: 

508 for job_id, job in self.jobs.items(): 

509 job_status = await job.get_status() 

510 requests = job.get_requests() 

511 health["jobs"][job_id] = { 

512 "job_id": job.job_id, 

513 "status": job_status.name, 

514 "num_requests": len(requests), 

515 } 

516 if self.service_manager: 

517 for service_name, service in self.service_manager.services.items(): 

518 health["services"][service_name] = { 

519 "name": service.name, 

520 "num_containers": len(service.containers), 

521 } 

522 return jsonify(health), HTTPStatus.OK 

523 

524 @route("/api/services", methods=["GET"]) 

525 async def api_get_services() -> Dict[str, List[Dict[str, Any]]]: 

526 """Get the list of available services.""" 

527 return await self.get_services() 

528 

529 @route("/files", methods=["GET"]) 

530 async def list_files() -> QuartReturn: 

531 return await self.file_manager.list_files() 

532 

533 @route("/file/<file_name>", methods=["GET"]) 

534 async def download_file(file_name: str) -> QuartReturn: 

535 return await self.file_manager.download_file(file_name) 

536 

537 @route("/file_stream/<job_id>/<file_name>", methods=["GET"]) 

538 async def file_stream(job_id: str, file_name: str) -> QuartReturn: 

539 return await self.file_manager.stream(job_id, file_name) 

540 

541 @route("/file_view/<job_id>/<file_name>", methods=["GET"]) 

542 async def file_view(job_id: str, file_name: str) -> QuartReturn: 

543 return await self.file_manager.view(job_id, file_name) 

544 

545 @route("/audio_waveform/<job_id>/<file_name>", methods=["GET"]) 

546 async def get_audio_waveform(job_id: str, file_name: str) -> QuartReturn: 

547 """Generate and return a waveform PNG image for a WAV audio file.""" 

548 if not job_id or not file_name: 

549 return jsonify({"error": "Job id and file name are required"}), HTTPStatus.BAD_REQUEST 

550 if not file_name or not file_name.endswith((".wav")): 

551 return jsonify({"error": f"Invalid file name: {file_name}"}), HTTPStatus.BAD_REQUEST 

552 

553 # Download the WAV file into a local temp file 

554 job_dir = f"{self.tmp_dir}/{job_id}" 

555 wav_path = f"{job_dir}/{file_name}" 

556 waveform_png_path = generate_waveform_plt(wav_path) 

557 

558 return await send_file( 

559 waveform_png_path, 

560 as_attachment=True, 

561 attachment_filename=f"waveform_{file_name}.png", 

562 mimetype="image/png") 

563 

564 @route("/job", methods=["GET"]) 

565 async def submit_job() -> str: 

566 """Render the submit job HTML page.""" 

567 return await render_template("submit_job.html") 

568 

569 @route("/api/jobs", methods=["GET"]) 

570 async def api_get_jobs() -> Dict[str, List[Dict[str, Any]]]: 

571 """Get the list of jobs.""" 

572 return await self.get_jobs() 

573 

574 @route("/api/job/<job_id>/status", methods=["GET"]) 

575 async def api_get_job_status(job_id: str) -> Dict[str, JobStatus]: 

576 """Get the status of a job.""" 

577 return await self.get_job_status(job_id) 

578 

579 @route("/api/job", methods=["POST"]) 

580 async def api_submit_job() -> QuartReturn: 

581 """Submit a new job.""" 

582 return await self.submit_job_handler() 

583 

584 @route("/api/job/<job_id>/<json_file_name>", methods=["GET"]) 

585 async def api_get_job_json_file(job_id: str, json_file_name: str) -> QuartReturn: 

586 """Get a JSON or JSONL file from the job directory.""" 

587 job_dir = f"{self.tmp_dir}/{job_id}" 

588 if json_file_name.endswith(".jsonl"): 

589 # JSONL files are served as raw newline-delimited text without appending .json 

590 request_file = f"{job_dir}/{json_file_name}" 

591 if not await aiofiles.os.path.exists(request_file): 

592 return { 

593 "status": "error", 

594 "error": f"Job {job_id} file {json_file_name} not found" 

595 } 

596 async with aiofiles.open(request_file, "r") as file: 

597 content = await file.read() 

598 return Response(content, mimetype="application/x-ndjson") 

599 request_file = f"{job_dir}/{json_file_name}.json" 

600 if not await aiofiles.os.path.exists(request_file): 

601 return { 

602 "status": "error", 

603 "error": f"Job {job_id} JSON file {json_file_name}.json not found" 

604 } 

605 try: 

606 async with aiofiles.open(request_file, "r") as file: 

607 content = await file.read() 

608 request_json = json.loads(content) 

609 return request_json 

610 except Exception as ex: 

611 logging.error(f"Error reading job config: {ex}") 

612 return { 

613 "status": "error", 

614 "error": str(ex), 

615 "traceback": traceback.format_exc() 

616 } 

617 

618 @route("/api/job/<job_id>/config", methods=["GET"]) 

619 async def api_get_job_config(job_id: str) -> QuartReturn: 

620 """Get the job configuration.""" 

621 return await api_get_job_json_file(job_id, "config") 

622 

623 @route("/api/job/<job_id>/request", methods=["GET"]) 

624 async def api_get_job_request(job_id: str) -> QuartReturn: 

625 """Get the job request JSON.""" 

626 return await api_get_job_json_file(job_id, "request") 

627 

628 @route("/job/<job_id>", methods=["GET"]) 

629 async def job_status(job_id: str) -> str: 

630 """Render the job status HTML page for a specific job ID.""" 

631 files = await self.file_manager.list_job_files(job_id) 

632 job_request = await api_get_job_request(job_id) 

633 job_config = await api_get_job_config(job_id) 

634 

635 status = await self.get_job_status(job_id) 

636 status = status.get("status", "unknown") 

637 status_history = await get_job_status_history(job_id) 

638 times = status_history_to_times(status_history) 

639 

640 job = self.jobs.get(job_id, None) 

641 requests = {} 

642 if job is not None: 

643 requests = job.get_requests().copy() 

644 # Convert deadline from seconds to milliseconds 

645 for req in requests.values(): 

646 if "deadline" in req: 

647 req["deadline"] = req["deadline"] * 1000 # ms 

648 return await render_template( 

649 "job.html", 

650 job_id=job_id, 

651 job=job, 

652 job_request=job_request, 

653 job_config=job_config, 

654 requests=requests, 

655 status=status, 

656 times=times, 

657 files=files) 

658 

659 @route("/api/job/<job_id>/status/history", methods=["GET"]) # type: ignore[type-var] 

660 async def get_job_status_history(job_id: str) -> Dict[float, str]: 

661 """Get the status of a job asynchronously.""" 

662 job_dir = f"{self.tmp_dir}/{job_id}" 

663 status_file = f"{job_dir}/status.txt" 

664 ret: Dict[float, str] = {} 

665 if not await aiofiles.os.path.exists(status_file): 

666 return ret 

667 

668 async with aiofiles.open(status_file, "r") as file: 

669 content = await file.read() 

670 contet_strip = content.strip() 

671 

672 for line in contet_strip.splitlines(): 

673 line_split = line.split(",") 

674 if len(line_split) == 2: 

675 timestamp_str, status_str = line_split 

676 else: 

677 timestamp_str = "-1.0" 

678 status_str = line_split[0] 

679 status_str = status_str.strip() 

680 if not status_str.isdigit(): 

681 continue 

682 status_val = int(status_str) 

683 status = JobStatus(status_val) 

684 timestamp_float = float(timestamp_str) 

685 ret[timestamp_float] = status.name 

686 return ret 

687 

688 @route("/api/job/<job_id>/requests", methods=["GET"]) 

689 async def api_get_job_requests(job_id: str) -> Dict[str, Any]: 

690 job = self.jobs.get(job_id, None) 

691 if not job: 

692 return {} 

693 requests = job.get_requests() 

694 ret = {} 

695 for request_id, req in requests.items(): 

696 ret[request_id] = req.json() 

697 return ret 

698 

699 

700class StreamWiseAppFileManager: 

701 """Manage files for the StreamWise application.""" 

702 

703 def __init__(self, tmp_dir: str): 

704 self.tmp_dir = tmp_dir 

705 

706 async def list_job_files( 

707 self, 

708 job_id: str 

709 ) -> List[Dict[str, Any]]: 

710 """Asynchronously list files in the job directory.""" 

711 job_dir = f"{self.tmp_dir}/{job_id}" 

712 if not await aiofiles.os.path.exists(job_dir): 

713 return [] 

714 files = [] 

715 file_names = await aiofiles.os.listdir(job_dir) 

716 for file_name in file_names: 

717 file_path = os.path.join(job_dir, file_name) 

718 file_date = await aiofiles.os.path.getmtime(file_path) 

719 file_size = await aiofiles.os.path.getsize(file_path) 

720 # file_type = get_file_type(file_name) 

721 mime_type, _ = mimetypes.guess_type(file_name) 

722 files.append({ 

723 "name": file_name, 

724 "size": file_size, 

725 "date": datetime.fromtimestamp(file_date), 

726 "mimetype": mime_type, 

727 }) 

728 return files 

729