msk_admin/Lumina-MSK
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
DatTT127
2026-06-24 10:33:07 +07:00
parent 16a91bd17e
commit f705113711
77 changed files with 8999 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
workspace/sprint_1_2/CODEBASE/infra/implementation/cicd/jenkin_run.sh Normal file
View File

@@ -0,0 +1,31 @@
# Install Docker image
docker network create jenkins
# install docker-integratable image
docker run --name jenkins-docker --rm --detach \
-p 8080:8080 -p 50000:50000 \
--restart=on-failure \
--privileged --network jenkins --network-alias docker \
--env DOCKER_TLS_CERTDIR=/certs \
--volume jenkins-docker-certs:/certs/client \
--volume jenkins-data:/var/jenkins_home \
--publish 2376:2376 \
docker:dind --storage-driver overlay2 \
-v jenkins_home:/var/jenkins_home jenkins/jenkins:lts
# install the docker images
docker run \
--name jenkins-blueocean \
--restart=on-failure \
--detach \
--network jenkins \
--env DOCKER_HOST=tcp://docker:2376 \
--env DOCKER_CERT_PATH=/certs/client \
--env DOCKER_TLS_VERIFY=1 \
--publish 8080:8080 \
--publish 50000:50000 \
--volume jenkins-data:/var/jenkins_home \
--volume jenkins-docker-certs:/certs/client:ro \
jenkins/jenkins:lts

16
workspace/sprint_1_2/CODEBASE/infra/implementation/observability/docker-compose.yaml Normal file
View File

@@ -0,0 +1,16 @@
version: '3.8'
services:
prometheus:
image: prom/prometheus:latest
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml
ports:
- "9090:9090"
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin

12
workspace/sprint_1_2/CODEBASE/infra/implementation/observability/prometheus.yml Normal file
View File

@@ -0,0 +1,12 @@
global:
scrape_interval: 30s # Poll every 30 seconds instead of hammering it every 5s
scrape_timeout: 25s # Give it 25 full seconds to respond before timing out
scrape_configs:
- job_name: 'triton'
metrics_path: '/metrics'
scheme: 'https'
tls_config:
insecure_skip_verify: true
static_configs:
- targets: ['dtj-tran--triton-s3-service-unified-triton-server.modal.run']

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/angle_classify_convnext_tiny/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "angle_classify_convnext_tiny"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 4 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/angle_classify_densenet/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "angle_classify_densenet"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 4 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/angle_classify_efficientnet/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "angle_classify_efficientnet"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 4 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/angle_classify_resnet50/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "angle_classify_resnet50"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 4 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/angle_classify_swin_v2_s/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "angle_classify_swin_v2_s"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 4 ]
}
]

412
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/generate_ensemble.py Normal file
View File

