|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from dataclasses import dataclass, field |
| 4 | +from typing import Dict, List, Optional, Tuple, Type, Union |
| 5 | + |
| 6 | +import torch |
| 7 | +from dataclasses_json import dataclass_json |
| 8 | +from torch.onnx import OperatorExportTypes, TrainingMode |
| 9 | +from typing_extensions import Annotated, get_args, get_origin |
| 10 | + |
| 11 | +from flytekit import FlyteContext |
| 12 | +from flytekit.core.type_engine import TypeEngine, TypeTransformer, TypeTransformerFailedError |
| 13 | +from flytekit.models.core.types import BlobType |
| 14 | +from flytekit.models.literals import Blob, BlobMetadata, Literal, Scalar |
| 15 | +from flytekit.models.types import LiteralType |
| 16 | +from flytekit.types.file import ONNXFile |
| 17 | + |
| 18 | + |
| 19 | +@dataclass_json |
| 20 | +@dataclass |
| 21 | +class PyTorch2ONNXConfig: |
| 22 | + args: Union[Tuple, torch.Tensor] |
| 23 | + export_params: bool = True |
| 24 | + verbose: bool = False |
| 25 | + training: TrainingMode = TrainingMode.EVAL |
| 26 | + opset_version: int = 9 |
| 27 | + input_names: List[str] = field(default_factory=list) |
| 28 | + output_names: List[str] = field(default_factory=list) |
| 29 | + operator_export_type: Optional[OperatorExportTypes] = None |
| 30 | + do_constant_folding: bool = False |
| 31 | + dynamic_axes: Union[Dict[str, Dict[int, str]], Dict[str, List[int]]] = field(default_factory=dict) |
| 32 | + keep_initializers_as_inputs: Optional[bool] = None |
| 33 | + custom_opsets: Dict[str, int] = field(default_factory=dict) |
| 34 | + export_modules_as_functions: Union[bool, set[Type]] = False |
| 35 | + |
| 36 | + |
| 37 | +@dataclass_json |
| 38 | +@dataclass |
| 39 | +class PyTorch2ONNX: |
| 40 | + model: Union[torch.nn.Module, torch.jit.ScriptModule, torch.jit.ScriptFunction] = field(default=None) |
| 41 | + |
| 42 | + |
| 43 | +def extract_config(t: Type[PyTorch2ONNX]) -> Tuple[Type[PyTorch2ONNX], PyTorch2ONNXConfig]: |
| 44 | + config = None |
| 45 | + if get_origin(t) is Annotated: |
| 46 | + base_type, config = get_args(t) |
| 47 | + if isinstance(config, PyTorch2ONNXConfig): |
| 48 | + return base_type, config |
| 49 | + else: |
| 50 | + raise TypeTransformerFailedError(f"{t}'s config isn't of type PyTorch2ONNXConfig") |
| 51 | + return t, config |
| 52 | + |
| 53 | + |
| 54 | +def to_onnx(ctx, model, config): |
| 55 | + local_path = ctx.file_access.get_random_local_path() |
| 56 | + |
| 57 | + torch.onnx.export( |
| 58 | + model, |
| 59 | + **config, |
| 60 | + f=local_path, |
| 61 | + ) |
| 62 | + |
| 63 | + return local_path |
| 64 | + |
| 65 | + |
| 66 | +class PyTorch2ONNXTransformer(TypeTransformer[PyTorch2ONNX]): |
| 67 | + ONNX_FORMAT = "onnx" |
| 68 | + |
| 69 | + def __init__(self): |
| 70 | + super().__init__(name="PyTorch ONNX", t=PyTorch2ONNX) |
| 71 | + |
| 72 | + def get_literal_type(self, t: Type[PyTorch2ONNX]) -> LiteralType: |
| 73 | + return LiteralType(blob=BlobType(format=self.ONNX_FORMAT, dimensionality=BlobType.BlobDimensionality.SINGLE)) |
| 74 | + |
| 75 | + def to_literal( |
| 76 | + self, |
| 77 | + ctx: FlyteContext, |
| 78 | + python_val: PyTorch2ONNX, |
| 79 | + python_type: Type[PyTorch2ONNX], |
| 80 | + expected: LiteralType, |
| 81 | + ) -> Literal: |
| 82 | + python_type, config = extract_config(python_type) |
| 83 | + |
| 84 | + if config: |
| 85 | + local_path = to_onnx(ctx, python_val.model, config.__dict__.copy()) |
| 86 | + remote_path = ctx.file_access.get_random_remote_path() |
| 87 | + ctx.file_access.put_data(local_path, remote_path, is_multipart=False) |
| 88 | + else: |
| 89 | + raise TypeTransformerFailedError(f"{python_type}'s config is None") |
| 90 | + |
| 91 | + return Literal( |
| 92 | + scalar=Scalar( |
| 93 | + blob=Blob( |
| 94 | + uri=remote_path, |
| 95 | + metadata=BlobMetadata( |
| 96 | + type=BlobType(format=self.ONNX_FORMAT, dimensionality=BlobType.BlobDimensionality.SINGLE) |
| 97 | + ), |
| 98 | + ) |
| 99 | + ) |
| 100 | + ) |
| 101 | + |
| 102 | + def to_python_value( |
| 103 | + self, |
| 104 | + ctx: FlyteContext, |
| 105 | + lv: Literal, |
| 106 | + expected_python_type: Type[ONNXFile], |
| 107 | + ) -> ONNXFile: |
| 108 | + if not (lv.scalar.blob.uri and lv.scalar.blob.metadata.format == self.ONNX_FORMAT): |
| 109 | + raise TypeTransformerFailedError(f"ONNX format isn't of the expected type {expected_python_type}") |
| 110 | + |
| 111 | + return ONNXFile(path=lv.scalar.blob.uri) |
| 112 | + |
| 113 | + def guess_python_type(self, literal_type: LiteralType) -> Type[PyTorch2ONNX]: |
| 114 | + if ( |
| 115 | + literal_type.blob is not None |
| 116 | + and literal_type.blob.dimensionality == BlobType.BlobDimensionality.SINGLE |
| 117 | + and literal_type.blob.format == self.ONNX_FORMAT |
| 118 | + ): |
| 119 | + return PyTorch2ONNX |
| 120 | + |
| 121 | + raise TypeTransformerFailedError(f"Transformer {self} cannot reverse {literal_type}") |
| 122 | + |
| 123 | + |
| 124 | +TypeEngine.register(PyTorch2ONNXTransformer()) |
0 commit comments