Real-time AI for surgery
with NVIDIA-Holoscan platform
ARC Collaborations, UCL
Miguel Xochicale, PhD
mxochicale
2024-04-20 @ Link for grid-worms-animation 2023
import os
from argparse import ArgumentParser
from holoscan.core import Application
from holoscan.operators import (
FormatConverterOp,
HolovizOp,
InferenceOp,
SegmentationPostprocessorOp,
VideoStreamReplayerOp,
)
from holoscan.resources import UnboundedAllocator
class BYOMApp(Application):
def __init__(self, data):
"""Initialize the application
Parameters
----------
data : Location to the data
"""
super().__init__()
# set name
self.name = "BYOM App"
if data == "none":
data = os.environ.get("HOLOSCAN_INPUT_PATH", "../data")
self.sample_data_path = data
self.model_path = os.path.join(os.path.dirname(__file__), "../model")
self.model_path_map = {
"byom_model": os.path.join(self.model_path, "identity_model.onnx"),
}
self.video_dir = os.path.join(self.sample_data_path, "racerx")
if not os.path.exists(self.video_dir):
raise ValueError(f"Could not find video data:{self.video_dir=}")
# Define the workflow
self.add_flow(source, viz, {("output", "receivers")})
self.add_flow(source, preprocessor, {("output", "source_video")})
self.add_flow(preprocessor, inference, {("tensor", "receivers")})
self.add_flow(inference, postprocessor, {("transmitter", "in_tensor")})
self.add_flow(postprocessor, viz, {("out_tensor", "receivers")})
def main(config_file, data):
app = BYOMApp(data=data)
# if the --config command line argument was provided, it will override this config_file
app.config(config_file)
app.run()
if __name__ == "__main__":
# Parse args
parser = ArgumentParser(description="BYOM demo application.")
parser.add_argument(
"-d",
"--data",
default="none",
help=("Set the data path"),
)
args = parser.parse_args()
config_file = os.path.join(os.path.dirname(__file__), "byom.yaml")
main(config_file=config_file, data=args.data)
%YAML 1.2
replayer: # VideoStreamReplayer
basename: "racerx"
frame_rate: 0 # as specified in timestamps
repeat: true # default: false
realtime: true # default: true
count: 0 # default: 0 (no frame count restriction)
preprocessor: # FormatConverter
out_tensor_name: source_video
out_dtype: "float32"
resize_width: 512
resize_height: 512
inference: # Inference
backend: "trt"
pre_processor_map:
"byom_model": ["source_video"]
inference_map:
"byom_model": ["output"]
postprocessor: # SegmentationPostprocessor
in_tensor_name: output
# network_output_type: None
data_format: nchw
viz: # Holoviz
width: 854
height: 480
color_lut: [
[0.65, 0.81, 0.89, 0.1],
]
real-time-ai-for-surgery
real-time-ai-for-surgery
real-time-ai-for-surgery
multi-ai.py
...
# Define the workflow
if is_v4l2:
self.add_flow(source, viz, {("signal", "receivers")})
self.add_flow(source, preprocessor_v4l2, {("signal", "source_video")})
self.add_flow(source, preprocessor_phasenet_v4l2, {("signal", "source_video")})
for op in [preprocessor_v4l2, preprocessor_phasenet_v4l2]:
self.add_flow(op, multi_ai_inference_v4l2, {("", "receivers")})
### connect infereceOp to postprocessors
self.add_flow(
multi_ai_inference_v4l2, multiheadOp, {("transmitter", "in_tensor_postproOp")}
)
self.add_flow(multi_ai_inference_v4l2, segpostprocessor, {("transmitter", "")})
self.add_flow(multi_ai_inference_v4l2, phasenetOp, {("", "in")})
else:
self.add_flow(source, viz, {("", "receivers")})
self.add_flow(source, preprocessor_replayer, {("output", "source_video")})
self.add_flow(source, preprocessor_phasenet_replayer, {("output", "source_video")})
for op in [preprocessor_replayer, preprocessor_phasenet_replayer]:
self.add_flow(op, multi_ai_inference_replayer, {("", "receivers")})
### connect infereceOp to postprocessors
self.add_flow(
multi_ai_inference_replayer, multiheadOp, {("transmitter", "in_tensor_postproOp")}
)
self.add_flow(multi_ai_inference_replayer, segpostprocessor, {("transmitter", "")})
self.add_flow(multi_ai_inference_replayer, phasenetOp, {("", "in")})
## connect postprocessors outputs for visualisation with holoviz
self.add_flow(multiheadOp, viz, {("out_tensor_postproOp", "receivers")})
self.add_flow(segpostprocessor, viz, {("", "receivers")})
self.add_flow(phasenetOp, viz, {("out", "receivers")})
self.add_flow(phasenetOp, viz, {("output_specs", "input_specs")})
...
multi-ai.yaml
...
multi_ai_inference_v4l2:
#
#
# Multi-AI Inference Operator InferenceOp()
#
#
backend: "trt"
pre_processor_map:
"pit_surg_model": ["prepro_v4l2"]
"phasenet_model": ["prepro_PNv4l2"]
inference_map:
"pit_surg_model": ["segmentation_masks", "landmarks"]
"phasenet_model": ["out"]
enable_fp16: False
parallel_inference: true # optional param, default to true
infer_on_cpu: false # optional param, default to false
input_on_cuda: true # optional param, default to true
output_on_cuda: true # optional param, default to true
transmit_on_cuda: true # optional param, default to true
is_engine_path: false # optional param, default to false
multi_ai_inference_replayer:
#
#
# Multi-AI Inference Operator InferenceOp()
#
#
backend: "trt"
pre_processor_map:
"pit_surg_model": ["prepro_replayer"]
"phasenet_model": ["prepro_PNreplayer"]
inference_map:
"pit_surg_model": ["segmentation_masks", "landmarks"]
"phasenet_model": ["out"]
enable_fp16: False
parallel_inference: true # optional param, default to true
infer_on_cpu: false # optional param, default to false
input_on_cuda: true # optional param, default to true
output_on_cuda: true # optional param, default to true
transmit_on_cuda: true # optional param, default to true
is_engine_path: false # optional param, default to false
...
real-time-ai-for-surgery
real-time-ai-for-surgery
real-time-ai-for-surgery
real-time-ai-for-surgery