@@ -0,0 +1,412 @@
#!/usr/bin/env python3
"""Generate a Triton ensemble model repository for the VKIST vision pipeline.
The VKIST design docs define the logical vision flow as:
1. Angle classification selects the scan view.
2. Inflammation detection checks whether synovitis/effusion is present.
3. Segmentation models produce anatomical masks for supported view branches.
The 11 resident Triton models in this repository all consume the same raw image
tensor (`input_image`) and return `logits`. Triton ensemble scheduling moves
those tensors through the ensemble graph internally, so this generator maps the
external client input once and exposes every component model's logits as a
terminal ensemble output.
"""
from __future__ import annotations
import argparse
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Iterable
ENSEMBLE_NAME = "my_vision_pipeline_ensemble"
MODEL_ROOT_DEFAULT = Path(__file__).resolve().parent
OUTPUT_DIR_DEFAULT = Path(__file__).resolve().parent / ENSEMBLE_NAME
VERSION_DIR = "1"
# Ordered by the architecture documents: angle classification -> inflammation
# detection -> segmentation. These are the 11 component models already resident
# under s3://vkist-ml-model/.
ANGLE_CLASSIFICATION_MODELS = [
"angle_classify_convnext_tiny",
"angle_classify_resnet50",
"angle_classify_swin_v2_s",
"angle_classify_densenet",
"angle_classify_efficientnet",
]
INFLAMMATION_MODELS = [
"inflammation_model_efficientnet_b0_ultrasound_2_cls",
]
SEGMENTATION_MODELS = [
"segmentation_model_unet_resnet101",
"segmentation_model_unet3plus_att",
"segmentation_model_post_deeplabv3_resnet101",
"segmentation_model_post_deeplabv3",
"segmentation_model_post_efficientfeedback",
]
ALL_MODEL_NAMES = [
*ANGLE_CLASSIFICATION_MODELS,
*INFLAMMATION_MODELS,
*SEGMENTATION_MODELS,
]
DEFAULT_INPUT_NAME = "input"
DEFAULT_INPUT_DATA_TYPE = "TYPE_UINT8"
DEFAULT_INPUT_DIMS = [-1, -1, -1, 3]
DEFAULT_MODEL_INPUT_NAME = "input_image"
DEFAULT_MODEL_OUTPUT_NAME = "logits"
DEFAULT_MODEL_OUTPUT_DATA_TYPE = "TYPE_FP32"
DEFAULT_MAX_BATCH_SIZE = 8
@dataclass(frozen=True)
class ModelConfig:
"""Parsed Triton config for one resident component model."""
name: str
platform: str
max_batch_size: int
input_name: str
input_data_type: str
input_dims: list[int]
output_name: str
output_data_type: str
output_dims: list[int]
@dataclass(frozen=True)
class EnsembleTensor:
"""One ensemble output and its matching internal model-output tensor."""
model_name: str
internal_name: str
output_name: str
data_type: str
dims: list[int]
def parse_int_list(text: str) -> list[int]:
"""Parse a Triton dims list such as '[ -1, 7, 512, 512 ]'."""
return [int(item) for item in re.findall(r"-?\d+", text)]
def parse_scalar(text: str, key: str, default: str | None = None) -> str | None:
"""Parse a quoted scalar from pbtxt text."""
match = re.search(rf"^\s*{re.escape(key)}:\s*\"([^\"]+)\"", text, flags=re.MULTILINE)
if match:
return match.group(1)
return default
def parse_block_fields(block_text: str) -> dict[str, str]:
"""Parse the simple scalar fields used by the existing model configs."""
fields: dict[str, str] = {}
for key in ("name", "platform", "data_type"):
value = parse_scalar(block_text, key)
if value is not None:
fields[key] = value
return fields
def parse_first_model_config(config_path: Path, fallback_name: str) -> ModelConfig:
"""Parse a component model config.pbtxt and fall back to known defaults."""
text = config_path.read_text(encoding="utf-8")
name = parse_scalar(text, "name", fallback_name) or fallback_name
platform = parse_scalar(text, "platform", "unknown") or "unknown"
max_batch_match = re.search(r"^\s*max_batch_size:\s*(-?\d+)", text, flags=re.MULTILINE)
max_batch_size = int(max_batch_match.group(1)) if max_batch_match else DEFAULT_MAX_BATCH_SIZE
input_match = re.search(r"input\s*\[(.*?)\]\s*output", text, flags=re.DOTALL)
output_match = re.search(r"output\s*\[(.*?)\]\s*$", text, flags=re.DOTALL)
input_fields = parse_block_fields(input_match.group(1)) if input_match else {}
output_fields = parse_block_fields(output_match.group(1)) if output_match else {}
input_dims_match = re.search(r"dims:\s*\[(.*?)\]", input_match.group(1), flags=re.DOTALL) if input_match else None
output_dims_match = re.search(r"dims:\s*\[(.*?)\]", output_match.group(1), flags=re.DOTALL) if output_match else None
input_dims = parse_int_list(input_dims_match.group(1)) if input_dims_match else DEFAULT_INPUT_DIMS
output_dims = parse_int_list(output_dims_match.group(1)) if output_dims_match else [4]
return ModelConfig(
name=name,
platform=platform,
max_batch_size=max_batch_size,
input_name=input_fields.get("name", DEFAULT_MODEL_INPUT_NAME),
input_data_type=input_fields.get("data_type", DEFAULT_INPUT_DATA_TYPE),
input_dims=input_dims,
output_name=output_fields.get("name", DEFAULT_MODEL_OUTPUT_NAME),
output_data_type=output_fields.get("data_type", DEFAULT_MODEL_OUTPUT_DATA_TYPE),
output_dims=output_dims,
)
def load_model_configs(model_root: Path, model_names: Iterable[str]) -> dict[str, ModelConfig]:
"""Load component configs from the local S3 mirror used by Triton."""
configs: dict[str, ModelConfig] = {}
for model_name in model_names:
config_path = model_root / model_name / "config.pbtxt"
configs[model_name] = parse_first_model_config(config_path, model_name)
return configs
def tensor_name_for_model(model_name: str) -> str:
"""Create a stable internal ensemble tensor name for one model."""
return f"{model_name}_logits"
def output_name_for_model(model_name: str) -> str:
"""Create the public ensemble output name for one component model."""
return model_name.upper()
def ensemble_output_dims(model_config: ModelConfig, max_batch_size: int) -> list[int]:
"""Return non-batch output dims for an ensemble output block.
Existing component configs include a leading `-1` output dim for dynamic
batching. Triton ensemble output blocks should declare only non-batch dims
when `max_batch_size` is positive.
"""
dims = list(model_config.output_dims)
if max_batch_size > 0 and dims and dims[0] == -1:
return dims[1:]
return dims
def format_dims(dims: Iterable[int]) -> str:
"""Format Triton dims as `[ 7, 512, 512 ]`."""
return "[ " + ", ".join(str(dim) for dim in dims) + " ]"
def build_ensemble_tensors(
model_configs: dict[str, ModelConfig],
max_batch_size: int,
) -> list[EnsembleTensor]:
"""Build ordered public outputs for the ensemble config."""
tensors: list[EnsembleTensor] = []
for model_name in ALL_MODEL_NAMES:
model_config = model_configs[model_name]
tensors.append(
EnsembleTensor(
model_name=model_name,
internal_name=tensor_name_for_model(model_name),
output_name=output_name_for_model(model_name),
data_type=model_config.output_data_type,
dims=ensemble_output_dims(model_config, max_batch_size),
)
)
return tensors
def format_input_block(input_name: str, input_data_type: str, input_dims: list[int]) -> str:
"""Render the ensemble input block."""
return f""" {{
name: "{input_name}"
data_type: {input_data_type}
dims: {format_dims(input_dims)}
}}"""
def format_output_block(tensor: EnsembleTensor) -> str:
"""Render one ensemble output block."""
return f""" {{
name: "{tensor.output_name}"
data_type: {tensor.data_type}
dims: {format_dims(tensor.dims)}
}}"""
def format_step_block(model_name: str, model_input_name: str, input_value: str, model_output_name: str, output_value: str) -> str:
"""Render one Triton ensemble_scheduling step."""
return f""" {{
model_name: "{model_name}"
model_version: -1
input_map {{
key: "{model_input_name}"
value: "{input_value}"
}}
output_map {{
key: "{model_output_name}"
value: "{output_value}"
}}
}}"""
def build_config_pbtxt(
ensemble_name: str,
max_batch_size: int,
input_name: str,
input_data_type: str,
input_dims: list[int],
tensors: list[EnsembleTensor],
model_configs: dict[str, ModelConfig],
) -> str:
"""Build the complete Triton ensemble config.pbtxt string."""
input_blocks = [format_input_block(input_name, input_data_type, input_dims)]
output_blocks = [format_output_block(tensor) for tensor in tensors]
steps: list[str] = []
for model_name in ALL_MODEL_NAMES:
model_config = model_configs[model_name]
tensor = next(item for item in tensors if item.model_name == model_name)
steps.append(
format_step_block(
model_name=model_name,
model_input_name=model_config.input_name,
input_value=input_name,
model_output_name=model_config.output_name,
output_value=tensor.internal_name,
)
)
sections = [
f"name: \"{ensemble_name}\"",
"platform: \"ensemble\"",
f"max_batch_size: {max_batch_size}",
"input [\n" + ",\n".join(input_blocks) + "\n]",
"output [\n" + ",\n".join(output_blocks) + "\n]",
"ensemble_scheduling {\n step [\n" + ",\n".join(steps) + "\n ]\n}",
"",
]
return "\n".join(sections)
def parse_dims_arg(value: str) -> list[int]:
"""Parse a comma-separated dims CLI value."""
return [int(item.strip()) for item in value.split(",") if item.strip()]
def prepare_output_dir(output_dir: Path, clean: bool) -> None:
"""Create or refresh the local Triton model repository directory."""
if clean and output_dir.exists():
import shutil
shutil.rmtree(output_dir)
version_dir = output_dir / VERSION_DIR
version_dir.mkdir(parents=True, exist_ok=True)
def generate_ensemble(
model_root: Path,
output_dir: Path,
ensemble_name: str,
max_batch_size: int,
input_name: str,
input_data_type: str,
input_dims: list[int],
clean: bool,
) -> Path:
"""Generate the ensemble repository and return the written config path."""
model_configs = load_model_configs(model_root, ALL_MODEL_NAMES)
tensors = build_ensemble_tensors(model_configs, max_batch_size)
config_text = build_config_pbtxt(
ensemble_name=ensemble_name,
max_batch_size=max_batch_size,
input_name=input_name,
input_data_type=input_data_type,
input_dims=input_dims,
tensors=tensors,
model_configs=model_configs,
)
prepare_output_dir(output_dir, clean=clean)
config_path = output_dir / VERSION_DIR / "config.pbtxt"
config_path.write_text(config_text, encoding="utf-8")
return config_path
def build_arg_parser() -> argparse.ArgumentParser:
"""Build CLI arguments for local generation."""
parser = argparse.ArgumentParser(
description="Generate Triton ensemble config for the VKIST 11-model vision pipeline."
)
parser.add_argument(
"--model-root",
type=Path,
default=MODEL_ROOT_DEFAULT,
help="Local directory containing the 11 resident model config.pbtxt files.",
)
parser.add_argument(
"--output-dir",
type=Path,
default=OUTPUT_DIR_DEFAULT,
help="Local Triton model repository directory to create.",
)
parser.add_argument(
"--ensemble-name",
default=ENSEMBLE_NAME,
help="Triton ensemble model name and S3 top-level folder name.",
)
parser.add_argument(
"--max-batch-size",
type=int,
default=DEFAULT_MAX_BATCH_SIZE,
help="Triton max_batch_size for the ensemble.",
)
parser.add_argument(
"--input-name",
default=DEFAULT_INPUT_NAME,
help="External client input tensor name.",
)
parser.add_argument(
"--input-data-type",
default=DEFAULT_INPUT_DATA_TYPE,
help="External client input tensor data type.",
)
parser.add_argument(
"--input-dims",
default=",".join(str(item) for item in DEFAULT_INPUT_DIMS),
help="Comma-separated external input dims, excluding batch dimension.",
)
parser.add_argument(
"--no-clean",
action="store_true",
help="Do not remove an existing output directory before generation.",
)
return parser
def main() -> None:
"""CLI entry point."""
args = build_arg_parser().parse_args()
config_path = generate_ensemble(
model_root=args.model_root,
output_dir=args.output_dir,
ensemble_name=args.ensemble_name,
max_batch_size=args.max_batch_size,
input_name=args.input_name,
input_data_type=args.input_data_type,
input_dims=parse_dims_arg(args.input_dims),
clean=not args.no_clean,
)
print(f"Wrote Triton ensemble config: {config_path}")
if __name__ == "__main__":
main()

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/inflammation_model_efficientnet_b0_ultrasound_2_cls/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "inflammation_model_efficientnet_b0_ultrasound_2_cls"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 2 ]
}
]

