Skip to content

migx

Type Aliases:

Classes:

  • MIGX –

    MIGraphX backend for AMD GPUs.

Attributes:

logger module-attribute 

logger = getLogger(__name__)

Shape 

Shape = tuple[int, int]

MIGX dataclass 

MIGX(
    *,
    device_id: int = 0,
    num_streams: int = 1,
    fp16: bool | None = None,
    bf16: bool | None = None,
    opt_shapes: Shape | None = None,
    fast_math: bool = True,
    exhaustive_tune: bool = False,
    custom_env: Mapping[str, str] = dict[str, str](),
    custom_args: Sequence[str] = list[str](),
    force_rebuild: bool = False,
)

Bases: BackendAutoConvertFloat

MIGraphX backend for AMD GPUs.

The ONNX model is compiled to an MXR program with migraphx-driver and cached before execution through core.migx.Model.

Methods:

Attributes:

bf16 class-attribute instance-attribute 

bf16: bool | None = None

Compile the program for BF16 where supported. Default to False.

custom_args class-attribute instance-attribute 

custom_args: Sequence[str] = field(default_factory=list[str])

Additional command-line arguments appended to migraphx-driver compile.

migraph-driver compile --help

custom_env class-attribute instance-attribute 

custom_env: Mapping[str, str] = field(default_factory=dict[str, str])

Extra environment variables for migraphx-driver.

https://rocm.docs.amd.com/projects/AMDMIGraphX/en/latest/reference/MIGraphX-dev-env-vars.html https://rocm.docs.amd.com/projects/MIOpen/en/latest/reference/env_variables.html

device_id class-attribute instance-attribute 

device_id: int = 0

AMD GPU device index.

exhaustive_tune class-attribute instance-attribute 

exhaustive_tune: bool = False

Enable exhaustive tuning during compilation.

fast_math class-attribute instance-attribute 

fast_math: bool = True

Keep MIGraphX fast math optimizations enabled.

flexible_output_prop class-attribute 

flexible_output_prop: str = 'MlrtFlexible'

force_rebuild class-attribute instance-attribute 

force_rebuild: bool = field(default=False, repr=False)

Force a full program rebuild, ignoring any cached program.

fp16 class-attribute instance-attribute 

fp16: bool | None = None

Compile the program for FP16 where supported. Default to True

num_streams class-attribute instance-attribute 

num_streams: int = 1

Number of parallel MIGraphX inference streams.

opt_shapes class-attribute instance-attribute 

opt_shapes: Shape | None = None

Optimization input tile size as (width, height). Defaults to the inference tile size.

plugin class-attribute instance-attribute 

plugin = core.lazy.migx

version property 

version: tuple[int, int]

autoselect staticmethod 

autoselect(device_id: int = 0, **kwargs: Any) -> Backend

Try to select the best backend for the current system.

Parameters:

  • device_id

    (int, default: 0 ) –

    The GPU device id.

  • **kwargs

    (Any, default: {} ) –

    Additional arguments to pass to the backend.

Returns:

  • Backend –

    The selected backend.

Source code in vsscale/mlrt/backend/base.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@staticmethod
def autoselect(device_id: int = 0, **kwargs: Any) -> Backend:
    """
    Try to select the best backend for the current system.

    Args:
        device_id: The GPU device id.
        **kwargs: Additional arguments to pass to the backend.

    Returns:
        The selected backend.
    """

    gpu = get_gpu(device_id)
    vendor = None if not gpu else str(gpu.vendor).strip()

    match vendor:
        # Windows & Linux
        case "NVIDIA Corporation":
            if hasattr(core, "trt"):
                backend = UserBackend.TRT
            elif hasattr(core, "trt_rtx"):
                backend = UserBackend.TRT_RTX
            elif platform.system().lower() == "windows" and hasattr(core, "ort"):
                backend = UserBackend.ORT_DML
            elif hasattr(core, "ort"):
                backend = UserBackend.ORT_CUDA
            elif hasattr(core, "ncnn"):
                backend = UserBackend.NCNN
            else:
                backend = UserBackend.OV_CPU
        # Windows & Linux
        case "Advanced Micro Devices, Inc.":
            if platform.system().lower() == "windows" and hasattr(core, "ort"):
                backend = UserBackend.ORT_DML
            elif hasattr(core, "migx"):
                backend = UserBackend.MIGX
            elif hasattr(core, "ncnn"):
                backend = UserBackend.NCNN_VK
            else:
                backend = UserBackend.OV_CPU
        # Windows & Linux
        case "Intel(R) Corporation":
            if hasattr(core, "ov"):
                backend = UserBackend.OV_GPU
            elif platform.system().lower() == "windows" and hasattr(core, "ort"):
                backend = UserBackend.ORT_DML
            elif hasattr(core, "ncnn"):
                backend = UserBackend.NCNN_VK
            else:
                backend = UserBackend.OV_CPU
        # macOS ARM64 & x86_64
        case "Apple":
            if hasattr(core, "ncnn"):
                backend = UserBackend.NCNN_VK
            elif hasattr(core, "ort"):
                backend = UserBackend.ORT_COREML
            else:
                backend = UserBackend.OV_CPU
        case _:
            backend = UserBackend.OV_CPU

    del gpu

    if hasattr(backend, "device_id"):
        kwargs["device_id"] = device_id

    return backend(**kwargs)

