AcumosXappAdapter/rmracumosadapter.py

   1 # ========================LICENSE_START=================================
   2 #   O-RAN-SC
   3 #   %%
   4 #   Copyright (c) 2019 AT&T Intellectual Property.
   5 #   %%
   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
   9 #
  10 #          http://www.apache.org/licenses/LICENSE-2.0
  11 #
  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===================================
  18
  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.
  21
  22
  23 from rmr import rmr
  24 import time
  25 import sys
  26 import signal
  27 import json
  28 import requests
  29
  30 verbose = True
  31 requireartifacts = True
  32
  33 confdir = '/conf/'
  34 conffilename = 'config.json'
  35 protobuffilename = 'model.proto'
  36 metadatafilename = 'metadata.json'
  37
  38 configfilename = confdir + conffilename
  39
  40 if verbose:
  41     print("Reading config file")
  42
  43 # Fetch and parse config file which must be mounted as a volume during deployment
  44 try:
  45     with open(configfilename) as f:
  46         conf = json.load(f)
  47 except:
  48     print('Cannot read/parse config file at', configfilename, '; aborting')
  49     exit(1)
  50
  51 methodurl = conf['microserviceRootURL'] + conf['methodRoot']
  52 artifacturl = conf["microserviceRootURL"] + conf['artifactRoot']
  53
  54 if verbose:
  55     print ('\nRetrieving artifacts from Acumos model microservice\n')
  56
  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
  59 try:
  60     r = requests.get(artifacturl + 'protobuf')
  61     protobuf = r.content
  62     with open(confdir + protobuffilename, 'wb') as f:
  63         f.write(protobuf)
  64     if verbose:
  65         print('Protbuf:')
  66         print(protobuf.decode('ascii'))
  67     r = requests.get(artifacturl + 'metadata')
  68     metadata = r.content
  69     with open(confdir + metadatafilename, 'wb') as f:
  70         f.write(metadata)
  71     if verbose:
  72         print('\nMetadata:')
  73         print(metadata.decode('ascii'))
  74 except:
  75     if requireartifacts:
  76         print('Problem with retrieving/saving model protobuf and/or metadata; aborting.')
  77
  78 method1 = conf['methods']['1']
  79 method1url = methodurl + method1['service']
  80 method1headers = {'content-type': method1['content-type'], 'accept': method1['return-type']}
  81
  82 if verbose:
  83     print('\nInitializing RMR\n')
  84
  85 if verbose:
  86     print('\Awaiting connections')
  87
  88
  89 # NNG cleanup on signal
  90 def signal_handler(sig, frame):
  91     if verbose:
  92         print('SIGINT received! Cleaning up rmr')
  93     rmr.rmr_close(mrc)
  94     print("Exiting")
  95     sys.exit(0)
  96
  97
  98 # Initialize RMR
  99 mrc = rmr.rmr_init("4560".encode('utf-8'), rmr.RMR_MAX_RCV_BYTES, 0x00)
 100 while rmr.rmr_ready(mrc) == 0:
 101     time.sleep(1)
 102     if verbose:
 103         print("Not yet ready")
 104 rmr.rmr_set_stimeout(mrc, 2)
 105
 106
 107 # Capture ctrl-c
 108 signal.signal(signal.SIGINT, signal_handler)
 109
 110
 111 sbuf = None
 112 while True:
 113     if verbose:
 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.")
 119     else:
 120         if verbose:
 121             print("Message received: {}".format(summary))
 122         payload = sbuf['payload']
 123         # Call Acumos microservice
 124         r = requests.post(method1url, headers=method1headers, body=payload)
 125         val = r.content
 126         rmr.set_payload_and_length(val, sbuf)
 127         sbuf = rmr.rmr_rts_msg(mrc, sbuf)