215
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/msk_vision_pipeline_ensemble/config.pbtxt Normal file
View File

@@ -0,0 +1,215 @@
name: "msk_vision_pipeline_ensemble"
platform: "ensemble"
backend: "ensemble"
max_batch_size: 8
# MODIFY HERE: Declare 2 separate input ports with fixed dimensions; no longer using the flexible -1 dimension
input [
{
name: "input_224"
data_type: TYPE_FP32
dims: [ 3, 224, 224 ]
},
{
name: "input_512"
data_type: TYPE_FP32
dims: [ 3, 512, 512 ]
}
]
output [
{
name: "angle_classify_convnext_tiny_logits"
data_type: TYPE_FP32
dims: [ 4 ]
},
{
name: "angle_classify_resnet50_logits"
data_type: TYPE_FP32
dims: [ 4 ]
},
{
name: "angle_classify_swin_v2_s_logits"
data_type: TYPE_FP32
dims: [ 4 ]
},
{
name: "angle_classify_densenet_logits"
data_type: TYPE_FP32
dims: [ 4 ]
},
{
name: "angle_classify_efficientnet_logits"
data_type: TYPE_FP32
dims: [ 4 ]
},
{
name: "inflammation_model_efficientnet_b0_ultrasound_2_cls_logits"
data_type: TYPE_FP32
dims: [ 2 ]
},
{
name: "segmentation_model_unet_resnet101_logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
},
{
name: "segmentation_model_unet3plus_att_logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
},
{
name: "segmentation_model_post_deeplabv3_resnet101_logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
},
{
name: "segmentation_model_post_deeplabv3_logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
},
{
name: "segmentation_model_post_efficientfeedback_logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
}
]
ensemble_scheduling {
step [
# ---- 224x224 MODEL GROUP (Processes input data from the input_224 variable) ----
{
model_name: "angle_classify_convnext_tiny"
model_version: -1
input_map {
key: "input_image"
value: "input_224"
}
output_map {
key: "logits"
value: "angle_classify_convnext_tiny_logits"
}
},
{
model_name: "angle_classify_resnet50"
model_version: -1
input_map {
key: "input_image"
value: "input_224"
}
output_map {
key: "logits"
value: "angle_classify_resnet50_logits"
}
},
{
model_name: "angle_classify_swin_v2_s"
model_version: -1
input_map {
key: "input_image"
value: "input_224"
}
output_map {
key: "logits"
value: "angle_classify_swin_v2_s_logits"
}
},
{
model_name: "angle_classify_densenet"
model_version: -1
input_map {
key: "input_image"
value: "input_224"
}
output_map {
key: "logits"
value: "angle_classify_densenet_logits"
}
},
{
model_name: "angle_classify_efficientnet"
model_version: -1
input_map {
key: "input_image"
value: "input_224"
}
output_map {
key: "logits"
value: "angle_classify_efficientnet_logits"
}
},
{
model_name: "inflammation_model_efficientnet_b0_ultrasound_2_cls"
model_version: -1
input_map {
key: "input_image"
value: "input_224"
}
output_map {
key: "logits"
value: "inflammation_model_efficientnet_b0_ultrasound_2_cls_logits"
}
},
# ---- 512x512 MODEL GROUP (Processes input data from the input_512 variable) ----
{
model_name: "segmentation_model_unet_resnet101"
model_version: -1
input_map {
key: "input_image"
value: "input_512"
}
output_map {
key: "logits"
value: "segmentation_model_unet_resnet101_logits"
}
},
{
model_name: "segmentation_model_unet3plus_att"
model_version: -1
input_map {
key: "input_image"
value: "input_512"
}
output_map {
key: "logits"
value: "segmentation_model_unet3plus_att_logits"
}
},
{
model_name: "segmentation_model_post_deeplabv3_resnet101"
model_version: -1
input_map {
key: "input_image"
value: "input_512"
}
output_map {
key: "logits"
value: "segmentation_model_post_deeplabv3_resnet101_logits"
}
},
{
model_name: "segmentation_model_post_deeplabv3"
model_version: -1
input_map {
key: "input_image"
value: "input_512"
}
output_map {
key: "logits"
value: "segmentation_model_post_deeplabv3_logits"
}
},
{
model_name: "segmentation_model_post_efficientfeedback"
model_version: -1
input_map {
key: "input_image"
value: "input_512"
}
output_map {
key: "logits"
value: "segmentation_model_post_efficientfeedback_logits"
}
}
]
}

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/segmentation_model_post_deeplabv3/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "segmentation_model_post_deeplabv3"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 512, 512 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/segmentation_model_post_deeplabv3_resnet101/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "segmentation_model_post_deeplabv3_resnet101"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 512, 512 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/segmentation_model_post_efficientfeedback/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "segmentation_model_post_efficientfeedback"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 512, 512 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/segmentation_model_unet3plus_att/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "segmentation_model_unet3plus_att"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 512, 512 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
}
]

