import os
import pickle
import subprocess
from pathlib import Path
from numpy import array
from GUIBRUSHR.GUI.WIDGET.MyButton import MyButton
from GUIBRUSHR.GUI.WIDGET.MyDropDown import MyDropdown
from GUIBRUSHR.GUI.WIDGET.MyFigure import MyFigure
from GUIBRUSHR.GUI.WIDGET.MyTextField import MyTextField
from GUIBRUSHR.GUI.WIDGET.MyCheckBox import MyCheckBox
from GUIBRUSHR.GUI.WIDGET.MyEntry import MyEntry
from GUIBRUSHR.GUI.LAYOUT.MyPanel import MyPanel
from GUIBRUSHR.GUI.DataInterface.DataInterface import (
get_target_info,
get_target_list,
get_target_nights,
get_target_instruments
)
from GUIBRUSHR.General_Constants.FunctionsAndConstants.Constant_Variables import ConstantVariables
from GUIBRUSHR.General_Constants.FunctionsAndConstants.general_functions import create_path_night
from GUIBRUSHR.General_Constants.Classes.HelpButton import HelpButton
[docs]
class FrameTellRem(MyPanel):
"""
A panel for handling telluric removal operations in the GUI.
This class provides a user interface for configuring and executing telluric removal
operations on astronomical data. It includes controls for target selection,
instrument configuration, and various telluric removal parameters.
"""
[docs]
def __init__(self, parent, row, column, color, path_default, frame_input, window, **kwargs):
"""
Initialize the FrameTellRem panel.
Args:
parent: Parent widget
row: Row position in parent
column: Column position in parent
color: Background color
path_default: Default path for file operations
frame_input: Input frame reference
window: Main window reference
**kwargs: Additional keyword arguments
"""
self.dropdown_hr_instrument = None
self.dropdown_target = None
self.list_hr_instrument = None
self.path_folder_targets = frame_input.frame_path.path_folder_targets.get_text()
self.color = color
self.global_exc = 0 # Flag to control initialization
self.frame_input = frame_input
self.path_default = path_default
self.window = window
self.path_night_complete = None
# Initialize parent class
super().__init__(parent, color, row, column, **kwargs)
# Connect to database and get initial target information
target_list = get_target_list(self.path_folder_targets)
self.target_information = get_target_info(self.path_folder_targets, target_list[0])
# Create panel rows for organizing widgets
self._create_panel_rows()
# Initialize UI components
self._init_target_selection()
self._init_rad_mode_selection()
self._init_hr_instrument_selection()
self._init_night_selection()
self._init_telluric_parameters()
self._init_removal_controls()
self._init_results_display()
# Setup exchange directory for temporary files
self._setup_exchange_directory()
# Set callbacks and finalize initialization
self._setup_callbacks()
self.global_exc = 1
self.p = None
def _create_panel_rows(self):
"""Create the panel rows for organizing widgets."""
self.row1 = MyPanel(self, self.color, 0, 1)
self.row2 = MyPanel(self, self.color, 1, 1)
self.row3 = MyPanel(self, self.color, 2, 1)
self.row4 = MyPanel(self, self.color, 3, 1)
self.row5 = MyPanel(self, self.color, 4, 1)
self.row6 = MyPanel(self, self.color, 5, 1)
self.row7 = MyPanel(self, self.color, 6, 1)
def _init_target_selection(self):
"""Initialize target selection dropdown and related components."""
target_list = get_target_list(self.path_folder_targets)
self.dropdown_target = MyDropdown(
self.row1, 0, 1, target_list, "Target:",
color=self.color, columnspan=2
)
self.dropdown_target.set_callback(self.update_target)
def _init_rad_mode_selection(self):
"""Initialize radiation mode selection dropdown."""
self.dropdown_rad_mode = MyDropdown(
self.row1, 0, 3, ConstantVariables.LIST_RAD_MODE,
"Rad. Transf. Mode:", initial_value=0,
color=self.color, columnspan=2
)
def _init_hr_instrument_selection(self):
"""Initialize high-resolution instrument selection."""
self.list_hr_instrument = get_target_instruments(
self.path_folder_targets,
self.dropdown_target.get_value(),
"HR",
"Transmission",
Path(__file__).name
)
self.dropdown_hr_instrument = MyDropdown(
self.row1, 0, 5, self.list_hr_instrument,
"HR Instrument:", color=self.color, columnspan=2
)
def _init_night_selection(self):
"""Initialize night selection components."""
nights = get_target_nights(
self.path_folder_targets,
self.dropdown_target.get_value(),
"Transmission",
self.list_hr_instrument[0],
0
)
self.textfield_nights = MyTextField(
self.row1, 0, 7, nights, "Nights:",
color=self.color, columnspan=3, height=3
)
self.checkbox_nights_sim = MyCheckBox(
self.row1, 0, 10, "Night sim/syn",
command=self.update_night_sim
)
def _init_telluric_parameters(self):
"""Initialize telluric removal parameters."""
# Row 2: ordersel, mean columns, tell rm method, nfcoo, save plot, smooth size
self.textfield_order_selection = MyTextField(
self.row2, 0, 1, "full", "Order Selection:",
color=self.color, columnspan=2, width=20
)
self.checkboc_mean_column = MyCheckBox(
self.row2, 0, 3, "Subtraction of the average spectrum",
initial_value=1
)
self.dropdown_tell_rm_method = MyDropdown(
self.row2, 0, 4, ConstantVariables.LIST_TELL_TABLE,
"Tell rm method:", color=self.color, columnspan=2
)
self.checkbox_nfcoo = MyCheckBox(self.row2, 0, 6, "Use last component selection")
self.checkbox_save_plot = MyCheckBox(self.row2, 0, 7, "Save plot")
self.entry_smooth_size = MyEntry(
self.row2, 0, 8, "40", "Smooth size:",
color=self.color, columnspan=2
)
# Additional parameters
self._init_additional_parameters()
def _init_additional_parameters(self):
"""Initialize additional telluric removal parameters."""
# Row 3: mask, threshold meanspec, cutoff teluric, sigma_coeff
self.textfield_yesmask = MyTextField(
self.row3, 0, 1, "1,1,1,1,1,1", "Mask night:",
color=self.color, columnspan=2, width=20
)
self.entry_threshold_mean_spec = MyEntry(
self.row3, 0, 3, "0.3", "Threshold mean spectrum:",
color=self.color, columnspan=2
)
self.entry_cutoff = MyEntry(
self.row3, 0, 5, "0.8", "Cutoff telluric:",
color=self.color, columnspan=2
)
self.textfield_sigma_coeff = MyTextField(
self.row3, 0, 7, "1.5", "Sigma Coefficient:",
color=self.color, columnspan=2, width=20
)
self.checkbox_do_not_fit_continuum = MyCheckBox(
self.row3, 0, 9, "Do not fit continuum"
)
# Row 4: fixed comp, tangent components method, tangent threshold, substitute
# Min and max component hidden but kept for exchange dictionary
# self.entry_min_comp = MyEntry(
# self.row4, 0, 1, "0", "Min Component:",
# color=self.color, columnspan=2
# )
#
# self.entry_max_comp = MyEntry(
# self.row4, 0, 3, "0", "Max Component:",
# color=self.color, columnspan=2
# )
self.entry_fixed_comp = MyEntry(
self.row4, 0, 1, "0", "Fixed Component:",
color=self.color, columnspan=2
)
self.checkbox_tangent_components_method = MyCheckBox(
self.row4, 0, 3, "Tangent components method",
command=self.toggle_fixed_comp_state
)
self.entry_tangent_value = MyEntry(
self.row4, 0, 4, "0.1", "Tangent threshold:",
color=self.color, columnspan=2
)
self.entry_substitute_for_comp = MyEntry(
self.row4, 0, 6, "4", "N. components substitute:",
color=self.color, columnspan=2
)
self.checkbox_standardize_pca = MyCheckBox(
self.row4, 0, 8, "Standardize PCA",
initial_value=0
)
# PCA mode dropdown: spatial (loop over orders) or temporal (loop over spectral channels)
self.dropdown_pca_mode = MyDropdown(
self.row4, 0, 9, ["spatial", "temporal"],
"PCA mode:", color=self.color, columnspan=2
)
def _init_removal_controls(self):
"""Initialize telluric removal control buttons."""
# Create help text dictionary
help_text = {
"Target:": "Select the exoplanet to process. The list is populated from available targets in the target folder specified in the configuration file.",
"Rad. Transf. Mode:": "Select the radiative transfer mode for atmospheric modeling. Could be Transmission or Emission.",
"HR Instrument:": "Select the high-resolution spectrograph used for observations. The list is filtered based on the selected target and only shows instruments with transmission spectroscopy data.",
"Nights:": "Displays the observing nights available for the selected target and instrument. Multiple nights are shown as a comma-separated list.",
"Night sim/syn": "When enabled, switches to simulated/synthetic night data instead of real observational data. Useful for testing and validation purposes.",
"Order Selection": "Specify the name of the folder containing the spectral orders information to process.",
"Subtraction of the average spectrum": "When enabled, subtracts the mean spectrum from each individual spectrum to remove common spectral features and improve telluric line detection.",
"Tell rm method": "Select the algorithm for telluric removal. Different methods use various mathematical approaches (PCA, polynomial fitting, etc.) to model and remove telluric lines.",
"Use last component selection": "When enabled, uses the number of components from the previous run instead of recalculating. Useful for consistency across multiple processing runs.",
"Save plot": "When enabled, generates and saves diagnostic plots showing the telluric removal process, mean spectra, masks, and PCA components for quality assessment.",
"Smooth size": "Size of the smoothing kernel applied after PCA removal. Larger values create smoother results but may remove fine spectral features. Set to 0 to disable smoothing.",
"Mask night": "Controls which nights have masking applied. Provide one value per night separated by commas. Use 1 to enable masking (recommended), 0 to disable for a specific night.",
"Threshold mean spectrum": "Masks pixels as bad where the channel signal (before continuum correction) is lower than this threshold. Value range: 0 (no signal) to 1 (maximum signal). Recommended: 0.3",
"Cutoff telluric": "Masks pixels as bad where the channel (after continuum correction) is lower than this threshold. Value range: 0 (line saturated) to 1 (continuum level). Recommended: 0.8",
"Sigma Coefficient": "Masks pixels as bad where the channel variance exceeds the mean variance multiplied by this coefficient. Higher values are more permissive. Recommended: 1.5",
"Fixed Component": "Specify a fixed number of PCA components to use for telluric removal. Set to 0 to use automatic component selection. This field is disabled when 'Tangent components method' is enabled.",
"Tangent components method": "When enabled, automatically determines the optimal number of PCA components using the tangent method, which analyzes the variance reduction curve. This overrides the Fixed Component value.",
"Tangent threshold": "Threshold for the tangent method's second derivative test. Lower values select fewer components (more conservative), higher values select more components (more aggressive). Recommended: 0.1",
"N. components substitute": "Fallback number of PCA components to use if the tangent method fails to find an optimal value above the threshold. Recommended: 4",
"Standardize PCA": "When enabled, applies StandardScaler before PCA to standardize the data (mean=0, std=1). This ensures all features have equal variance before PCA decomposition. The mean-centering is already performed in previous steps, so this option only affects the standard deviation normalization. Enable this if you want all spectral channels to have unit variance before PCA.",
"PCA mode:": "Select the PCA decomposition mode. 'spatial' (default): loops over spectral orders, with exposures as samples and pixels as features - removes temporal variations of telluric lines. 'temporal': loops over spectral channels, with pixels as samples and exposures as features - removes spatial patterns across pixels.",
"TELLURIC REMOVAL": "Executes the telluric removal process using the configured parameters. This button spawns a subprocess that applies the selected method to remove telluric contamination from the spectral data. The button is disabled during processing and re-enabled upon completion.",
"Results:": "Displays the output and status messages from the telluric removal process. Shows information about processed nights, number of components used, number of removed pixels."
}
# Create help button
self.help_button = HelpButton(
self.row6, 0, 0,
"Telluric Removal Parameters Help",
help_text,
columnspan=1
)
# Create removal button
self.button_Remove = MyButton(
self.row6, 0, 1, 'TELLURIC REMOVAL', "#9af098",
command=self.remove_tell, columnspan=3
)
def _init_results_display(self):
"""Initialize results display area."""
self.textfield_results = MyTextField(
self.row7, 0, 1, "", "Results:",
color=self.color, columnspan=3, width=100, height=15
)
def _setup_exchange_directory(self):
"""Setup directory for temporary file exchange."""
username = os.environ.get("USER")
self.path_exchange = str(
Path(self.path_default, "GUIBRUSHR", "Files", "Temp_Files_GUI",
username, "Temp_folder")
)
os.system(f"mkdir -p {self.path_exchange}")
Path(self.path_exchange).mkdir(parents=True, exist_ok=True)
def _setup_callbacks(self):
"""Setup callback functions for interactive elements."""
self.dropdown_hr_instrument.set_callback(self.update_hr)
self.dropdown_rad_mode.set_callback(self.update_from_rad_mode)
[docs]
def update_target(self):
"""Update target-related information when target selection changes."""
if self.global_exc == 0:
return
target = self.dropdown_target.get_value()
nights = get_target_nights(
self.path_folder_targets, target,
self.dropdown_rad_mode.get_value(),
self.list_hr_instrument[0],
self.checkbox_nights_sim.get_value()
)
self.target_information = get_target_info(
self.path_folder_targets, target,
self.dropdown_rad_mode.get_value()
)
self.textfield_nights.insert_text(nights)
self.list_hr_instrument = get_target_instruments(
self.path_folder_targets, target, "HR",
self.dropdown_rad_mode.get_value(), Path(__file__).name
)
self.update_from_rad_mode()
[docs]
def update_hr(self):
"""Update high-resolution instrument related information."""
if self.global_exc == 0:
return
instrument = self.dropdown_hr_instrument.get_value()
target_id = self.dropdown_target.get_value()
nights = get_target_nights(
self.path_folder_targets, target_id,
self.dropdown_rad_mode.get_value(),
instrument, 0
)
self.textfield_nights.insert_text(nights)
[docs]
def remove_tell(self):
"""Execute telluric removal process."""
path_instrument = create_path_night(
self.path_folder_targets,
self.dropdown_target.get_value(),
self.dropdown_rad_mode.get_value(),
self.dropdown_hr_instrument.get_value()
)
# Save parameters to exchange file
self._save_parameters_to_exchange(path_instrument)
# Disable remove button and start process
self.button_Remove.button.configure(state="disabled")
self._start_telluric_removal_process()
def _save_parameters_to_exchange(self, path_instrument):
"""Save all parameters to exchange file for processing."""
exchange_dictionary = {
"path_instrument": path_instrument,
"order_selection": array(self.textfield_order_selection.get_text().split(",")),
"nights": array(self.textfield_nights.get_text().split(",")),
"tell_rm_method": self.dropdown_tell_rm_method.get_value(),
"use_last_component_selection": self.checkbox_nfcoo.get_value(),
"sigma_coeff": float(self.textfield_sigma_coeff.get_text()),
"smooth_size": self.entry_smooth_size.get_value(),
"yesmask": self.textfield_yesmask.get_text().split(","),
"min_comp": 0, # self.entry_min_comp.get_value(),
"max_comp": 0, # self.entry_max_comp.get_value(),
"fixed_comp": self.entry_fixed_comp.get_value(),
"save_plot": self.checkbox_save_plot.get_value(),
"limit_phase_t14": self.target_information.limit_phase_t14,
"limit_phase_t12": self.target_information.limit_phase_t12,
"limit_phase_t23": self.target_information.limit_phase_t23,
"cutoff": float(self.entry_cutoff.get_value()),
"tangent_components_method": int(self.checkbox_tangent_components_method.get_value()),
"tangent_value": float(self.entry_tangent_value.get_value()),
"substitute_for_comp": int(self.entry_substitute_for_comp.get_value()),
"threshold_mean_spec": float(self.entry_threshold_mean_spec.get_value()),
"subtraction_of_the_average_spectrum": int(self.checkboc_mean_column.get_value()),
"do_not_fit_continuum": int(self.checkbox_do_not_fit_continuum.get_value()),
"apply_standardize": bool(self.checkbox_standardize_pca.get_value()),
"pca_mode": self.dropdown_pca_mode.get_value() # "spatial" or "temporal"
}
with open(str(Path(self.path_exchange, "exchange.pkl")), "wb") as fo:
pickle.dump(exchange_dictionary, fo)
def _start_telluric_removal_process(self):
"""Start the telluric removal subprocess."""
self.p = subprocess.Popen(
[
"python3",
str(Path(
self.path_default, "GUIBRUSHR", "GUI", "Input_Output_Panels",
"Input_Panels", "TabPanels", "FrameTellRem",
"tell_rem_pca_wl_log_final.py"
)),
str(self.path_exchange),
self.path_default
],
stderr=subprocess.STDOUT,
universal_newlines=True,
)
self.window.after(500, self.checkbox_proc)
[docs]
def checkbox_proc(self):
"""Check the status of the telluric removal process."""
if self.p.poll() is not None:
self.button_Remove.button.configure(state="normal")
if os.path.exists(str(Path(self.path_exchange, "error_tell_rm.pkl"))):
with open(str(Path(self.path_exchange, "error_tell_rm.pkl")), "rb") as fo:
message = pickle.load(fo)
MyFigure(title="Error in tell rem", message=message)
else:
with open(str(Path(self.path_exchange, "string_tell_rem.pkl")), "rb") as fo:
string_tell_rem = pickle.load(fo)
self.textfield_results.insert_text(string_tell_rem)
else:
# Not done yet. Check again later.
self.window.after(500, self.checkbox_proc)
[docs]
def modify_target_list(self, target_list):
"""Update the target list dropdown with new targets."""
self.dropdown_target.clean_widget()
self.dropdown_target = MyDropdown(
self.row1, 0, 0, target_list, "Target:",
color=self.color, columnspan=2
)
self.dropdown_target.set_callback(self.update_target)
self.update_from_rad_mode()
[docs]
def update_from_rad_mode(self):
"""Update components based on radiation mode selection."""
if self.global_exc == 0:
return
target_id = self.dropdown_target.get_value()
rad_mode = self.dropdown_rad_mode.get_value()
self.list_hr_instrument = get_target_instruments(
self.path_folder_targets, target_id, "HR",
rad_mode, Path(__file__).name
)
self.dropdown_hr_instrument.clean_widget()
self.dropdown_hr_instrument = MyDropdown(
self.row1, 0, 5, self.list_hr_instrument,
"HR Instrument:", color=self.color, columnspan=2
)
self.dropdown_hr_instrument.set_callback(self.update_hr)
# # Control subtract average spectrum checkbox based on radiation mode
# if rad_mode == "Emission":
# # Emission: uncheck and disable
# self.checkboc_mean_column.set_value(0)
# self.checkboc_mean_column.checkbox.configure(state="disabled")
# else: # Transmission
# # Transmission: check and enable
# self.checkboc_mean_column.set_value(1)
# self.checkboc_mean_column.checkbox.configure(state="normal")
self.update_hr()
[docs]
def update_night_sim(self):
"""Update night selection based on simulation checkbox."""
if self.global_exc == 0:
return
target_id = self.dropdown_target.get_value()
nights = get_target_nights(
self.path_folder_targets, target_id,
self.dropdown_rad_mode.get_value(),
self.dropdown_hr_instrument.get_value(),
self.checkbox_nights_sim.get_value()
)
self.textfield_nights.insert_text(nights)
[docs]
def toggle_fixed_comp_state(self):
"""Toggle fixed component entry state based on tangent method checkbox."""
if self.checkbox_tangent_components_method.get_value():
# Disable fixed component entry when tangent method is checked
self.entry_fixed_comp.entry.configure(state="disabled")
else:
# Enable fixed component entry when tangent method is unchecked
self.entry_fixed_comp.entry.configure(state="normal")