# SPDX-FileCopyrightText: 2024-2025 Helmholtz-Zentrum Dresden-Rossendorf e.V (HZDR)
# SPDX-License-Identifier: Apache-2.0
import os
import pandas as pd
from labfrog.common import read_config
# uncomment and provide config file if using diagnostics_maker by itself
HZDR_MONGODB = False
Docker_MongoDB = True
[docs]
def get_data_and_docs(config_name):
# Get the directory of the current script
current_directory = os.getcwd()
# Assuming 'labfrog' folder is to be removed from the path
# instance_directory = os.path.abspath(os.path.join(os.path.dirname(current_directory), "..", "..", "instance"))
instance_directory = os.path.abspath(os.path.join(current_directory, "instance"))
full_config_file_path = os.path.join(instance_directory, config_name)
config = read_config(full_config_file_path)
if "USE_FULL_CUSTOM" in config:
# CUSTOM_OPTIONS = str(config['USE_FULL_CUSTOM']).strip().lower() in ["true", "1", "yes"]
raw_path = config["USE_FULL_CUSTOM_FILE"].replace('"', "")
else:
exit()
file_path = os.path.abspath(
os.path.join(current_directory, "labfrog", "customize", raw_path)
)
# 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)
# READ OUT PARAMETERS
always_include_df = pd.read_excel(
file_path, sheet_name="always_include", engine=file_engine
)
always_include_shot = always_include_df["shot"].dropna().tolist()
always_include_set = always_include_df["set"].dropna().tolist()
# Read the relevant templates (ODS pages)
parameters_by_section_SHOT_df = pd.read_excel(
file_path, sheet_name="parameters_by_section_SHOT", engine=file_engine
)
parameters_by_section_SET_df = pd.read_excel(
file_path, sheet_name="parameters_by_section_SET", engine=file_engine
)
# print(parameters_by_section_SHOT_df.head()) # Display first 5 rows of the DataFrame
# Now, create the dictionary: Each column will map to a list of values from the rows beneath the first row
shot_parameters_by_sections = {}
for column in parameters_by_section_SHOT_df.columns:
shot_parameters_by_sections[column] = (
parameters_by_section_SHOT_df[column].dropna().tolist()
) # Exclude NaN values
# Now, create the dictionary: Each column will map to a list of values from the rows beneath the first row
set_parameters_by_sections = {}
for column in parameters_by_section_SET_df.columns:
set_parameters_by_sections[column] = (
parameters_by_section_SET_df[column].dropna().tolist()
) # Exclude NaN values
# Optionally, print the results to verify
# print("Shot Parameters:")
# print(json.dumps(shot_parameters_by_sections, indent=4))
# print("\nSet Parameters:")
# print(json.dumps(set_parameters_by_sections, indent=4))
# READ OUT DIAGNOSTICS
diagnostics_df = pd.read_excel(
file_path, sheet_name="diagnostics", engine=file_engine
)
# Replace NaN values with empty strings
diagnostics_df = diagnostics_df.fillna("")
# Convert DataFrame to JSON
data = []
for _index, row in diagnostics_df.iterrows():
entry = {
"DisplayName": row["DisplayName"],
"Tooltip": row["Tooltip"],
"Details": row["Details"],
"Class": row["Class"],
"Wiki-Link(s)": row["Wiki-Link(s)"],
"Team": row["Team"],
"Responsible": row["Responsible"],
"Valid since": str(row["Valid since"]),
"Valid until": str(row["Valid until"]),
}
data.append(entry)
# Get unique DisplayNames from the DataFrame
choices = diagnostics_df["DisplayName"].unique().tolist()
# Create the template dictionary
template = {
"layout_name": "All Fields",
"description": "DESCRIPTION",
"responsible_person": "KT",
"selected_fields": choices,
}
# UPLOAD THESE NEW MONGODB PARAMETER GUIDES TO THE MONGODB AS DEFAULT ONLY FOR your personal DOCKER MONGODB
# FOR NOW WE MAKE A NEW COLLECTION AND PUT THINGS THERE AND ADAPT CONFIG TO TELL PROGAM WHERE TO READ IT FROM
from datetime import datetime
shot_document = {
"layout_name": "DEFAULT",
"date_time": datetime.now(),
"responsible_person": "tippey27",
"description": "fields to start with",
"mode": "shot",
"always_include": always_include_shot,
"field_sections_dict": shot_parameters_by_sections,
"diagnostics_list": choices,
}
set_document = {
"layout_name": "DEFAULT",
"date_time": datetime.now(),
"responsible_person": "tippey27",
"description": "fields to start with",
"mode": "set",
"always_include": always_include_set,
"field_sections_dict": set_parameters_by_sections,
"diagnostics_list": choices,
}
# Optional aliases sheet (field_name, alias)
aliases = []
try:
aliases_df = pd.read_excel(file_path, sheet_name="aliases", engine=file_engine)
aliases_df.columns = (
aliases_df.columns.str.strip().str.lower().str.replace('"', "")
)
for _index, row in aliases_df.iterrows():
field_name = str(row.get("field_name", "")).strip()
alias = str(row.get("alias", "")).strip()
campaign = str(row.get("campaign", "")).strip() or None
if field_name and alias:
aliases.append({
"field_name": field_name,
"alias": alias,
"campaign": campaign,
})
except Exception:
aliases = []
return data, shot_document, set_document, aliases
[docs]
def make_shot_and_set_defaults(collection, shot_document, set_document):
# Delete old documents with layout_name "DEFAULT"
collection.delete_many({"layout_name": "DEFAULT"})
try:
collection.insert_one(shot_document)
collection.insert_one(set_document)
except Exception as e:
print(f"Problem getting collection: {e}")
[docs]
def create_new_diagnostics(db, collection, data):
collection.delete_many({})
# Add default value OFF to each diagnostic
for diagnostic in data:
# Create a new collection for the diagnostic
diagnostic_collection_name = f"diagnostics.{diagnostic['DisplayName']}"
diagnostic_collection = db[diagnostic_collection_name]
# Check if the diagnostic collection is empty
if diagnostic_collection.count_documents({}) == 0:
off_entry = {
"name": "OFF",
"description": "",
"set_up_date": "",
"details": "",
"responsible_person": "",
"pc": "",
"file_path": "",
"filename_schema": "",
"counter_mode": "",
"active": True,
}
# Insert the diagnostic into the main diagnostics collection
collection.insert_one(diagnostic)
# Insert the OFF entry into the diagnostic's specific collection
diagnostic_collection.insert_one(off_entry)
else:
# If the collection is not empty, insert the diagnostic
# into the main diagnostics collection only
collection.insert_one(diagnostic)
if __name__ == "__main__":
# if running via just running this script, import configs from within same folder and use those values
from pymongo import MongoClient
connection_string, database, default_collection = ""
# USE_HZDR
if HZDR_MONGODB:
from labfrog.configs.configs_private import (
connection_string,
database,
default_collection,
)
# USE DOCKER
if Docker_MongoDB:
from labfrog.configs.configs_docker import (
connection_string,
database,
default_collection,
)
client = MongoClient(connection_string)
db = client[database]
layout_collection = db[default_collection]
diagnostics_collection = db["diagnostics"]
data, shot_document, set_document, _aliases = get_data_and_docs("default.cfg")
make_shot_and_set_defaults(layout_collection, shot_document, set_document)
# create_new_diagnostics(db, diagnostics_collection)