17
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/segmentation_model_unet_resnet101/config.pbtxt Normal file
View File

@@ -0,0 +1,17 @@
name: "segmentation_model_unet_resnet101"
platform: "pytorch_libtorch"
max_batch_size: 8
input [
{
name: "input_image"
data_type: TYPE_FP32
dims: [ 3, 512, 512 ]
}
]
output [
{
name: "logits"
data_type: TYPE_FP32
dims: [ 7, 512, 512 ]
}
]

346
workspace/sprint_1_2/CODEBASE/infra/implementation/s3/upload_ensemble.py Normal file
View File

@@ -0,0 +1,346 @@
#!/usr/bin/env python3
"""Upload the generated Triton ensemble repository to AWS S3.
This script mirrors the local Triton model repository folder into the active
VKIST model bucket root:
s3://vkist-ml-model/my_vision_pipeline_ensemble/
It uploads every file and also creates zero-byte directory marker objects so the
S3 prefix reflects the same nested structure Triton expects in a model
repository.
"""
from __future__ import annotations
import argparse
import mimetypes
import os
from pathlib import Path
from typing import Iterable
try:
import boto3
from boto3.s3.transfer import TransferConfig
except ImportError: # pragma: no cover - exercised only when boto3 is absent.
boto3 = None
TransferConfig = None
ENSEMBLE_NAME = "my_vision_pipeline_ensemble"
DEFAULT_SOURCE_DIR = Path(__file__).resolve().parent / ENSEMBLE_NAME
DEFAULT_BUCKET_URI = "s3://vkist-ml-model/"
DEFAULT_TRANSFER_CONFIG = None
def require_env(name: str) -> str:
"""Read a required AWS credential/environment value."""
value = os.environ.get(name)
if not value:
raise RuntimeError(f"Missing required environment variable: {name}")
return value
def parse_s3_uri(uri: str) -> tuple[str, str]:
"""Parse an S3 URI into bucket and prefix."""
if not uri.startswith("s3://"):
raise ValueError(f"S3 URI must start with 's3://': {uri}")
body = uri.removeprefix("s3://").strip()
if not body:
raise ValueError("S3 URI must include a bucket name.")
parts = body.split("/", 1)
bucket = parts[0]
prefix = parts[1] if len(parts) > 1 else ""
return bucket, normalize_prefix(prefix)
def normalize_prefix(prefix: str) -> str:
"""Ensure an S3 prefix ends with '/' when non-empty."""
prefix = prefix.strip().replace("\\", "/")
if prefix and not prefix.endswith("/"):
return prefix + "/"
return prefix
def relpath_for(path: Path, root: Path) -> Path:
"""Return a POSIX-style relative path under a root directory."""
return path.relative_to(root).as_posix()
def collect_local_tree(source_dir: Path) -> tuple[list[Path], list[Path]]:
"""Collect local files and directories to mirror into S3."""
if not source_dir.exists():
raise FileNotFoundError(f"Source directory does not exist: {source_dir}")
if not source_dir.is_dir():
raise NotADirectoryError(f"Source path is not a directory: {source_dir}")
files = [path for path in source_dir.rglob("*") if path.is_file()]
directories = [path for path in source_dir.rglob("*") if path.is_dir()]
directories.append(source_dir)
return sorted(files), sorted(directories, reverse=True)
def file_key_for(source_dir: Path, file_path: Path, prefix: str) -> str:
"""Build the S3 object key for a local file."""
return prefix + relpath_for(file_path, source_dir).replace("\\", "/")
def directory_marker_key_for(source_dir: Path, directory_path: Path, prefix: str) -> str:
"""Build the S3 directory marker key for a local directory."""
if directory_path == source_dir:
return prefix
return prefix + relpath_for(directory_path, source_dir).replace("\\", "/") + "/"
def content_type_for(path: Path) -> str:
"""Guess a safe MIME type for an S3 object."""
guessed_type, _ = mimetypes.guess_type(str(path))
return guessed_type or "application/octet-stream"
def create_s3_client() -> object:
"""Create a Boto3 S3 client from local AWS environment variables."""
if boto3 is None:
raise RuntimeError("boto3 is required. Install it with: pip install boto3")
access_key = require_env("AWS_ACCESS_KEY_ID")
secret_key = require_env("AWS_SECRET_ACCESS_KEY")
client_kwargs = {
"aws_access_key_id": access_key,
"aws_secret_access_key": secret_key,
}
session_token = os.environ.get("AWS_SESSION_TOKEN")
if session_token:
client_kwargs["aws_session_token"] = session_token
region = os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION")
if region:
client_kwargs["region_name"] = region
endpoint_url = os.environ.get("AWS_ENDPOINT_URL")
if endpoint_url:
client_kwargs["endpoint_url"] = endpoint_url
return boto3.client("s3", **client_kwargs)
def put_directory_marker(
s3_client: object,
bucket: str,
key: str,
dry_run: bool = False,
) -> None:
"""Create a zero-byte S3 marker object for one directory prefix."""
if dry_run:
print(f"[dry-run] marker s3://{bucket}/{key}")
return
s3_client.put_object(
Bucket=bucket,
Key=key,
Body=b"",
ContentType="application/x-directory",
)
def upload_file(
s3_client: object,
bucket: str,
local_path: Path,
key: str,
transfer_config: TransferConfig | None,
dry_run: bool = False,
) -> None:
"""Upload one local file to S3."""
if dry_run:
print(f"[dry-run] upload {local_path} -> s3://{bucket}/{key}")
return
if transfer_config is None:
if TransferConfig is None:
raise RuntimeError("boto3 is required. Install it with: pip install boto3")
transfer_config = TransferConfig(
multipart_threshold=64 * 1024 * 1024,
multipart_chunksize=16 * 1024 * 1024,
max_concurrency=8,
use_threads=True,
)
s3_client.upload_file(
Filename=str(local_path),
Bucket=bucket,
Key=key,
ExtraArgs={
"ContentType": content_type_for(local_path),
"CacheControl": "no-store",
},
Config=transfer_config,
)
def list_existing_keys(s3_client: object, bucket: str, prefix: str) -> set[str]:
"""List all existing object keys below an S3 prefix."""
paginator = s3_client.get_paginator("list_objects_v2")
keys: set[str] = set()
for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
for item in page.get("Contents", []):
keys.add(item["Key"])
return keys
def delete_stale_keys(
s3_client: object,
bucket: str,
prefix: str,
desired_keys: Iterable[str],
dry_run: bool = False,
) -> int:
"""Delete objects under the prefix that are not present locally."""
desired = set(desired_keys)
existing = list_existing_keys(s3_client, bucket, prefix)
stale = sorted(existing - desired)
if not stale:
return 0
if dry_run:
for key in stale:
print(f"[dry-run] delete s3://{bucket}/{key}")
return len(stale)
for key in stale:
s3_client.delete_object(Bucket=bucket, Key=key)
return len(stale)
def mirror_to_s3(
source_dir: Path,
bucket_uri: str,
prefix: str | None = None,
delete: bool = False,
dry_run: bool = False,
) -> dict[str, int]:
"""Mirror a local Triton model repository directory into S3."""
bucket, bucket_prefix = parse_s3_uri(bucket_uri)
effective_prefix = normalize_prefix(prefix if prefix is not None else source_dir.name + "/")
s3_prefix = bucket_prefix + effective_prefix
files, directories = collect_local_tree(source_dir)
s3_client = create_s3_client() if (not dry_run) or delete else None
desired_keys: set[str] = set()
for directory_path in directories:
key = directory_marker_key_for(source_dir, directory_path, s3_prefix)
desired_keys.add(key)
put_directory_marker(s3_client, bucket, key, dry_run=dry_run)
for file_path in files:
key = file_key_for(source_dir, file_path, s3_prefix)
desired_keys.add(key)
upload_file(
s3_client=s3_client,
bucket=bucket,
local_path=file_path,
key=key,
transfer_config=DEFAULT_TRANSFER_CONFIG,
dry_run=dry_run,
)
deleted = 0
if delete:
deleted = delete_stale_keys(
s3_client=s3_client,
bucket=bucket,
prefix=s3_prefix,
desired_keys=desired_keys,
dry_run=dry_run,
)
return {
"files_uploaded": len(files),
"directories_marked": len(directories),
"keys_desired": len(desired_keys),
"keys_deleted": deleted,
}
def build_arg_parser() -> argparse.ArgumentParser:
"""Build CLI arguments for S3 upload."""
parser = argparse.ArgumentParser(
description="Mirror a generated Triton ensemble repository to AWS S3."
)
parser.add_argument(
"--source-dir",
type=Path,
default=DEFAULT_SOURCE_DIR,
help="Local generated Triton model repository directory.",
)
parser.add_argument(
"--bucket-uri",
default=DEFAULT_BUCKET_URI,
help="S3 model bucket root URI, for example s3://vkist-ml-model/.",
)
parser.add_argument(
"--prefix",
default=None,
help="Optional S3 prefix under the bucket. Defaults to the source folder name.",
)
parser.add_argument(
"--delete",
action="store_true",
help="Delete stale objects under the target prefix that are missing locally.",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print S3 actions without writing or deleting objects.",
)
return parser
def main() -> None:
"""CLI entry point."""
args = build_arg_parser().parse_args()
summary = mirror_to_s3(
source_dir=args.source_dir,
bucket_uri=args.bucket_uri,
prefix=args.prefix,
delete=args.delete,
dry_run=args.dry_run,
)
print(
"Uploaded ensemble mirror: "
f"files={summary['files_uploaded']}, "
f"directories={summary['directories_marked']}, "
f"desired_keys={summary['keys_desired']}, "
f"deleted={summary['keys_deleted']}"
)
if __name__ == "__main__":
main()

