forked from microsoft/MLOpsPython
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmodel_helper.py
More file actions
91 lines (76 loc) · 2.66 KB
/
model_helper.py
File metadata and controls
91 lines (76 loc) · 2.66 KB
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
"""
model_helper.py
"""
from azureml.core import Run
from azureml.core import Workspace
from azureml.core.model import Model as AMLModel
def get_current_workspace() -> Workspace:
"""
Retrieves and returns the latest model from the workspace
by its name and tag. Will not work when ran locally.
Parameters:
None
Return:
The current workspace.
"""
run = Run.get_context(allow_offline=False)
experiment = run.experiment
return experiment.workspace
def _get_model_by_tag(
model_name: str,
tag_name: str,
tag_value: str,
aml_workspace: Workspace = None
) -> AMLModel:
"""
Retrieves and returns the latest model from the workspace
by its name and tag.
Parameters:
aml_workspace (Workspace): aml.core Workspace that the model lives.
model_name (str): name of the model we are looking for
tag (str): the tag value the model was registered under.
Return:
A single aml model from the workspace that matches the name and tag.
"""
# Validate params. cannot be None.
if model_name is None:
raise ValueError("model_name[:str] is required")
if tag_name is None:
raise ValueError("tag_name[:str] is required")
if tag_value is None:
raise ValueError("tag[:str] is required")
if aml_workspace is None:
aml_workspace = get_current_workspace()
# get model by tag.
model_list = AMLModel.list(
aml_workspace, name=model_name,
tags=[[tag_name, tag_value]], latest=True
)
# latest should only return 1 model, but if it does, then maybe
# internal code was accidentally changed or the source code has changed.
should_not_happen = ("THIS SHOULD NOT HAPPEN: found more than one model "
"for the latest with {{tag_name: {tag_name},"
"tag_value: {tag_value}. "
"Models found: {model_list}}}")\
.format(model_name=model_name, tag_name=tag_value,
model_list=model_list)
if len(model_list) > 1:
raise ValueError(should_not_happen)
return model_list
def get_model_by_tag(
model_name: str,
tag_name: str,
tag_value: str,
aml_workspace: Workspace = None
) -> AMLModel:
"""
Wrapper function for get_model_by_tag that throws an error if model is none
"""
model_list = _get_model_by_tag(
model_name, tag_name, tag_value, aml_workspace)
if model_list:
return model_list[0]
no_model_found = ("Model not found with model_name: {model_name}"
"tag {tag_name} {tag_value}")\
.format(model_name=model_name, tag_name=tag_value)
raise Exception(no_model_found)