# SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved. # SPDX-License-Identifier: Apache-2.0 # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from contextlib import asynccontextmanager from dataclasses import dataclass import logging import os import pickle import traceback from fastapi import FastAPI from fastapi.requests import Request from fastapi.responses import Response import imageio.v3 as imageio import numpy as np from api_types import CompressedSeedingRequest from server_base import InferenceModel from server_cosmos import CosmosModel # ------------------------------ @dataclass class ServerSettings(): """ Note: we use a dataclass + env variables because we can't easily pass command line arguments through the `fastapi` launcher. """ model: str = os.environ.get("GEN3C_MODEL", "cosmos-predict1") checkpoint_path: str | None = os.environ.get("GEN3C_CKPT_PATH") data_path: str | None = os.environ.get("GEN3C_DATA_PATH") #: Additional latency to add to any inference request, in milliseconds. inference_latency: int = int(os.environ.get("GEN3C_INFERENCE_LATENCY", 0)) #: Number of inference results to keep in cache. #: This may be useful when multiple requests are in flight and the user hasn't #: retrieved the results yet. inference_cache_size: int = int(os.environ.get("GEN3C_INFERENCE_CACHE_SIZE", 15)) #: Number of GPUs to use for inference. Leave at 0 to automatically select #: based on available hardware. gpu_count: int = int(os.environ.get("GEN3C_GPU_COUNT", 0)) settings = ServerSettings() model: InferenceModel | None = None @asynccontextmanager async def lifespan(app: FastAPI): global model model_name = settings.model.lower() if model_name in ("cosmos", "cosmos-predict1"): cls = CosmosModel else: raise ValueError(f"Unsupported model type: '{settings.model}'") model = cls(checkpoint_path=settings.checkpoint_path, data_path=settings.data_path, fake_delay_ms=settings.inference_latency, inference_cache_size=settings.inference_cache_size, gpu_count=settings.gpu_count) # --- Startup code # Pre-render at least one image to make sure everything is running if not model.requires_seeding(): await model.make_test_image() yield # --- Shutdown code model.cleanup() del model app = FastAPI(lifespan=lifespan) logger = logging.getLogger('uvicorn.error') # ------------------------------ def get_bool_query_param(request: Request, name: str, default: bool) -> bool: b_str = request.query_params.get(name, "1" if default else "0") return b_str.lower() in ("1", "true", "yes", "") @app.post("/request-inference", response_class=Response, response_model=None) async def request_inference(request: Request): """ Start a new asynchronous inference job. """ sync = get_bool_query_param(request, "sync", default=False) req: bytes = await request.body() req = pickle.loads(req) try: if sync: result = await model.request_inference_sync(req) return Response(content=pickle.dumps(result), media_type="application/octet-stream") else: model.request_inference(req) except Exception as e: logging.error("Inference request failed with exception:" f"\n{e}\n{traceback.format_exc()}") return Response(str(e), status_code=400) return Response("Request accepted.", status_code=202) @app.post("/seed-model", response_class=Response, response_model=None) async def seed_model(request: Request): """ Start a new asynchronous inference job. """ sync = get_bool_query_param(request, "sync", default=False) req: bytes = await request.body() req = pickle.loads(req) if isinstance(req, CompressedSeedingRequest): req.decompress() try: # There isn't really anything async about the seeding request being done on the server # so far, so we just await. This could be changed in the future. result = await model.seed_model(req) except Exception as e: logging.error(f"Seeding request failed with exception:" f"\n{e}\n{traceback.format_exc()}") return Response(str(e), status_code=400) # return Response("Seeding request accepted.", status_code=(200 if sync else 202)) return Response(content=pickle.dumps(result), media_type="application/octet-stream") @app.get("/inference-result", response_class=Response, response_model=None) async def inference_results_or_none(request_id: str): try: result = model.inference_result_or_none(request_id) except Exception as e: # TODO: try to differentiate the status codes (doesn't exist, inference failed, etc) logging.error(f"Inference results request failed with exception:" f"\n{e}\n{traceback.format_exc()}") return Response(str(e), status_code=500) if result is None: return Response(content="Result not ready", status_code=503) else: return Response(content=pickle.dumps(result), media_type="application/octet-stream") @app.get("/image", response_class=Response) def latest_rgb(format: str = "jpg"): # We return the data as pickled bytes to avoid the JSON serialization / deserialization overhead. image = model.get_latest_rgb() if image is None: return Response(content="No image available yet.", status_code=404) if format == "pickle": content = pickle.dumps( { "image": image, } ) return Response(content=content, media_type="application/octet-stream") elif format in ("jpg", "png"): image = image.copy() # Allow alpha channel to be omitted for faster transfers if image.shape[-1] == 3: image = np.concatenate([ image, np.ones((*image.shape[:2], 1)) ], axis=-1) if image.dtype != np.uint8: # TODO: proper handling of gamma compression, etc image[:, :, :3] = np.power(image[:, :, :3], 1 / 2.2) * 255 image[:, :, 3] = image[:, :, 3] * 255 if format != "png": image = image[:, :, :3] content = imageio.imwrite(uri="", image=image.astype(np.uint8), extension="." + format) return Response(content=content, media_type=f"image/{format}") else: return Response(f"Unsupported image format: {format}", status_code=400) @app.get("/metadata") def metadata(): return model.metadata()