29
workspace/sprint_1_2/CODEBASE/infra/implementation/sh_files/mv_file.sh Executable file
View File

@@ -0,0 +1,29 @@
# Create folder
# aws s3api put-object --bucket vkist-ml-model --key angle_classify_densenet/
aws s3api put-object --bucket vkist-ml-model --key angle_classify_efficientnet/ # best_efficientnet_b2.pth
aws s3api put-object --bucket vkist-ml-model --key angle_classify_resnet50/ # best_resnet50.pth
aws s3api put-object --bucket vkist-ml-model --key angle_classify_swin_v2_s/ # best_swin_v2_s.pth
aws s3api put-object --bucket vkist-ml-model --key segmentation_model_post_deeplabv3_resnet101/ # best_model_deeplabv3_resnet101_seed_16.pth
aws s3api put-object --bucket vkist-ml-model --key segmentation_model_post_deeplabv3/ # best_model_Deeplav3.pth
aws s3api put-object --bucket vkist-ml-model --key segmentation_model_post_efficientfeedback/ # efficientfeedback.pth
aws s3api put-object --bucket vkist-ml-model --key segmentation_model_unet_resnet101/ # unet_resnet101.pth
aws s3api put-object --bucket vkist-ml-model --key segmentation_model_unet3plus_att/ # unet3plus_att.pth
aws s3api put-object --bucket vkist-ml-model --key inflammation_model_efficientnet_b0_ultrasound_2_cls/ # efficientnet_b0_ultrasound_2_class.pth
aws s3api put-object --bucket vkist-ml-model --key msk_vision_pipeline_ensemble
# upload model
aws s3 mv s3://vkist-ml-model/best_densenet.pth s3://vkist-ml-model/angle_classify_densenet/best_densenet.pth
aws s3 mv s3://vkist-ml-model/best_efficientnet_b2.pth s3://vkist-ml-model/angle_classify_efficientnet/best_efficientnet_b2.pth
aws s3 mv s3://vkist-ml-model/best_model_deeplabv3_resnet101_seed_16.pth s3://vkist-ml-model/segmentation_model_post_deeplabv3_resnet101/best_model_deeplabv3_resnet101_seed_16.pth
aws s3 mv s3://vkist-ml-model/best_model_Deeplav3.pth s3://vkist-ml-model/segmentation_model_post_deeplabv3/best_model_Deeplav3.pth
aws s3 mv s3://vkist-ml-model/best_resnet50.pth s3://vkist-ml-model/angle_classify_resnet50/best_resnet50.pth
aws s3 mv s3://vkist-ml-model/best_swin_v2_s.pth s3://vkist-ml-model/angle_classify_swin_v2_s/best_swin_v2_s.pth
aws s3 mv s3://vkist-ml-model/efficientfeedback.pth s3://vkist-ml-model/segmentation_model_post_efficientfeedback/efficientfeedback.pth
aws s3 mv s3://vkist-ml-model/efficientnet_b0_ultrasound_2_class.pth s3://vkist-ml-model/inflammation_model_efficientnet_b0_ultrasound_2_cls/efficientnet_b0_ultrasound_2_class.pth
aws s3 mv s3://vkist-ml-model/unet_resnet101.pth s3://vkist-ml-model/segmentation_model_unet_resnet101/unet_resnet101.pth
aws s3 mv s3://vkist-ml-model/unet3plus_att.pth s3://vkist-ml-model/segmentation_model_unet3plus_att/unet3plus_att.pth
aws s3 mv s3://vkist-ml-model/best_convnext_tiny.pth s3://vkist-ml-model/angle_classify_convnext_tiny/best_convnext_tiny.pth