build_program 

build_program(
    network_path: Path,
    channels: int,
    tilesize: Shape,
    input_name: str = "input",
) -> Path
Source code in vsscale/mlrt/backend/migx.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def build_program(self, network_path: Path, channels: int, tilesize: Shape, input_name: str = "input") -> Path:
    if rocm_path := os.getenv("ROCM_PATH"):
        migraph_driver = f"{rocm_path}/bin/migraphx-driver"
    else:
        migraph_driver = shutil.which("migraphx-driver") or "migraphx-driver"

    identity = self.get_identity(network_path, tilesize)
    program_path = get_artifact_path(f"{identity}.mxr", fallback=not self.force_rebuild)

    if not self.force_rebuild and program_path.is_file() and program_path.stat().st_size >= 1024:
        return program_path

    program_path.parent.mkdir(parents=True, exist_ok=True)

    command: list[Any] = [
        migraph_driver,
        "compile",
        "--onnx",
        network_path,
        "--gpu",
        "--optimize",
        "--binary",
        "--output",
        program_path,
    ]

    opt_shapes = self.opt_shapes or tilesize
    command.extend(["--input-dim", f"@{input_name}", "1", f"{channels}", f"{opt_shapes[1]}", f"{opt_shapes[0]}"])
    if self.fp16:
        command.append("--fp16")
    if self.bf16:
        command.append("--bf16")
    if not self.fast_math:
        command.append("--disable-fast-math")
    if self.exhaustive_tune:
        command.append("--exhaustive-tune")

    command.extend(self.custom_args)

    logger.debug("%s: Calling migraphx-driver with the command:", self)
    logger.debug(command)

    try:
        subprocess.run(command, env=os.environ | self.custom_env, check=True, stdout=sys.stderr)
    except subprocess.CalledProcessError as e:
        logger.debug("MIGraphx STDERR:\n%s", e.stderr)
        logger.debug("MIGraphx STDOUT:\n%s", e.stdout)
        raise CustomRuntimeError("The program compilation failed") from e

    return program_path

get_args 

get_args(clips: VideoNode | Sequence[VideoNode]) -> dict[str, Any]

Return backend plugin arguments derived from this configuration.

Source code in vsscale/mlrt/backend/migx.py 
122
123
def get_args(self, clips: vs.VideoNode | Sequence[vs.VideoNode]) -> dict[str, Any]:
    return {"device_id": self.device_id, "num_streams": self.num_streams}

get_identity 

get_identity(network_path: Path, tilesize: Shape) -> int
Source code in vsscale/mlrt/backend/migx.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def get_identity(self, network_path: Path, tilesize: Shape) -> int:
    checksum = zlib.crc32(network_path.read_bytes())

    device_props = self.plugin.DeviceProperties(self.device_id)
    device = [device_props["name"].decode().replace(" ", "-"), device_props["driver_version"]]

    components = (
        str(self),
        str(self.version),
        network_path.name,
        f"{checksum:x}",
        str(tilesize),
        *map(str, device),
    )
    logger.debug("%s: Identity %r", self.get_identity, components)
    return zlib.crc32(bytes("|".join(components), "utf-8"))

inference 

inference(
    clips: VideoNode | Sequence[VideoNode],
    network_path: str | PathLike[str],
    /,
    overlap: tuple[int, int],
    tilesize: tuple[int, int],
    *,
    flexible: bool = False,
    **kwargs: Any,
) -> VideoNode | list[VideoNode]

Run inference with this backend.

Parameters:

  • clips

    (VideoNode | Sequence[VideoNode]) –

    Input clip or clips passed to the backend model.

  • network_path

    (str | PathLike[str]) –

    Path to the model file or backend artifact.

  • overlap

    (tuple[int, int]) –

    Horizontal and vertical tile overlap in pixels.

  • tilesize

    (tuple[int, int]) –

    Horizontal and vertical tile size in pixels.

  • flexible

    (bool, default: False ) –

    Return each flexible output plane as a separate clip.

  • **kwargs

    (Any, default: {} ) –

    Additional backend plugin arguments forwarded unchanged.

Returns:

  • VideoNode | list[VideoNode] –

    A single output clip, or a list of output clips when flexible is enabled.

Source code in vsscale/mlrt/backend/migx.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@copy_signature(BackendAutoConvertFloat.inference)
def inference(
    self,
    clips: vs.VideoNode | Sequence[vs.VideoNode],
    network_path: str | os.PathLike[str],
    /,
    overlap: tuple[int, int],
    tilesize: tuple[int, int],
    *,
    flexible: bool = False,
    **kwargs: Any,
) -> vs.VideoNode | list[vs.VideoNode]:
    UnsupportedSampleTypeError.check(clips, vs.FLOAT, self.__class__)

    clips = to_arr(clips)
    channels = sum(c.format.num_planes for c in clips)
    program_path = self.build_program(Path(network_path), channels, tilesize)
    return super().inference(clips, program_path, overlap, tilesize, flexible=flexible, **kwargs)