1 # ========================LICENSE_START=================================
4 # Copyright (c) 2019 AT&T Intellectual Property.
6 # Licensed under the Apache License, Version 2.0 (the "License");
7 # you may not use this file except in compliance with the License.
8 # You may obtain a copy of the License at
10 # http://www.apache.org/licenses/LICENSE-2.0
12 # Unless required by applicable law or agreed to in writing, software
13 # distributed under the License is distributed on an "AS IS" BASIS,
14 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
17 # ========================LICENSE_END===================================
19 # Adapter from RMR to standard Acumos model microservices. Must be deployed in the same pod as the Acumos model.
20 # Translates RMR protocol messages into calls into Acumos RPC calls.
31 requireartifacts = True
34 conffilename = 'config.json'
35 protobuffilename = 'model.proto'
36 metadatafilename = 'metadata.json'
38 configfilename = confdir + conffilename
41 print("Reading config file")
43 # Fetch and parse config file which must be mounted as a volume during deployment
45 with open(configfilename) as f:
48 print('Cannot read/parse config file at', configfilename, '; aborting')
51 methodurl = conf['microserviceRootURL'] + conf['methodRoot']
52 artifacturl = conf["microserviceRootURL"] + conf['artifactRoot']
55 print ('\nRetrieving artifacts from Acumos model microservice\n')
57 # See if we can retrieve protobuf and metadata artifacts from running model. Not all models may provide these, but we
58 # should have a retry mechanism added for robustness
60 r = requests.get(artifacturl + 'protobuf')
62 with open(confdir + protobuffilename, 'wb') as f:
66 print(protobuf.decode('ascii'))
67 r = requests.get(artifacturl + 'metadata')
69 with open(confdir + metadatafilename, 'wb') as f:
73 print(metadata.decode('ascii'))
76 print('Problem with retrieving/saving model protobuf and/or metadata; aborting.')
78 method1 = conf['methods']['1']
79 method1url = methodurl + method1['service']
80 method1headers = {'content-type': method1['content-type'], 'accept': method1['return-type']}
83 print('\nInitializing RMR\n')
86 print('\Awaiting connections')
89 # NNG cleanup on signal
90 def signal_handler(sig, frame):
92 print('SIGINT received! Cleaning up rmr')
99 mrc = rmr.rmr_init("4560".encode('utf-8'), rmr.RMR_MAX_RCV_BYTES, 0x00)
100 while rmr.rmr_ready(mrc) == 0:
103 print("Not yet ready")
104 rmr.rmr_set_stimeout(mrc, 2)
108 signal.signal(signal.SIGINT, signal_handler)
114 print("Waiting for a message; will time out after 2000ms")
115 sbuf = rmr.rmr_torcv_msg(mrc, sbuf, 2000)
116 summary = rmr.message_summary(sbuf)
117 if verbose and summary['message state'] == 12:
118 print("Nothing received.")
121 print("Message received: {}".format(summary))
122 payload = sbuf['payload']
123 # Call Acumos microservice
124 r = requests.post(method1url, headers=method1headers, body=payload)
126 rmr.set_payload_and_length(val, sbuf)
127 sbuf = rmr.rmr_rts_msg(mrc, sbuf)