14
workspace/sprint_1_2/CODEBASE/infra/implementation/sh_files/mv_sh_2.sh Executable file
View File

@@ -0,0 +1,14 @@
# Classification Models
aws s3 mv s3://vkist-ml-model/angle_classify_densenet/best_densenet.pth s3://vkist-ml-model/angle_classify_densenet/1/best_densenet.pth
aws s3 mv s3://vkist-ml-model/angle_classify_efficientnet/best_efficientnet_b2.pth s3://vkist-ml-model/angle_classify_efficientnet/1/best_efficientnet_b2.pth
aws s3 mv s3://vkist-ml-model/angle_classify_resnet50/best_resnet50.pth s3://vkist-ml-model/angle_classify_resnet50/1/best_resnet50.pth
aws s3 mv s3://vkist-ml-model/angle_classify_swin_v2_s/best_swin_v2_s.pth s3://vkist-ml-model/angle_classify_swin_v2_s/1/best_swin_v2_s.pth
aws s3 mv s3://vkist-ml-model/angle_classify_convnext_tiny/best_convnext_tiny.pth s3://vkist-ml-model/angle_classify_convnext_tiny/1/best_convnext_tiny.pth
aws s3 mv s3://vkist-ml-model/inflammation_model_efficientnet_b0_ultrasound_2_cls/efficientnet_b0_ultrasound_2_class.pth s3://vkist-ml-model/inflammation_model_efficientnet_b0_ultrasound_2_cls/1/efficientnet_b0_ultrasound_2_class.pth
# Segmentation Models
aws s3 mv s3://vkist-ml-model/segmentation_model_post_deeplabv3_resnet101/best_model_deeplabv3_resnet101_seed_16.pth s3://vkist-ml-model/segmentation_model_post_deeplabv3_resnet101/1/best_model_deeplabv3_resnet101_seed_16.pth
aws s3 mv s3://vkist-ml-model/segmentation_model_post_deeplabv3/best_model_Deeplav3.pth s3://vkist-ml-model/segmentation_model_post_deeplabv3/1/best_model_Deeplav3.pth
aws s3 mv s3://vkist-ml-model/segmentation_model_post_efficientfeedback/efficientfeedback.pth s3://vkist-ml-model/segmentation_model_post_efficientfeedback/1/efficientfeedback.pth
aws s3 mv s3://vkist-ml-model/segmentation_model_unet_resnet101/unet_resnet101.pth s3://vkist-ml-model/segmentation_model_unet_resnet101/1/unet_resnet101.pth
aws s3 mv s3://vkist-ml-model/segmentation_model_unet3plus_att/unet3plus_att.pth s3://vkist-ml-model/segmentation_model_unet3plus_att/1/unet3plus_att.pth

29
workspace/sprint_1_2/CODEBASE/infra/implementation/sh_files/mv_sh_3.sh Executable file
View File

