mirror of
https://github.com/Telecominfraproject/wlan-lanforge-scripts.git
synced 2025-11-03 12:18:00 +00:00
@@ -26,19 +26,20 @@ class LFDataCollection:
|
|||||||
|
|
||||||
def check_json_validity(self, keyword=None, json_response=None):
|
def check_json_validity(self, keyword=None, json_response=None):
|
||||||
if json_response is None:
|
if json_response is None:
|
||||||
raise ValueError("Cannot find columns requested to be searched in port manager. Exiting script, please retry.")
|
raise ValueError(
|
||||||
|
"Cannot find columns requested to be searched in port manager. Exiting script, please retry.")
|
||||||
if keyword is not None and keyword not in json_response:
|
if keyword is not None and keyword not in json_response:
|
||||||
raise ValueError("Cannot find proper information from json. Please check your json request. Exiting script, please retry.")
|
raise ValueError(
|
||||||
|
"Cannot find proper information from json. Please check your json request. Exiting script, please retry.")
|
||||||
|
|
||||||
def get_milliseconds(self,
|
def get_milliseconds(self,
|
||||||
timestamp):
|
timestamp):
|
||||||
return (timestamp - datetime.datetime(1970, 1, 1)).total_seconds() * 1000
|
return (timestamp - datetime.datetime(1970, 1, 1)).total_seconds() * 1000
|
||||||
|
|
||||||
def get_seconds(self,
|
def get_seconds(self,
|
||||||
timestamp):
|
timestamp):
|
||||||
return (timestamp - datetime.datetime(1970, 1, 1)).total_seconds()
|
return (timestamp - datetime.datetime(1970, 1, 1)).total_seconds()
|
||||||
|
|
||||||
|
|
||||||
# only for ipv4_variable_time at the moment
|
# only for ipv4_variable_time at the moment
|
||||||
def monitor_interval(self, header_row_=None,
|
def monitor_interval(self, header_row_=None,
|
||||||
start_time_=None, sta_list_=None,
|
start_time_=None, sta_list_=None,
|
||||||
@@ -54,7 +55,8 @@ class LFDataCollection:
|
|||||||
# get responses from json
|
# get responses from json
|
||||||
layer_3_response = self.json_get("/endp/%s?fields=%s" % (created_cx_, layer3_fields_), debug_=self.debug)
|
layer_3_response = self.json_get("/endp/%s?fields=%s" % (created_cx_, layer3_fields_), debug_=self.debug)
|
||||||
if port_mgr_fields_ is not None:
|
if port_mgr_fields_ is not None:
|
||||||
port_mgr_response=self.json_get("/port/1/1/%s?fields=%s" % (sta_list_, port_mgr_fields_), debug_=self.debug)
|
port_mgr_response = self.json_get("/port/1/1/%s?fields=%s" % (sta_list_, port_mgr_fields_),
|
||||||
|
debug_=self.debug)
|
||||||
|
|
||||||
# check json response validity
|
# check json response validity
|
||||||
self.check_json_validity(keyword="endpoint", json_response=layer_3_response)
|
self.check_json_validity(keyword="endpoint", json_response=layer_3_response)
|
||||||
@@ -86,6 +88,4 @@ class LFDataCollection:
|
|||||||
temp_list.append(merge[name])
|
temp_list.append(merge[name])
|
||||||
return temp_list
|
return temp_list
|
||||||
|
|
||||||
|
|
||||||
# class WebSocket():
|
# class WebSocket():
|
||||||
|
|
||||||
Reference in New Issue
Block a user