1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
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
|
# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------
# pylint: disable=redefined-builtin
import re
from typing import Any, Dict, Optional, overload
from typing_extensions import Literal
from azure.ai.ml.constants import AssetTypes
from azure.ai.ml.constants._component import IOConstants
from azure.ai.ml.entities._assets.intellectual_property import IntellectualProperty
from azure.ai.ml.exceptions import UserErrorException
from .base import _InputOutputBase
from .utils import _remove_empty_values
class Output(_InputOutputBase):
_IO_KEYS = ["name", "version", "path", "path_on_compute", "type", "mode", "description", "early_available"]
@overload
def __init__(
self,
*,
type: str,
path: Optional[str] = None,
mode: Optional[str] = None,
description: Optional[str] = None,
**kwargs: Any,
): ...
@overload
def __init__(
self,
type: Literal["uri_file"] = "uri_file",
path: Optional[str] = None,
mode: Optional[str] = None,
description: Optional[str] = None,
):
"""Define a URI file output.
:keyword type: The type of the data output. Can only be set to 'uri_file'.
:paramtype type: str
:keyword path: The remote path where the output should be stored.
:paramtype path: str
:keyword mode: The access mode of the data output. Accepted values are
* 'rw_mount': Read-write mount the data,
* 'upload': Upload the data from the compute target,
* 'direct': Pass in the URI as a string
:paramtype mode: str
:keyword description: The description of the output.
:paramtype description: str
:keyword name: The name to be used to register the output as a Data or Model asset. A name can be set without
setting a version.
:paramtype name: str
:keyword version: The version used to register the output as a Data or Model asset. A version can be set only
when name is set.
:paramtype version: str
"""
def __init__( # type: ignore[misc]
self,
*,
type: str = AssetTypes.URI_FOLDER,
path: Optional[str] = None,
mode: Optional[str] = None,
description: Optional[str] = None,
**kwargs: Any,
) -> None:
"""Define an output.
:keyword type: The type of the data output. Accepted values are 'uri_folder', 'uri_file', 'mltable',
'mlflow_model', 'custom_model', and user-defined types. Defaults to 'uri_folder'.
:paramtype type: str
:keyword path: The remote path where the output should be stored.
:paramtype path: Optional[str]
:keyword mode: The access mode of the data output. Accepted values are
* 'rw_mount': Read-write mount the data
* 'upload': Upload the data from the compute target
* 'direct': Pass in the URI as a string
:paramtype mode: Optional[str]
:keyword path_on_compute: The access path of the data output for compute
:paramtype path_on_compute: Optional[str]
:keyword description: The description of the output.
:paramtype description: Optional[str]
:keyword name: The name to be used to register the output as a Data or Model asset. A name can be set without
setting a version.
:paramtype name: str
:keyword version: The version used to register the output as a Data or Model asset. A version can be set only
when name is set.
:paramtype version: str
:keyword is_control: Determine if the output is a control output.
:paramtype is_control: bool
:keyword early_available: Mark the output for early node orchestration.
:paramtype early_available: bool
:keyword intellectual_property: Intellectual property associated with the output.
It can be an instance of `IntellectualProperty` or a dictionary that will be used to create an instance.
:paramtype intellectual_property: Union[
~azure.ai.ml.entities._assets.intellectual_property.IntellectualProperty, dict]
.. admonition:: Example:
.. literalinclude:: ../samples/ml_samples_misc.py
:start-after: [START create_inputs_outputs]
:end-before: [END create_inputs_outputs]
:language: python
:dedent: 8
:caption: Creating a CommandJob with a folder output.
"""
super(Output, self).__init__(type=type)
# As an annotation, it is not allowed to initialize the _port_name.
self._port_name = None
self.name = kwargs.pop("name", None)
self.version = kwargs.pop("version", None)
self._is_primitive_type = self.type in IOConstants.PRIMITIVE_STR_2_TYPE
self.description = description
self.path = path
self.path_on_compute = kwargs.pop("path_on_compute", None)
self.mode = mode
# use this field to mark Output for early node orchestrate, currently hide in kwargs
self.early_available = kwargs.pop("early_available", None)
self._intellectual_property = None
intellectual_property = kwargs.pop("intellectual_property", None)
if intellectual_property:
self._intellectual_property = (
intellectual_property
if isinstance(intellectual_property, IntellectualProperty)
else IntellectualProperty(**intellectual_property)
)
self._assert_name_and_version()
# normalize properties
self._normalize_self_properties()
def _get_hint(self, new_line_style: bool = False) -> Optional[str]:
comment_str = self.description.replace('"', '\\"') if self.description else self.type
return '"""%s"""' % comment_str if comment_str and new_line_style else comment_str
def _to_dict(self) -> Dict:
"""Convert the Output object to a dict.
:return: The dictionary representation of Output
:rtype: Dict
"""
keys = self._IO_KEYS
result = {key: getattr(self, key) for key in keys}
res: dict = _remove_empty_values(result)
return res
def _to_rest_object(self) -> Dict:
# this is for component rest object when using Output as component outputs, as for job output usage,
# rest object is generated by extracting Output's properties, see details in to_rest_data_outputs()
return self._to_dict()
def _simple_parse(self, value: Any, _type: Any = None) -> Any:
if _type is None:
_type = self.type
if _type in IOConstants.PARAM_PARSERS:
return IOConstants.PARAM_PARSERS[_type](value)
return value
def _normalize_self_properties(self) -> None:
# parse value from string to its original type. eg: "false" -> False
if self.early_available:
self.early_available = self._simple_parse(getattr(self, "early_available", "false"), _type="boolean")
@classmethod
def _from_rest_object(cls, obj: Dict) -> "Output":
# this is for component rest object when using Output as component outputs
return Output(**obj)
def _assert_name_and_version(self) -> None:
if self.name and not (re.match("^[A-Za-z0-9_-]*$", self.name) and len(self.name) <= 255):
raise UserErrorException(
f"The output name {self.name} can only contain alphanumeric characters, dashes and underscores, "
f"with a limit of 255 characters."
)
if self.version and not self.name:
raise UserErrorException("Output name is required when output version is specified.")
|