@@ -0,0 +1,29 @@
#!/bin/bash
S3_BUCKET="s3://vkist-ml-model"
# Set the exact local baseline path where your updated config.pbtxt models reside
LOCAL_CONFIG_DIR="/Users/davestran/Downloads/vkist_internship/PILOT_PROJECT/workspace/sprint_1_2/codebase/infra/implementation/s3"
echo "📤 Syncing local config.pbtxt modifications up to S3 bucket repository..."
# Classification Configs
aws s3 cp "$LOCAL_CONFIG_DIR/angle_classify_convnext_tiny/config.pbtxt" "$S3_BUCKET/angle_classify_convnext_tiny/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/angle_classify_densenet/config.pbtxt" "$S3_BUCKET/angle_classify_densenet/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/angle_classify_efficientnet/config.pbtxt" "$S3_BUCKET/angle_classify_efficientnet/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/angle_classify_resnet50/config.pbtxt" "$S3_BUCKET/angle_classify_resnet50/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/angle_classify_swin_v2_s/config.pbtxt" "$S3_BUCKET/angle_classify_swin_v2_s/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/inflammation_model_efficientnet_b0_ultrasound_2_cls/config.pbtxt" "$S3_BUCKET/inflammation_model_efficientnet_b0_ultrasound_2_cls/config.pbtxt"
# Segmentation Configs
aws s3 cp "$LOCAL_CONFIG_DIR/segmentation_model_post_deeplabv3/config.pbtxt" "$S3_BUCKET/segmentation_model_post_deeplabv3/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/segmentation_model_post_deeplabv3_resnet101/config.pbtxt" "$S3_BUCKET/segmentation_model_post_deeplabv3_resnet101/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/segmentation_model_post_efficientfeedback/config.pbtxt" "$S3_BUCKET/segmentation_model_post_efficientfeedback/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/segmentation_model_unet3plus_att/config.pbtxt" "$S3_BUCKET/segmentation_model_unet3plus_att/config.pbtxt"
aws s3 cp "$LOCAL_CONFIG_DIR/segmentation_model_unet_resnet101/config.pbtxt" "$S3_BUCKET/segmentation_model_unet_resnet101/config.pbtxt"
# Ensemble Config
aws s3 cp "$LOCAL_CONFIG_DIR/msk_vision_pipeline_ensemble/config.pbtxt" "$S3_BUCKET/msk_vision_pipeline_ensemble/config.pbtxt"
echo "✅ Configuration files successfully synced to S3 backend!"

32
workspace/sprint_1_2/CODEBASE/infra/implementation/sh_files/mv_sh_4.sh Executable file
View File

@@ -0,0 +1,32 @@
#!/bin/bash
# Define the absolute local path to your compiled TorchScript folder
LOCAL_DIR="PILOT_PROJECT/workspace/sprint_1_2/codebase/infra/implementation/MODEL_ZIP_PILOT_LT"
S3_BUCKET="s3://vkist-ml-model"
echo "📤 Starting upload workflow of LibTorch binaries to S3 bucket layout..."
# ==========================================
# 1. Classification Models
# ==========================================
echo "🔄 Uploading: Classification models..."
aws s3 cp "$LOCAL_DIR/best_densenet.pth" "$S3_BUCKET/angle_classify_densenet/1/model.pt"
aws s3 cp "$LOCAL_DIR/best_efficientnet_b2.pth" "$S3_BUCKET/angle_classify_efficientnet/1/model.pt"
aws s3 cp "$LOCAL_DIR/best_resnet50.pth" "$S3_BUCKET/angle_classify_resnet50/1/model.pt"
aws s3 cp "$LOCAL_DIR/best_swin_v2_s.pth" "$S3_BUCKET/angle_classify_swin_v2_s/1/model.pt"
aws s3 cp "$LOCAL_DIR/best_convnext_tiny.pth" "$S3_BUCKET/angle_classify_convnext_tiny/1/model.pt"
aws s3 cp "$LOCAL_DIR/efficientnet_b0_ultrasound_2_class.pth" "$S3_BUCKET/inflammation_model_efficientnet_b0_ultrasound_2_cls/1/model.pt"
# ==========================================
# 2. Segmentation Models
# ==========================================
echo "🔄 Uploading: Segmentation models..."
aws s3 cp "$LOCAL_DIR/best_model_deeplabv3_resnet101_seed_16.pth" "$S3_BUCKET/segmentation_model_post_deeplabv3_resnet101/1/model.pt"
aws s3 cp "$LOCAL_DIR/best_model_Deeplav3.pth" "$S3_BUCKET/segmentation_model_post_deeplabv3/1/model.pt"
aws s3 cp "$LOCAL_DIR/efficientfeedback.pth" "$S3_BUCKET/segmentation_model_post_efficientfeedback/1/model.pt"
aws s3 cp "$LOCAL_DIR/unet_resnet101.pth" "$S3_BUCKET/segmentation_model_unet_resnet101/1/model.pt"
aws s3 cp "$LOCAL_DIR/unet3plus_att.pth" "$S3_BUCKET/segmentation_model_unet3plus_att/1/model.pt"
echo "🎉 All local LibTorch models compiled down and synchronized with S3 Triton targets successfully!"

216
workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/modal_triton.py Normal file
View File

