1import logging
2
3from omegaml.runtimes.proxies.baseproxy import RuntimeProxyBase
4
5logger = logging.getLogger(__name__)
6
7
[docs]
8class OmegaModelProxy(RuntimeProxyBase):
9 """
10 proxy to a remote model in a celery worker
11
12 The proxy provides the same methods as the model but will
13 execute the methods using celery tasks and return celery
14 AsyncResult objects
15
16 Usage::
17
18 om = Omega()
19 # train a model
20 # result is AsyncResult, use .get() to return it's result
21 result = om.runtime.model('foo').fit('datax', 'datay')
22 result.get()
23
24 # predict
25 result = om.runtime.model('foo').predict('datax')
26 # result is AsyncResult, use .get() to return it's result
27 print result.get()
28
29 Notes:
30 The actual methods of ModelProxy are defined in its mixins
31
32 See Also:
33 * ModelMixin
34 * GridSearchMixin
35 """
36
37 # Implementation note:
38 #
39 # We decided to implement each method call explicitely in both
40 # this class (mixins) and the celery tasks. While it would be possible to
41 # implement a generic method and task that passes the method and
42 # arguments to be called, maintainability would suffer and the
43 # documentation would become very unspecific. We think it is much
44 # cleaner to have an explicit interface at the chance of missing
45 # features. If need should arise we can still implement a generic
46 # method call.
47
48 def __init__(self, modelname, runtime=None):
49 super().__init__(modelname, runtime=runtime, store=runtime.omega.models)
50 self.modelname = modelname
51
[docs]
52 def task(self, name):
53 """
54 return the task from the runtime with requirements applied
55 """
56 return self.runtime.task(name)
57
[docs]
58 def experiment(self, experiment=None, label=None, provider=None, **tracker_kwargs):
59 """return the experiment for this model
60
61 If an experiment does not exist yet, it will be created. The
62 experiment is automatically set to track this model, unless another
63 experiment has already been set to track this model for the same label.
64 If a previous model has been set to track this model it will be
65 returned. If an experiment name is passed it will be used.
66
67 Args:
68 experiment (str): the experiment name, defaults to the modelname
69 label (str): the runtime label, defaults to 'default'
70 provider (str): the provider to use, defaults to 'default'
71 tracker_kwargs (dict): additional kwargs to pass to the tracker
72
73 Returns:
74 OmegaTrackingProxy() instance
75 """
76 label = label or self.runtime._default_label or 'default'
77 label = self.runtime._default_label if label in ('default', 'local') else label
78 exps = self.experiments(label=label) if experiment is None else None
79 exp = exps.get(label) if exps else None
80 experiment = experiment or self.modelname
81 if exp is None:
82 exp = self.runtime.experiment(experiment, provider=provider, **tracker_kwargs)
83 if not label in self.experiments():
84 exp.track(self.modelname, label=label)
85 return exp
86
[docs]
87 def experiments(self, label=None, raw=False):
88 """return list of experiments tracking this model
89
90 Args:
91 label (None|str): the label for which to return the experiments, or None for all
92 raw (bool): if True return the metadata for the experiment, else return the OmegaTrackingProxy
93
94 Returns:
95 instances (dict): mapping of label => instance of OmegaTrackingProxy if not raw, else Metadata,
96 includes a dummy label '_all_', listing all experiments that track this model.
97
98 .. versionchanged:: 0.17
99 returns a dict instead of a list
100 """
101 store = self.store
102 tracking = store.metadata(self.modelname).attributes.get('tracking', {})
103 by_label = {
104 label: self.runtime.experiment(name) if not raw else store.metadata(f'experiments/{name}')
105 for label, name in tracking.items()
106 if label not in ['experiments', 'monitors']
107 }
108 unlabeled = {
109 '_all_': [
110 self.runtime.experiment(name) if not raw else store.metadata(f'experiments/{name}')
111 for name in tracking.get('experiments', [])
112 ]
113 }
114 all_exps = dict(**by_label, **unlabeled)
115 return {k: v for k, v in all_exps.items() if not label or k == label}
116
[docs]
117 def monitor(self, label=None, provider=None, **tracker_kwargs):
118 """return the monitor for this model
119
120 Returns:
121 OmegaMonitorProxy() instance
122 """
123 return self.experiment(label=label, provider=provider, **tracker_kwargs).as_monitor(self.modelname)