1 # ============LICENSE_START====================================================
3 # =============================================================================
4 # Copyright (c) 2019-2020 AT&T Intellectual Property. All rights reserved.
5 # =============================================================================
6 # Copyright (c) 2021 highstreet technologies GmbH. All rights reserved.
7 # =============================================================================
8 # Licensed under the Apache License, Version 2.0 (the "License");
9 # you may not use this file except in compliance with the License.
10 # You may obtain a copy of the License at
12 # http://www.apache.org/licenses/LICENSE-2.0
14 # Unless required by applicable law or agreed to in writing, software
15 # distributed under the License is distributed on an "AS IS" BASIS,
16 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 # See the License for the specific language governing permissions and
18 # limitations under the License.
19 # ============LICENSE_END======================================================
23 from pkg_resources import resource_string
27 from urllib.parse import unquote_plus
28 from socketserver import ThreadingMixIn
29 from http.server import BaseHTTPRequestHandler, HTTPServer
31 from urllib import unquote_plus
32 from SocketServer import ThreadingMixIn
33 from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
37 from aoconversion import convert
40 def _derefconfig(value):
41 if value.startswith('@'):
42 with open(value[1:], 'r') as f:
43 return f.readline().strip()
49 Configuration parameters as attributes, make sure the required ones are there,
52 def __init__(self, dcaeuser, onboardingurl, onboardinguser, onboardingpass, certfile, dockerregistry, dockeruser, dockerpass, http_proxy, https_proxy, no_proxy, acumosurl=None, interval=900, dockerhost='tcp://localhost:2375', tmpdir='/var/tmp/aoadapter', certverify=True, catalogs=None, port=None, **extras):
53 self.dcaeuser = dcaeuser
55 def x(fmt, *args, **kwargs):
56 return onboardingurl + fmt.format(*args, **kwargs)
58 self._onboardingpass = onboardingpass
59 self._onboardinguser = onboardinguser
60 self.acumosurl = acumosurl
61 self.certfile = certfile
62 self.certverify = certverify
63 self.dockerhost = dockerhost
64 self.dockerregistry = dockerregistry
65 self.dockeruser = dockeruser
66 self._dockerpass = dockerpass
67 self.interval = interval
69 self.http_proxy = http_proxy if http_proxy is not None else ""
70 self.https_proxy = https_proxy if http_proxy is not None else ""
71 self.no_proxy = no_proxy if no_proxy is not None else ""
72 if catalogs is not None and type(catalogs) is not list:
74 self.catalogs = catalogs
78 return (self._onboardinguser, _derefconfig(self._onboardingpass))
81 return _derefconfig(self._dockerpass)
84 class _AcumosAccess(object):
85 def __init__(self, config, url):
86 self.cert = config.certfile
87 self.verify = config.certverify
88 self.url = url.strip().rstrip('/')
90 def artgetter(self, xrev, matcher):
91 for art in xrev['artifacts']:
94 nurl = '{}/artifacts/{}/content'.format(self.url, art['artifactId'])
95 resp = requests.get(nurl, stream=True, cert=self.cert, verify=self.verify)
96 if resp.status_code == 500:
97 resp = requests.get(nurl, stream=True, cert=self.cert, verify=self.verify)
98 resp.raise_for_status()
103 def jsonget(self, format, *args, **kwargs):
104 nurl = self.url + format.format(*args, **kwargs)
105 resp = requests.get(nurl, cert=self.cert, verify=self.verify)
106 if resp.status_code == 500:
107 resp = requests.get(nurl, cert=self.cert, verify=self.verify)
108 resp.raise_for_status()
109 return resp.json()['content']
112 def _x_proto_matcher(art):
113 """ Is this artifact the x.proto file? """
114 return art['name'].endswith('.proto')
117 def _x_zip_matcher(art):
118 """ Is this artifact the x.zip file? """
119 return art['name'].endswith('.zip')
122 def _md_json_matcher(art):
123 """ Is this artifact the metadata.json file? """
124 return art['name'].__contains__('metadata') & art['name'].endswith('.json')
129 Walk the Federation E5 interface of an Acumos instance
131 url = config.acumosurl
132 callback = _makecallback(config)
133 catalogs = config.catalogs
134 aa = _AcumosAccess(config, url)
135 for catalog in aa.jsonget('/catalogs'):
136 if catalogs is not None and catalog['catalogId'] not in catalogs and catalog['name'] not in catalogs:
138 for solution in aa.jsonget('/solutions?catalogId={}', catalog['catalogId']):
139 for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
140 onboard(aa, callback, solution, revision['revisionId'])
143 def onboard(aa, callback, solution, revid):
145 xrev = aa.jsonget('/solutions/{}/revisions/{}', solution['solutionId'], revid)
146 callback(model_name=solution['name'], model_version=xrev['version'], model_last_updated=xrev['modified'], rating=solution['ratingAverageTenths'] / 10.0, proto_getter=aa.artgetter(xrev, _x_proto_matcher), zip_getter=aa.artgetter(xrev, _x_zip_matcher), metadata_getter=aa.artgetter(xrev, _md_json_matcher))
149 def _pullfile(source, dest):
150 with open(dest, 'wb') as f:
151 for chunk in source().iter_content(65536):
155 _loadedformats = set()
156 _loadedcomponents = set()
163 def _makecallback(config):
164 workdir = config.tmpdir
165 obauth = config.obauth()
168 def callback(model_name, model_version, model_last_updated, rating, proto_getter, zip_getter, metadata_getter):
169 model_name = model_name.lower()
170 model_version = model_version.lower()
171 compid = (model_name, model_version)
172 if compid in _loadedcomponents:
173 print('Skipping component {}: already analyzed'.format(compid))
175 if proto_getter is None or zip_getter is None or metadata_getter is None:
176 print('Skipping component {}: does not have required artifacts'.format(compid))
177 _loadedcomponents.add(compid)
179 modeldir = '{}/{}'.format(workdir, model_name)
180 shutil.rmtree(modeldir, True)
181 os.makedirs(modeldir)
183 _pullfile(proto_getter, '{}/model.proto'.format(modeldir))
184 _pullfile(zip_getter, '{}/model.zip'.format(modeldir))
185 _pullfile(metadata_getter, '{}/metadata.json'.format(modeldir))
187 print('Skipping component {}: artifact access error'.format(compid))
188 _loadedcomponents.add(compid)
191 docker_uri, data_formats, spec = convert.gen_dcae_artifacts_for_model(config, model_name, model_version)
192 shutil.rmtree(modeldir)
194 print('Error analyzing artifacts for {}'.format(compid))
195 traceback.print_exc()
197 for data_format in data_formats:
198 fmtid = (data_format['self']['name'], data_format['self']['version'])
199 if fmtid in _loadedformats:
200 print('Skipping data format {}: already analyzed'.format(fmtid))
203 resp = requests.post(oburl('/dataformats'), json={'owner': config.dcaeuser, 'spec': data_format}, auth=obauth)
204 if resp.status_code == 409:
205 print('Skipping data format {}: previously loaded'.format(fmtid))
206 _loadedformats.add(fmtid)
208 resp.raise_for_status()
209 requests.patch(resp.json()['dataFormatUrl'], json={'owner': config.dcaeuser, 'status': 'published'}, auth=obauth).raise_for_status()
210 print('Loaded data format {}'.format(fmtid))
211 _loadedformats.add(fmtid)
213 print('Error loading data format {}'.format(fmtid))
214 traceback.print_exc()
217 resp = requests.post(oburl('/components'), json={'owner': config.dcaeuser, 'spec': spec}, auth=obauth)
218 if resp.status_code == 409:
219 print('Skipping component {}: previously loaded'.format(compid))
220 _loadedcomponents.add(compid)
222 resp.raise_for_status()
223 requests.patch(resp.json()['componentUrl'], json={'owner': config.dcaeuser, 'status': 'published'}, auth=obauth).raise_for_status()
224 print('Loaded component {}'.format(compid))
225 _loadedcomponents.add(compid)
227 print('Error loading component {}'.format(compid))
228 traceback.print_exc()
233 class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
234 protocol_version = "HTTP/1.1"
237 class Apihandler(BaseHTTPRequestHandler):
240 if not self.path or '?' not in self.path:
242 self.path, qp = self.path.split('?', 1)
243 for x in qp.split('&'):
244 k, v = x.split('=', 1)
245 self.qparams[unquote_plus(k)] = unquote_plus(v)
247 def replyjson(self, body, ctype='application/json'):
248 self.replyraw(json.dumps(body).encode('utf-8'), ctype)
250 def replyraw(self, data, ctype):
251 self.send_response(200)
252 self.send_header('Content-Type', ctype)
253 self.send_header('Content-Length', len(data))
255 self.wfile.write(data)
259 if self.path == '/' or self.path == '/index.html' or self.path == '/acumos-adapter/' or self.path == '/acumos-adapter/index.html':
260 self.replyraw(self.server.index, 'text/html')
262 if 'acumos' not in self.qparams:
265 aa = _AcumosAccess(self.server.config, self.qparams['acumos'])
266 if self.path == '/acumos-adapter/listCatalogs.js':
267 self.replyjson([{'name': x['name'], 'id': x['catalogId']} for x in aa.jsonget('/catalogs')])
269 if self.path == '/acumos-adapter/listSolutions.js':
270 if 'catalogId' not in self.qparams:
273 self.replyjson([{'name': x['name'], 'id': x['solutionId']} for x in aa.jsonget('/solutions?catalogId={}', self.qparams['catalogId'])])
275 if self.path == '/acumos-adapter/listRevisions.js':
276 if 'solutionId' not in self.qparams:
279 self.replyjson([{'name': x['version'], 'id': x['revisionId']} for x in aa.jsonget('/solutions/{}/revisions', self.qparams['solutionId'])])
285 if self.path == '/acumos-adapter/onboard.js':
286 if 'acumos' not in self.qparams:
289 aa = _AcumosAccess(self.server.config, self.qparams['acumos'])
290 callback = self.server.callback
291 if 'catalogId' not in self.qparams:
292 for catalog in aa.jsonget('/catalogs'):
293 for solution in aa.jsonget('/solutions?catalogId={}', catalog['catalogId']):
294 for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
295 onboard(aa, callback, solution, revision['revisionId'])
296 elif 'solutionId' not in self.qparams:
297 for solution in aa.jsonget('/solutions?catalogId={}', self.qparams['catalogId']):
298 for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
299 onboard(aa, callback, solution, revision['revisionId'])
300 elif 'revisionId' not in self.qparams:
301 solution = aa.jsonget('/solutions/{}', self.qparams['solutionId'])
302 for revision in aa.jsonget('/solutions/{}/revisions', solution['solutionId']):
303 onboard(aa, callback, solution, revision['revisionId'])
305 solution = aa.jsonget('/solutions/{}', self.qparams['solutionId'])
306 onboard(aa, callback, solution, self.qparams['revisionId'])
307 self.replyraw('OK'.encode('utf-8'), 'text/plain')
313 server = ThreadedHTTPServer(('', config.port), Apihandler)
314 server.config = config
315 server.callback = _makecallback(config)
316 server.index = resource_string(__name__, 'index.html')
317 server.serve_forever()