diff --git a/tools/telemetryInputCli/.gitignore b/tools/telemetryInputCli/.gitignore new file mode 100644 index 0000000..29249e0 --- /dev/null +++ b/tools/telemetryInputCli/.gitignore @@ -0,0 +1,4 @@ +*.csv + +!examples/data_temp.csv +!examples/data_lumin.csv \ No newline at end of file diff --git a/tools/telemetryInputCli/README.MD b/tools/telemetryInputCli/README.MD index 1d21756..d57e56b 100644 --- a/tools/telemetryInputCli/README.MD +++ b/tools/telemetryInputCli/README.MD @@ -16,7 +16,43 @@ - cmd - influxdb_client - dotenv + - yaml ## functionality - wells data input + +## environment + +### .env file + +The script requires a file called '.env' with the following format: + +```env +URL = +DB_BUCKET = +DB_ORG = +DB_USER = +DB_PASSWORD = +DB_TOKEN = +... +``` + +### config.yaml + +The script's bulk insert function requires a `config.yaml` to externally store the csv format of each table with the following format (to be extended): + +```yaml +tables: # for read the data frame file + - name: temp # need to match the field name in .env and influx table + columns: + - timestamp # UNIX timestamp + - well_num + - temperature + + - name: lumin + columns: + - timestamp # UNIX timestamp + - well_num + - luminosity +``` diff --git a/tools/telemetryInputCli/config.yaml b/tools/telemetryInputCli/config.yaml new file mode 100644 index 0000000..422b86e --- /dev/null +++ b/tools/telemetryInputCli/config.yaml @@ -0,0 +1,12 @@ +tables: # for read the data frame file + - name: temp # need to match the field name in .env and influx table + columns: + - timestamp + - well_num + - temperature + + - name: lumin + columns: + - timestamp + - well_num + - luminosity \ No newline at end of file diff --git a/tools/telemetryInputCli/examples/data_lumin.csv b/tools/telemetryInputCli/examples/data_lumin.csv new file mode 100644 index 0000000..0024b4e --- /dev/null +++ b/tools/telemetryInputCli/examples/data_lumin.csv @@ -0,0 +1,13 @@ +timestamp,well_num,luminosity +1756749600,1,600 +1756749600,2,350 +1756749600,3,750 +1756749600,4,500 +1756836000,1,800 +1756836000,2,450 +1756836000,3,300 +1756836000,4,650 +1756922400,1,200 +1756922400,2,400 +1756922400,3,700 +1756922400,4,250 diff --git a/tools/telemetryInputCli/examples/data_temp.csv b/tools/telemetryInputCli/examples/data_temp.csv new file mode 100644 index 0000000..86cb86e --- /dev/null +++ b/tools/telemetryInputCli/examples/data_temp.csv @@ -0,0 +1,13 @@ +timestamp,well_num,temperature +1756749600,1,3 +1756749600,2,4 +1756749600,3,1 +1756749600,4,-3 +1756836000,1,2 +1756836000,2,2 +1756836000,3,4 +1756836000,4,-2 +1756922400,1,-1 +1756922400,2,0 +1756922400,3,3 +1756922400,4,0 diff --git a/tools/telemetryInputCli/telemetryInputCli.py b/tools/telemetryInputCli/telemetryInputCli.py index ac5417b..14426c6 100644 --- a/tools/telemetryInputCli/telemetryInputCli.py +++ b/tools/telemetryInputCli/telemetryInputCli.py @@ -1,12 +1,37 @@ import cmd +import pandas as pd +import os +import yaml +from typing import List, Dict from datetime import datetime, timezone, timedelta from influxdb_client import InfluxDBClient, Point, WritePrecision from influxdb_client.client.write_api import SYNCHRONOUS from dotenv import load_dotenv, dotenv_values +CONFIG_FILE = 'config.yaml' + # Load environment variables load_dotenv() -config = dotenv_values(".env") +env = dotenv_values(".env") + +def load_table_configs() -> Dict[str, List[str]]: + raw = yaml.safe_load(open(CONFIG_FILE, 'r')) + table_configs: Dict[str, List[str]] = {} + for entry in raw["tables"]: + name = entry["name"] + columns = entry["columns"] + if not isinstance(columns, list): + print(f"Invalid columns for table {name}. Expected a list, got {type(columns)}") + continue + table_configs[name] = (columns) + return table_configs + +# convert unix timestamp to datetime +def convert_unix_to_datetime(unix_timestamp: int): + if unix_timestamp < 0: + print(f"Invalid unix timestamp: {unix_timestamp}") + return None + return datetime.fromtimestamp(unix_timestamp, tz=timezone.utc) # ask for period of time def askingForPeriod() -> str: @@ -38,17 +63,30 @@ def insertDataIntoWell(well: int, field: str, data: float, time: datetime, host: point = Point("well").tag("well", well).field(field, data).time(time, WritePrecision.NS).tag("host", host) write_api.write(bucket, org, point) +# ask for file path +def askingForFilePath() -> str: + path = input("Please enter the file path: ") + if not path: + print("Invalid file path") + return None + if not os.path.exists(path): + print("File does not exist") + return None + return path + class TelemetryInputCli(cmd.Cmd): + # load up config file + table_configs = load_table_configs() # Set up the CLI - url = config.get('URL') - bucket = config.get('DB_BUCKET') - org = config.get('DB_ORG') - user = config.get('DB_USER') - password = config.get('DB_PASSWORD') - token = config.get('DB_TOKEN') - payloadTag = config.get('PAYLOAD_TAG') - wellTempField = config.get('WELL_TEMP_FIELD') - wellLuminField = config.get('WELL_LUMIN_FIELD') + url = env.get('URL') + bucket = env.get('DB_BUCKET') + org = env.get('DB_ORG') + user = env.get('DB_USER') + password = env.get('DB_PASSWORD') + token = env.get('DB_TOKEN') + payloadTag = env.get('PAYLOAD_TAG') + wellTempField = env.get('WELL_TEMP_FIELD') + wellLuminField = env.get('WELL_LUMIN_FIELD') client = InfluxDBClient(url=url, token=token, org=org) write_api = client.write_api(write_options=SYNCHRONOUS) prompt = '> ' @@ -69,16 +107,34 @@ def do_insert_temp_well(self, arg): 'Insert temperature data into a specific well with specified time' self.insert_data_into_well(self.wellTempField) - # insert humidity data into all wells with specified time + # insert luminosity data into all wells with specified time def do_insert_lumin_wells(self, arg): - 'Insert luminance data into all wells with specified time' + 'Insert luminosity data into all wells with specified time' self.insert_data_into_wells(self.wellLuminField) - # insert humidity data into a specific well with specified time + # insert luminosity data into a specific well with specified time def do_insert_lumin_well(self, arg): - 'Insert luminance data into a specific well with specified time' + 'Insert luminosity data into a specific well with specified time' self.insert_data_into_well(self.wellLuminField) + def do_bulk_insert(self, arg): + 'Bulk insert data from a CSV file into the specified table' + # print out available option from table_configs + if not self.table_configs: + print("No tables available for bulk insert.") + return + print("Available tables for bulk insert:") + for table in self.table_configs.keys(): + print(f"- {table}") + table = input("Please enter the table name: ") + if table not in self.table_configs: + print(f"Table {table} does not exist.") + return + filePath = askingForFilePath() + if not filePath: + return + self.insert_data_from_df(table, filePath) + # helper function to insert data into all wells def insert_data_into_wells(self, field): period = askingForPeriod() @@ -102,6 +158,47 @@ def insert_data_into_well(self, field): return insertDataIntoWell(well, field, data, calculatedTime, self.payloadTag, self.write_api, self.bucket, self.org) print("Data inserted successfully") + + # bulk insert data into bucket + def insert_data_from_df(self, table:str, filePath:str): + if not filePath: + return + if filePath.endswith('.csv'): + try: + df = pd.read_csv(filePath, usecols=self.table_configs[table]) + except ValueError as e: + print(f"Error reading CSV file: {e}") + return + else: + print("Unsupported file format. Please provide a CSV file.") + return + # print out record + print(f"Loading {len(df)} records from {filePath} for table {table}.") + if df.empty: + print("No data to insert.") + return + # convert unix timestamp to datetime + if 'timestamp' in df.columns: + df['timestamp'] = df['timestamp'].apply(lambda x: convert_unix_to_datetime(x) if pd.notnull(x) else None) + else: + print("No timestamp column found in the data.") + return + # insert data into influxdb + if table == self.wellTempField or table == self.wellLuminField: + for index, row in df.iterrows(): + well_num = row.get('well_num') + if pd.notnull(well_num) and 1 <= well_num <= 16: + time = row.get('timestamp') + if time is not None: + data = row.get('temperature') if table == self.wellTempField else row.get('luminosity') + if pd.notnull(data): + insertDataIntoWell(well_num, table, data, time, self.payloadTag, self.write_api, self.bucket, self.org) + else: + print(f"Invalid well number {well_num} at index {index}. Skipping this record.") + else: + print(f"Unsupported table {table}. Only 'temp' and 'lumin' are supported for bulk insert.") + return + print(f"Data from {filePath} inserted into {table} table successfully.") # helper function to get data from user def get_data_from_user(self):