#!/usr/bin/env python3 # Copies the data from a CSV file from the KPI file generated from a Wifi Capacity test to an Influx database # The CSV requires three columns in order to work: Date, test details, and numeric-score. # Date is a unix timestamp, test details is the variable each datapoint is measuring, and numeric-score is the value for that timepoint and variable. import sys import os from pprint import pprint from influx2 import RecordInflux if sys.version_info[0] != 3: print("This script requires Python 3") exit(1) if 'py-json' not in sys.path: sys.path.append(os.path.join(os.path.abspath('..'), 'py-json')) import argparse import datetime def influx_add_parser_args(parser): parser.add_argument('--influx_host', help='Hostname for the Influx database', default=None) parser.add_argument('--influx_port', help='IP Port for the Influx database', default=8086) parser.add_argument('--influx_org', help='Organization for the Influx database', default=None) parser.add_argument('--influx_token', help='Token for the Influx database', default=None) parser.add_argument('--influx_bucket', help='Name of the Influx bucket', default=None) parser.add_argument('--influx_tag', action='append', nargs=2, help='--influx_tag Can add more than one of these.', default=[]) class CSVtoInflux(): def __init__(self, _exit_on_error=False, _exit_on_fail=False, _proxy_str=None, _capture_signal_list=[], influxdb=None, _influx_tag=[], target_csv=None): self.influxdb = influxdb self.target_csv = target_csv.replace('/home/lanforge/html-reports/', '') self.influx_tag = _influx_tag # Submit data to the influx db if configured to do so. def post_to_influx(self): with open(self.target_csv) as fp: line = fp.readline() line = line.split('\t') # indexes tell us where in the CSV our data is located. We do it this way so that even if the columns are moved around, as long as they are present, the script will still work. numeric_score_index = line.index('numeric-score') test_id_index = line.index('test-id') date_index = line.index('Date') test_details_index = line.index('test details') short_description_index = line.index('short-description') graph_group_index = line.index('Graph-Group') units_index = line.index('Units') testbed_index = line.index('test-rig') duthwversion = line.index('dut-hw-version') dutswversion = line.index('dut-sw-version') dutserialnum = line.index('dut-serial-num') line = fp.readline() while line: line = line.split('\t') #split the line by tabs to separate each item in the string date = line[date_index] date = datetime.datetime.utcfromtimestamp(int(date) / 1000).isoformat() #convert to datetime so influx can read it, this is required numeric_score = line[numeric_score_index] numeric_score = float(numeric_score) #convert to float, InfluxDB cannot short_description = line[short_description_index] tags = dict() tags['script'] = line[test_id_index] tags['short-description'] = line[short_description_index] tags['test_details'] = line[test_details_index] tags['Graph-Group'] = line[graph_group_index] tags['DUT-HW-version'] = line[duthwversion] tags['DUT-SW-version'] = line[dutswversion] tags['DUT-Serial-Num'] = line[dutserialnum] tags['testbed'] = line[testbed_index] tags['Units'] = line[units_index] for item in self.influx_tag: # Every item in the influx_tag command needs to be added to the tags variable tags[item[0]] = item[1] self.influxdb.post_to_influx(short_description, numeric_score, tags, date) line = fp.readline() #influx wants to get data in the following format: # variable n ame, value, tags, date # total-download-mbps-speed-for-the-duration-of-this-iteration 171.085494 {'script': 'WiFi Capacity'} 2021-04-14T19:04:04.902000 def script_name(self): with open(self.target_csv) as fp: line = fp.readline() line = line.split('\t') test_id_index = line.index('test-id') line = fp.readline() line.split('\t') return line[test_id_index] def create_dashboard(self, dashboard_name=None): #Create a dashboard in Grafana to look at the data you just posted to Influx dashboard_name def main(): lfjson_host = "localhost" lfjson_port = 8080 endp_types = "lf_udp" debug = False parser = argparse.ArgumentParser( prog='csv_to_influx.py', # formatter_class=argparse.RawDescriptionHelpFormatter, formatter_class=argparse.RawTextHelpFormatter, epilog=''' ''', description='''\ csv_to_influx.py: -------------------- Summary : ---------- Copies the data from a CSV file generated by a wifi capacity test to an influx database. Column names are designed for the KPI file generated by our Wifi Capacity Test. A user can of course change the column names to match these in order to input any csv file. The CSV file needs to have the following columns: --date - which is a UNIX timestamp --test details - which is the variable being measured by the test --numeric-score - which is the value for that variable at that point in time. Generic command layout: ----------------------- python .\\csv_to_influx.py Command: python3 csv_to_influx.py --influx_host localhost --influx_org Candela --influx_token random_token --influx_bucket lanforge --target_csv kpi.csv ''') influx_add_parser_args(parser) # This argument is specific to this script, so not part of the generic influxdb parser args # method above. parser.add_argument('--target_csv', help='CSV file to record to influx database', default="") args = parser.parse_args() influxdb = RecordInflux(_lfjson_host=lfjson_host, _lfjson_port=lfjson_port, _influx_host=args.influx_host, _influx_port=args.influx_port, _influx_org=args.influx_org, _influx_token=args.influx_token, _influx_bucket=args.influx_bucket) csvtoinflux = CSVtoInflux(influxdb=influxdb, target_csv=args.target_csv, _influx_tag=args.influx_tag) csvtoinflux.post_to_influx() if __name__ == "__main__": main()