-
Notifications
You must be signed in to change notification settings - Fork 1
/
cli.py
140 lines (112 loc) · 3.95 KB
/
cli.py
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
import collections
import inspect
import typer
from pathlib import Path
from rich.console import Console
import subprocess
app = typer.Typer(name="remote")
console = Console()
@app.callback()
def remote_main():
console.print(
"""
📶 Connect multiple AutoGOAL instances and share algorithms.
"""
)
@app.command("connect")
def remote_connect(
ip: str = typer.Argument(None, help="Interface ip to be used by the HTTP API"),
port: int = typer.Argument(None, help="Port to be bind by the server"),
connection_alias: int = typer.Argument(
None,
help="Connection alias for future references to the remote AutoGOAL instance",
),
verbose: bool = False,
):
try:
import autogoal_remote as rm_server
except:
raise Exception("autogoal-remote not installed")
try:
from autogoal_contrib import find_remote_classes
except:
raise Exception("autogoal-contrib not installed")
from autogoal_contrib import (
find_remote_classes,
)
"""
📡 Connect to an existing AutoGOAL instance.
"""
# try connection and request algorithms
sources = []
if ip is None or port is None:
if connection_alias is not None:
sources.append(connection_alias)
else:
if connection_alias is not None:
sources.append((ip, port, connection_alias))
else:
sources.append((ip, port))
classes = find_remote_classes(sources)
typer.echo(f"⚙️ Successfully connected to remote AutoGOAL!", color="green")
if connection_alias:
typer.echo(
f"⚙️ Stored connection to {ip}:{port} with alias '{connection_alias}' for future usage.",
color="blue",
)
classes_by_contrib = collections.defaultdict(list)
max_cls_name_length = 0
for cls in classes:
classes_by_contrib[cls.contrib].append(cls)
typer.echo(
f"⚙️ Found a total of {len(classes)} matching remote algorithms.", color="blue"
)
for contrib, clss in classes_by_contrib.items():
typer.echo(f"🛠️ {connection_alias} -> {contrib}: {len(clss)} algorithms.")
if verbose:
for cls in clss:
sig = inspect.signature(cls.run)
typer.echo(
f" 🔹 {cls.__name__.ljust(max_cls_name_length)} : {sig.parameters['input'].annotation} -> {sig.return_annotation}"
)
@app.command("share-contribs")
def share_contribs(
ip: str = typer.Argument(
"0.0.0.0", help="Interface ip of listening AutoGOAL service"
),
port: int = typer.Argument(8000, help="Port of listening AutoGOAL service"),
):
"""
Expose algorithms from installed contribs to other AutoGOAL instances over the network.
"""
try:
import autogoal_remote as rm_server
except:
raise Exception("autogoal-remote installation not detected")
rm_server.distributed.run(ip, port)
@app.command("serve")
def automl_server(
path: str = typer.Argument(None, help="Autogoal serialized model"),
ip: str = typer.Argument("0.0.0.0", help="Interface ip to be used by the HTTP API"),
port: int = typer.Argument(8000, help="Port to be bind by the server"),
):
"""
Load and serve a previously trained AutoML instance as a service.
"""
from autogoal.ml import AutoML
from autogoal_remote.production import run
import os
default_path = Path(os.getcwd()) / "autogoal-export"
console.print(f"Loading model from folder: {path or default_path}")
model = AutoML.folder_load(Path(path or default_path))
run(model, ip, port)
global typer_app
typer_app = app
if __name__ == "__main__":
from autogoal.ml import AutoML
from autogoal_remote.production import run
import os
default_path = Path(os.getcwd()) / "autogoal-export"
console.print(f"Loading model from folder: {default_path}")
model = AutoML.folder_load(Path(default_path))
run(model, "0.0.0.0", 8000)