2 # ==================================================================================
3 # Copyright (c) 2022 Nokia
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
9 # http://www.apache.org/licenses/LICENSE-2.0
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 # ==================================================================================
31 from time import gmtime, strftime
33 #sys.path.insert(0, os.path.abspath("./"))
34 #sys.path.insert(0, os.path.abspath("./ricxappframe"))
35 sys.path.append(os.getcwd())
36 from ricxappframe.xapp_frame import RMRXapp, rmr
37 from ricxappframe.xapp_sdl import SDLWrapper
38 from ricxappframe.xapp_symptomdata import Symptomdata
39 import ricxappframe.xapp_subscribe as subscribe
40 import ricxappframe.xapp_rest as ricrest
41 from mdclogpy import Logger, Level
44 def __init__(self, xapp_name, config_file):
45 self.config_file = config_file
46 self.xapp_name = xapp_name
52 with open(self.config_file, 'r') as file:
55 self.cfg = json.loads(cfg)
56 if self.cfg is not None:
57 self.controls = self.cfg['controls']
59 def get_key_item(self, key):
61 if self.keys.get(key) is not None:
67 with open(self.config_file, 'r') as file:
70 self.cfg = json.loads(cfg)
71 # following is required by the appmgr - don't know why.
72 cfgescaped = cfg.replace('"', '\\"').replace('\n', '\\n')
73 data = '[{ "config": "' + cfgescaped + '", "metadata":{"configType":"json","xappName":"' + self.xapp_name + '"}}]'
75 logging.error("Config file %s empty or does not exists" % (self.config_file))
78 def read_file(filename):
80 with open(filename, 'r') as f:
85 except IOError as error:
91 def __init__(self, name, service, address, port, config, loglevel):
92 signal.signal(signal.SIGQUIT, self.signal_handler)
93 signal.signal(signal.SIGTERM, self.signal_handler)
94 signal.signal(signal.SIGINT, self.signal_handler)
97 self.configData = Config(name, config)
99 if self.configData.controls.get('logger'):
100 level = self.configData.controls['logger'].get('level')
102 if port == None and self.configData.cfg['messaging'].get('ports'):
103 for item in self.configData.cfg['messaging'].get('ports'):
104 if item['name'] == http:
105 port = int(item['port'])
106 if address == None and self.configData.cfg.get('name'):
107 address = self.configData.cfg.get('name')
109 # save the listen address and port for later use
111 self.address = address
113 self.logger = Logger(name, loglevel)
114 # setup the symptomdata
115 symptomCfg = self.GetSymptomConfig()
116 self.symptomHndl = Symptomdata(service, name, "/tmp/", symptomCfg['url'], symptomCfg['timeout'])
118 # create the thread HTTP server and set the uri handler callbacks
119 self.server = ricrest.ThreadedHTTPServer(address, port)
120 # trick to get the own handler with defined
121 self.server.handler.add_handler(self.server.handler, "GET", "config", "/ric/v1/config", self.configGetHandler)
122 self.server.handler.add_handler(self.server.handler, "GET", "healthAlive", "/ric/v1/health/alive", self.healthyGetAliveHandler)
123 self.server.handler.add_handler(self.server.handler, "GET", "healthReady", "/ric/v1/health/ready", self.healthyGetReadyHandler)
124 self.server.handler.add_handler(self.server.handler, "GET", "symptomdata", "/ric/v1/symptomdata", self.symptomdataGetHandler)
128 self.startRMR(service, 4)
130 # now we can subscribe
133 def startRMR(self, service, level):
134 # handle the RMR_SEED_RT and RMR_RTG_SVC which is different in mcxapp
136 os.environ["RMR_SRC_ID"] = service
137 os.environ["RMR_LOG_VLEVEL"] = str(level)
138 os.environ["RMR_RTG_SVC"] = "4561"
139 rmrseed = os.environ.get('RMR_SEED_RT')
140 if rmrseed is not None:
141 data = read_file(rmrseed)
143 self.logger.warning("RMR seed file %s does not exists or is empty" % (rmrseed))
145 self.logger.info("RMR_SEED_RT seed file not set in environment")
146 data = read_file('uta-rtg.rt')
148 os.environ['RMR_SEED_RT'] = "./uta-rtg.rt"
149 self.logger.info("Setting the default RMR_SEED_RT=uta-rtg.rt - content:")
151 self.logger.info("Try to export the RMR_SEED_RT file if your RMR is not getting ready")
152 self.rmrInit(b"4560")
154 def signal_handler(self, sig, frame):
155 if self.running is True:
157 rmr.rmr_close(self.rmr_mrc)
161 def rmrInit(self, initbind):
163 self.rmr_mrc = rmr.rmr_init(initbind, rmr.RMR_MAX_RCV_BYTES, 0x00)
164 while rmr.rmr_ready(self.rmr_mrc) == 0:
166 self.logger.info('RMR not yet ready')
167 rmr.rmr_set_stimeout(self.rmr_mrc, 1)
168 rmr.rmr_set_vlevel(5)
169 self.logger.info('RMR ready')
171 def GetSymptomConfig(self):
172 if self.configData.cfg['controls'].get('symptomdata').get('lwsd'):
173 return self.configData.cfg['controls'].get('symptomdata').get('lwsd')
175 def GetSubsConfig(self):
176 if self.configData.cfg['controls'].get('subscription'):
177 return self.configData.cfg['controls'].get('subscription')
180 self.subsCfgDetail = self.GetSubsConfig()
181 if self.subsCfgDetail != None:
182 # this is example subscription, for your use case fill the attributes according to your needs
183 self.subscriber = subscribe.NewSubscriber(self.subsCfgDetail['url'] + 'ric/v1')
184 # add as well the own subscription response callback handler
185 if self.subscriber.ResponseHandler(self.subsResponseCB, self.server) is not True:
186 self.logger.error("Error when trying to set the subscription reponse callback")
187 # setup the subscription data
188 subEndPoint = self.subscriber.SubscriptionParamsClientEndpoint(self.subsCfgDetail['clientEndpoint'], self.port, 4061)
189 subsDirective = self.subscriber.SubscriptionParamsE2SubscriptionDirectives(10, 2, False)
190 subsequentAction = self.subscriber.SubsequentAction("continue", "w10ms")
191 actionDefinitionList = self.subscriber.ActionToBeSetup(1, "policy", (11,12,13,14,15), subsequentAction)
192 subsDetail = self.subscriber.SubscriptionDetail(12110, (1,2,3,4,5), actionDefinitionList)
193 # subscription data ready, make the subscription
194 subObj = self.subscriber.SubscriptionParams("sub10", subEndPoint,"gnb123456",1231, subsDirective, subsDetail)
195 self.logger.info("Sending the subscription to %s" %(self.subsCfgDetail['url'] + 'ric/v1'))
196 self.logger.info(subObj.to_dict())
198 data, reason, status = self.subscriber.Subscribe(subObj)
199 # returns the json data, make it dictionary
200 self.logger.info("Getting the subscription reponse")
201 self.logger.info(json.loads(data))
203 def Unsubscribe(self):
204 reason, status = self.subscriber.UnSubscribe("ygwefwebw")
206 def QuerySubscribtions(self):
207 data, reason, status = self.subscriber.QuerySubscriptions()
209 def healthyGetReadyHandler(self, name, path, data, ctype):
210 response = server.initResponse()
211 response['payload'] = ("{'status': 'ready'}")
214 def healthyGetAliveHandler(self, name, path, data, ctype):
215 response = server.initResponse()
216 response['payload'] = ("{'status': 'alive'}")
219 def subsResponseCB(self, name, path, data, ctype):
220 response = server.initResponse()
221 response['payload'] = ("{}")
224 def getSymptomData(self, uriparams):
225 paramlist = urllib.parse.parse_qs(uriparams)
226 [x.upper() for x in paramlist]
230 if paramlist.get('fromTime'):
231 fromtime = getSeconds(paramlist.get('fromTime')[0])
232 if paramlist.get('toTime'):
233 totime = getSeconds(paramlist.get('toTime')[0])
234 zipfile = self.symptomHndl.collect("symptomdata"+'-%Y-%m-%d-%H-%M-%S.zip', ('examples/.*.py',), fromtime, totime)
236 (zipfile, size, data) = self.symptomHndl.read()
237 return (zipfile, size, data)
238 return (None, 0, None)
240 def symptomdataGetHandler(self, name, path, data, ctype):
241 reponse = ricrest.initResponse()
242 (zipfile, size, filedata) = self.getSymptomData(self.path[20:])
244 reponse['payload'] = filedata
245 reponse['ctype'] = 'application/zip'
246 reponse['attachment'] = "symptomdata.zip"
247 reponse['mode'] = 'binary'
249 logging.error("Symptom data does not exists")
250 reponse['response'] = 'System error - symptomdata does not exists'
251 reponse['status'] = 500
254 def configGetHandler(self, name, path, data, ctype):
255 response = server.initResponse()
256 response['payload'] = (self.configData.get_config())
259 def removeEnvVar(evar):
260 if evar in os.environ:
264 parser = argparse.ArgumentParser()
265 parser.add_argument('-port', dest='port', help='HTTP server listen port, default 8088', required=False, type=int)
266 parser.add_argument('-address', dest='address', help='IP listen address, default all interfaces', required=False, type=str)
267 parser.add_argument('-config', dest='config', help='config file path name, default /opt/ric/config/config.json', required=False, type=str)
268 parser.add_argument('-xapp', dest='xapp', help='xapp name', required=True, type=str)
269 parser.add_argument('-service', dest='service', help='xapp service name (same as pod host name)', required=True, type=str)
270 parser.add_argument('-verbose', dest='verbose', help='verbose logging level', required=False, type=int)
272 args = parser.parse_args()
274 if args.port is None:
276 if args.address is None:
277 args.address = "0.0.0.0"
278 if args.config is None:
279 args.config = '/opt/ric/config/config.json'
281 # remove proxy so that it won't impact to rest calls
282 removeEnvVar('HTTPS_PROXY')
283 removeEnvVar('HTTP_PROXY')
284 removeEnvVar('https_proxy')
285 removeEnvVar('http_proxy')
287 # starting argument option will overwrite the config settings
288 if args.verbose is None:
291 loglevel = Level.INFO
292 if args.verbose == 0:
293 loglevel = Level.ERROR
294 if args.verbose == 1:
295 loglevel = Level.WARNING
296 elif args.verbose == 2:
297 loglevel = Level.INFO
298 elif args.verbose >= 3:
299 loglevel = Level.DEBUG
301 myxapp = MyXapp(args.xapp, args.service, args.address, args.port, args.config, loglevel)
304 print("Waiting for a message, will timeout after 10s")
305 rmr_sbuf = rmr.rmr_torcv_msg(myxapp.rmr_mrc, None, 10000)
306 summary = rmr.message_summary(rmr_sbuf)
307 if summary[rmr.RMR_MS_MSG_STATE] == 12:
308 print("Nothing received")
310 print("Message received!: {}".format(summary))
311 data = rmr.get_payload(rmr_sbuf)
312 rmr.rmr_free_msg(rmr_sbuf)
314 if __name__ == '__main__':