@@ -0,0 +1,216 @@
import subprocess
import modal
import time
# run from root: vkist_internship (will manaage later)
triton_image = (
modal.Image.from_registry(
tag="nvcr.io/nvidia/tritonserver:24.02-py3",
add_python="3.12"
)
# Step B: Install minimal system dependencies (replacing your apt-get RUN command)
.apt_install(
"libgl1",
"libglib2.0-0" # Crucial runtime hook for OpenCV / Ultralytics
)
# Step C: Install PyTorch pinned strictly to CUDA 12.1 wheel indices
.run_commands(
"python3 -m pip install --upgrade pip setuptools",
"python3 -m pip install torch==2.5.0 torchaudio==2.5.0 torchvision==0.20.0 --index-url https://download.pytorch.org/whl/cu121",
"python3 -m pip install transformers==4.57.3 timm==1.0.22 ultralytics==8.3.0 opencv-python grpcio protobuf",
"python3 -m pip install fastapi[standard]",
"python3 -m pip install tritonclient[http,cuda]"
)
)
app = modal.App("triton-s3-service", image=triton_image)
from fastapi import FastAPI, Response, Request,HTTPException
from fastapi.responses import StreamingResponse # 👈 ADD THIS IMPORT
import httpx
web_app = FastAPI()
# -------------------------------------------------------------
# FASTAPI PROXY ROUTING (Living inside the container)
# -------------------------------------------------------------
@web_app.get("/v2/health/ready")
async def forward_health():
"""Proxies external HTTP REST calls straight to Triton's internal inference engine"""
async with httpx.AsyncClient() as client:
try:
response = await client.get("http://127.0.0.1:8000/v2/health/ready")
return Response(content=response.content, status_code=response.status_code, media_type=response.headers.get("content-type"))
except Exception as e:
return Response(content=f"Triton booting models from S3... Error: {str(e)}", status_code=503)
@web_app.get("/metrics")
@web_app.get("/")
async def forward_metrics():
"""Proxies external metric calls straight to Triton's internal metrics engine"""
async with httpx.AsyncClient() as client:
try:
response = await client.get("http://127.0.0.1:8002/metrics")
return Response(content=response.content, status_code=response.status_code, media_type=response.headers.get("content-type"))
except Exception as e:
return Response(content=f"Waiting for metrics channel... Error: {str(e)}", status_code=503)
# 👇 ADD THIS CATCH-ALL ROUTE HERE 👇
@web_app.api_route("/v2/{path:path}", methods=["GET", "POST"])
async def proxy_all_triton_request(path: str, request: Request):
import tritonclient.grpc.aio as grpcclient
from tritonclient.grpc import service_pb2, service_pb2_grpc
from tritonclient.grpc import _utils as grpc_utils #InferenceServerClient
import grpc
import numpy as np
# 1. Keep HTTP proxy ONLY for metadata/health checks
if "infer" not in path:
async with httpx.AsyncClient(timeout=60.0) as client:
url = f"http://127.0.0.1:8000/v2/{path}"
headers = dict(request.headers)
headers.pop("host", None)
triton_response = await client.request(
method=request.method, url=url, headers=headers, content=await request.body()
)
return Response(
content=triton_response.content,
status_code=triton_response.status_code,
headers=dict(triton_response.headers)
)
# 2. 🚀 FOR INFERENCE: Convert the incoming HTTP raw body into a gRPC call
if "infer" in path:
try:
# Extract model name from the route path (e.g., v2/models/MODEL_NAME/infer)
parts = path.split("/")
model_name = parts[parts.index("models") + 1]
# Read incoming raw binary HTTP payload
raw_http_body = await request.body()
header_length_str = request.headers.get("Inference-Header-Content-Length")
if not header_length_str:
raise HTTPException(
status_code=400,
detail="Missing 'Inference-Header-Content-Length' header required for binary Triton transcoding."
)
header_length = int(header_length_str)
# --- 💥 KSERVE V2 MULTI-PART BODY PARSING ---
# Extract the front JSON metadata and the trailing raw binary tensors
import json
json_bytes = raw_http_body[:header_length]
binary_data = raw_http_body[header_length:]
request_metadata = json.loads(json_bytes.decode('utf-8'))
# Setup async gRPC connection
triton_url = "127.0.0.1:8001"
# Configure channels to accept large payload returns (100MB limit override)
max_msg_length = 100 * 1024 * 1024
channel_options = [
('grpc.max_receive_message_length', max_msg_length),
('grpc.max_send_message_length', max_msg_length),
]
async with grpc.aio.insecure_channel(triton_url, options=channel_options) as channel:
stub = service_pb2_grpc.GRPCInferenceServiceStub(channel=channel)
# Construct the native ModelInferRequest protobuf
grpc_request = service_pb2.ModelInferRequest()
grpc_request.model_name = model_name
grpc_request.model_version = ""
# Populate inputs dynamically from incoming KServe metadata
binary_offset = 0
for input_tensor in request_metadata.get("inputs", []):
# Correct Protobuf repeated field instantiation via .add()
infer_input = grpc_request.inputs.add()
infer_input.name = input_tensor["name"]
infer_input.datatype = input_tensor["datatype"]
infer_input.shape.extend(input_tensor["shape"]) # Explicit clean integers!
# Extract the binary slice matching this tensor out of the raw payload block
if "parameters" in input_tensor and "binary_data_size" in input_tensor["parameters"]:
data_size = input_tensor["parameters"]["binary_data_size"]
grpc_request.raw_input_contents.append(
binary_data[binary_offset : binary_offset + data_size]
)
binary_offset += data_size
# Request output tensor mappings dynamically based on what the client requested
for output_tensor in request_metadata.get("outputs", []):
infer_output = grpc_request.outputs.add()
infer_output.name = output_tensor["name"]
# Signal Triton to return output via raw binary buffers
infer_output.parameters["binary_data"].bool_param = True
# ✅ Send the transcoding payload straight into Triton over internal gRPC loop
grpc_response = await stub.ModelInfer(request=grpc_request, timeout=None)
# --- 💥 TRANSCODE gRPC RESPONSE BACK TO MULTI-PART KSERVE HTTP ---
response_metadata = {
"model_name": grpc_response.model_name,
"model_version": grpc_response.model_version,
"outputs": []
}
response_binary_body = b""
for i, output in enumerate(grpc_response.outputs):
out_desc = {
"name": output.name,
"datatype": output.datatype,
"shape": list(output.shape),
"parameters": {
"binary_data_size": len(grpc_response.raw_output_contents[i])
}
}
response_metadata["outputs"].append(out_desc)
response_binary_body += grpc_response.raw_output_contents[i]
# Re-bundle into [JSON metadata] + [Raw binary output chunks]
response_json_bytes = json.dumps(response_metadata).encode('utf-8')
output_http_body = response_json_bytes + response_binary_body
return Response(
content=output_http_body,
status_code=200,
headers={
"Content-Type": "application/octet-stream",
"Inference-Header-Content-Length": str(len(response_json_bytes))
}
)
except Exception as e:
import traceback
print(f"CRITICAL TRANSLATION EXCEPTION: {traceback.format_exc()}")
return Response(
content=f"Internal gRPC Pipeline Multiplex Error: {str(e)}",
status_code=502
)
# -------------------------------------------------------------
# THE UNIFIED SERVICE FUNCTION (1 Container, 1 GPU, 1 Triton Process)
# -------------------------------------------------------------
@app.function(
gpu="T4", # for the expense
timeout=3600,
max_containers=3, # Strict production capping
min_containers=1, # for keeping warm and prevention,
buffer_containers=2, # Number of additional idle containers to maintain under active load.
scaledown_window=30, # Max time (in seconds) a container can remain idle while scaling down.
secrets=[modal.Secret.from_name("aws-secrets")]
)
@modal.asgi_app()
def unified_triton_server():
print("🚀 Booting ONE Triton Instance inside ONE A100 Container...")
# Spawns Triton in the background. It will automatically read
# your "aws-secrets" environment keys to mount s3://vkist-ml-model/
cmd = ["tritonserver", "--model-repository=s3://vkist-ml-model/"]
subprocess.Popen(cmd)
print("📋 Triton background process delegated. Handing routing control over to FastAPI.")
# Returns immediately! FastAPI now takes over the container lifecycle
return web_app

1
workspace/sprint_1_2/CODEBASE/infra/implementation/triton_run/run.sh Executable file
View File

@@ -0,0 +1 @@
modal deploy PILOT_PROJECT/workspace/sprint_1_2/codebase/infra/implementation/triton_run/modal_triton.py