# SPDX-FileCopyrightText: 2024-2026 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
from urllib.parse import quote
import pandas as pd
from pymongo import MongoClient
# Load Excel or ODS file
# CAMPAIGNS_FILE = "labfrog\customize\CUSTOM_CAMPAIGNS.ods" # Change this to your file path
# TARGETS_FILE = "labfrog\customize\CUSTOM_TARGETS.ods" # Change this to your file path
def _normalize_setting(value, default=""):
text = default if value is None else str(value).strip()
if len(text) >= 2 and text[0] == text[-1] and text[0] in {'"', "'"}:
text = text[1:-1].strip()
return text or default
def _resolve_custom_options_connection_settings(connection_settings=None):
settings = dict(connection_settings or {})
return {
"host": _normalize_setting(settings.get("host"), "localhost"),
"port": int(_normalize_setting(settings.get("port"), "27018")),
"username": _normalize_setting(settings.get("username"), "root"),
"password": _normalize_setting(settings.get("password"), "mypasswd"),
"auth_source": _normalize_setting(settings.get("auth_source"), "admin"),
"database": _normalize_setting(settings.get("database"), "shotsheet"),
}
def _build_connection_string(connection_settings=None):
settings = _resolve_custom_options_connection_settings(connection_settings)
username = quote(settings["username"])
password = quote(settings["password"])
return (
f"mongodb://{username}:{password}@{settings['host']}:{settings['port']}/"
f"?authMechanism=DEFAULT&authSource={settings['auth_source']}"
)
[docs]
def read_custom_campaigns(
file_path, CUSTOM_OPTIONS_COLLECTION, *, connection_settings=None
):
# Check file extension to determine the file format
if file_path.endswith(".xlsx"):
file_engine = "openpyxl" # Specify engine for Excel files
elif file_path.endswith(".ods"):
file_engine = "odf" # Specify engine for ODS files
else:
print("Unsupported file format.")
exit()
# Read data from file into a pandas DataFrame
df = pd.read_excel(file_path, engine=file_engine)
df.columns = df.columns.str.strip().str.replace('"', "")
# Convert DataFrame to JSON
data = []
for _index, row in df.iterrows():
entry = {}
# Iterate through each column name and its corresponding value in the row
for column_name in df.columns:
entry[column_name] = row[column_name]
# Append the entry to the data list
data.append(entry)
# Get unique DisplayNames from the DataFrame
campaign_choices = df["Campaign Name"].unique().tolist()
# Create an empty dictionary to store campaign data
campaign_dict = {}
# Iterate over the list of dictionaries in `data` to populate `campaign_dict`
for entry in data:
campaign_name = entry["Campaign Name"]
if campaign_name not in campaign_dict:
campaign_dict[
campaign_name
] = {} # Initialize a nested dictionary for each campaign
# Add further details directly into the nested dictionary
for key, value in entry.items():
if key != "Campaign Name": # Exclude the campaign name from being nested
campaign_dict[campaign_name][key] = value
campaign_options_upload = {
"option": "campaigns",
"campaigns_dict": campaign_dict,
"choices": campaign_choices,
}
client = MongoClient(_build_connection_string(connection_settings))
db = client[
_resolve_custom_options_connection_settings(connection_settings)["database"]
]
actual_collection = db[CUSTOM_OPTIONS_COLLECTION]
actual_collection.delete_many({"option": "campaigns"})
actual_collection.insert_one(campaign_options_upload)
return campaign_choices, campaign_dict
[docs]
def read_custom_campaigns_local(file_path):
"""Read campaigns from file without writing to MongoDB."""
if file_path.endswith(".xlsx"):
file_engine = "openpyxl"
elif file_path.endswith(".ods"):
file_engine = "odf"
else:
print("Unsupported file format.")
exit()
df = pd.read_excel(file_path, engine=file_engine)
df.columns = df.columns.str.strip().str.replace('"', "")
data = []
for _index, row in df.iterrows():
entry = {}
for column_name in df.columns:
entry[column_name] = row[column_name]
data.append(entry)
campaign_choices = df["Campaign Name"].unique().tolist()
campaign_dict = {}
for entry in data:
campaign_name = entry["Campaign Name"]
if campaign_name not in campaign_dict:
campaign_dict[campaign_name] = {}
for key, value in entry.items():
if key != "Campaign Name":
campaign_dict[campaign_name][key] = value
return campaign_choices, campaign_dict
[docs]
def read_custom_targets(
file_path, CUSTOM_OPTIONS_COLLECTION, *, connection_settings=None
):
# Check file extension to determine the file format
if file_path.endswith(".xlsx"):
file_engine = "openpyxl" # Specify engine for Excel files
elif file_path.endswith(".ods"):
file_engine = "odf" # Specify engine for ODS files
else:
print("Unsupported file format.")
exit()
# Read data from file into a pandas DataFrame
df = pd.read_excel(file_path, engine=file_engine)
df.columns = df.columns.str.strip().str.replace('"', "")
# Convert DataFrame to JSON
data = []
for _index, row in df.iterrows():
entry = {}
# Iterate through each column name and its corresponding value in the row
for column_name in df.columns:
entry[column_name] = row[column_name]
# Append the entry to the data list
data.append(entry)
# Get unique DisplayNames from the DataFrame
target_choices = df["Target Name"].unique().tolist()
# Optionally, update `target_dict` if you want to merge the dictionary created from cargoquery with the new DataFrame-based data
# Assuming the keys in `target_dict` are unique and match the target names
target_dict = {}
for entry in data: # iterate over each dictionary in data list
target_name = entry["Target Name"]
if target_name not in target_dict:
target_dict[
target_name
] = {} # Initialize a nested dictionary for each target
# Add further details directly into the nested dictionary
for key, value in entry.items():
if key != "Target Name": # Exclude the target name from being nested
target_dict[target_name][key] = value
target_options_upload = {
"option": "targets",
"targets_dict": target_dict,
"choices": target_choices,
}
# print("target_options_upload", target_options_upload)
client = MongoClient(_build_connection_string(connection_settings))
db = client[
_resolve_custom_options_connection_settings(connection_settings)["database"]
]
actual_collection = db[CUSTOM_OPTIONS_COLLECTION]
actual_collection.delete_many({"option": "targets"})
actual_collection.insert_one(target_options_upload)
return target_choices, target_dict
[docs]
def read_custom_targets_local(file_path):
"""Read targets from file without writing to MongoDB."""
if file_path.endswith(".xlsx"):
file_engine = "openpyxl"
elif file_path.endswith(".ods"):
file_engine = "odf"
else:
print("Unsupported file format.")
exit()
df = pd.read_excel(file_path, engine=file_engine)
df.columns = df.columns.str.strip().str.replace('"', "")
data = []
for _index, row in df.iterrows():
entry = {}
for column_name in df.columns:
entry[column_name] = row[column_name]
data.append(entry)
target_choices = df["Target Name"].unique().tolist()
target_dict = {}
for entry in data:
target_name = entry["Target Name"]
if target_name not in target_dict:
target_dict[target_name] = {}
for key, value in entry.items():
if key != "Target Name":
target_dict[target_name][key] = value
return target_choices, target_dict