[FXC-5593] feat(): VGPU access.

flow360/cloud/flow360_requests.py

@@ -238,6 +238,7 @@ class DraftRunRequest(Flow360RequestsV2):
     source_item_type: Literal["Geometry", "SurfaceMesh", "VolumeMesh", "Case"] = pd.Field(
         exclude=True
     )
+    job_type: Optional[Literal["TIME_SHARED_VGPU", "FLEX_CREDIT"]] = pd.Field(None)
 
     @pd_v2.model_validator(mode="after")
     def _validate_force_creation_config(self):

flow360/component/project.py

@@ -14,6 +14,7 @@
 from pydantic import PositiveInt
 
 from flow360.cloud.flow360_requests import LengthUnitType, RenameAssetRequestV2
+from flow360.cloud.http_util import http
 from flow360.cloud.rest_api import RestApi
 from flow360.component.case import Case
 from flow360.component.cloud_examples import (
@@ -1822,6 +1823,7 @@ def _run(
         raise_on_error: bool,
         tags: List[str],
         draft_only: bool,
+        job_type: Optional[Literal["TIME_SHARED_VGPU", "FLEX_CREDIT"]] = None,
         **kwargs,
     ):
         """
@@ -1851,6 +1853,8 @@ def _run(
             A list of tags to add to the target asset.
         draft_only: bool, optional
             Whether to only create and submit a draft and not run the simulation.
+        job_type: Optional[Literal["TIME_SHARED_VGPU", "FLEX_CREDIT"]]
+            The billing job type to use for the run request.
 
         Returns
         -------
@@ -1952,6 +1956,7 @@ def _run(
                 use_beta_mesher=params.private_attribute_asset_cache.use_inhouse_mesher,
                 use_geometry_AI=use_geometry_AI,
                 start_from=start_from,
+                job_type=job_type,
             )
         except RuntimeError:
             if raise_on_error:
@@ -2148,6 +2153,7 @@ def run_case(
         raise_on_error: bool = True,
         tags: List[str] = None,
         draft_only: bool = False,
+        billing_method: Optional[Literal["VirtualGPU", "FlexCredit"]] = None,
         **kwargs,
     ):
         """
@@ -2177,6 +2183,8 @@ def run_case(
             A list of tags to add to the case.
         draft_only: bool, optional
             Whether to only create and submit a draft and not run the case.
+        billing_method: Optional[Literal["VirtualGPU", "FlexCredit"]]
+            Override to default billing method.
 
         Returns
         -------
@@ -2188,6 +2196,29 @@ def run_case(
             raise Flow360ConfigError(
                 "Interpolation to mesh is only supported when forking from a case."
             )
+        # Map user-facing billing_method to API-level job_type
+        job_type = None
+        if billing_method is not None and not draft_only:
+            if billing_method == "FlexCredit":
+                log.info("The case will be submitted to regular queue and billed with FlexCredits.")
+                job_type = "FLEX_CREDIT"
+            elif billing_method == "VirtualGPU":
+                account_info = http.get("flow360/account")
+                if not account_info.get("timeSharedVGpuEnabled", False):
+                    raise Flow360ValueError(
+                        "Virtual GPU billing is not enabled for this account. "
+                        "Please contact support to enable Virtual GPU access."
+                    )
+                log.info(
+                    "The case will be submitted to Virtual GPU "
+                    "and billed with Virtual GPU allocation."
+                )
+                job_type = "TIME_SHARED_VGPU"
+        elif draft_only and billing_method is not None:
+            log.info(
+                "`billing_method` input to `run_case()` ignored since"
+                " no billing is necessary when submitting just the draft."
+            )
 
         self._check_initialized()
         case_or_draft = self._run(
@@ -2203,6 +2234,7 @@ def run_case(
             raise_on_error=raise_on_error,
             tags=tags,
             draft_only=draft_only,
+            job_type=job_type,
             **kwargs,
         )
 

flow360/component/project_utils.py

@@ -716,7 +716,7 @@ def set_up_params_for_uploading(  # pylint: disable=too-many-arguments
     return params
 
 
-def validate_params_with_context(params, root_item_type, up_to):
+def validate_params_with_context(params: SimulationParams, root_item_type, up_to):
     """Validate the simulation params with the simulation path."""
 
     # pylint: disable=protected-access

flow360/component/simulation/web/draft.py

@@ -5,7 +5,7 @@
 import ast
 import json
 from functools import cached_property
-from typing import Literal, Union
+from typing import TYPE_CHECKING, Literal, Optional, Union
 
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -26,6 +26,9 @@
 from flow360.exceptions import Flow360RuntimeError, Flow360WebError
 from flow360.log import log
 
+if TYPE_CHECKING:
+    from flow360.component.simulation.simulation_params import SimulationParams
+
 
 class DraftMetaModel(BaseModel):
     """Draft metadata deserializer"""
@@ -129,7 +132,7 @@ def from_cloud(cls, draft_id: IDStringType) -> Draft:
         """Load draft from cloud"""
         return Draft(draft_id=draft_id)
 
-    def update_simulation_params(self, params):
+    def update_simulation_params(self, params: SimulationParams):
         """update the SimulationParams of the draft"""
         params_dict = params.model_dump(mode="json", exclude_none=True)
         params_dict = collect_and_tokenize_selectors_in_place(params_dict)
@@ -175,6 +178,7 @@ def run_up_to_target_asset(
         use_geometry_AI: bool,  # pylint: disable=invalid-name
         source_item_type: Literal["Geometry", "SurfaceMesh", "VolumeMesh", "Case"],
         start_from: Union[None, Literal["SurfaceMesh", "VolumeMesh", "Case"]],
+        job_type: Optional[Literal["TIME_SHARED_VGPU", "FLEX_CREDIT"]] = None,
     ) -> str:
         """run the draft up to the target asset"""
 
@@ -195,15 +199,20 @@ def run_up_to_target_asset(
             force_creation_config = (
                 ForceCreationConfig(start_from=start_from) if start_from else None
             )
+
             run_request = DraftRunRequest(
                 source_item_type=source_item_type,
                 up_to=target_asset._cloud_resource_type_name,
                 use_in_house=use_beta_mesher,
                 use_gai=use_geometry_AI,
                 force_creation_config=force_creation_config,
+                job_type=job_type,
             )
+            request_body = run_request.model_dump(by_alias=True)
+            if request_body.get("jobType") is None:
+                request_body.pop("jobType", None)
             run_response = self.post(
-                run_request.model_dump(by_alias=True),
+                request_body,
                 method="run",
             )
             destination_id = run_response["id"]