diff --git a/.gitignore b/.gitignore index 913f905..63a2d65 100644 --- a/.gitignore +++ b/.gitignore @@ -178,3 +178,4 @@ start-devices.sh # Generated data files *.csv +.DS_Store diff --git a/caen-watch_acq_times.py b/caen-watch_acq_times.py new file mode 100644 index 0000000..a01b51f --- /dev/null +++ b/caen-watch_acq_times.py @@ -0,0 +1,524 @@ +#!/usr/bin/env python3 +""" +watch_acq_times.py + +Monitors a DAQ directory (and subfolders) to detect acquisition runs by: + - Printing the acquisition START time when a new settings.xml appears + - Printing the acquisition END time when a dedicated .txt file appears + - On startup, scans existing run folders and syncs them into Google Sheets + - After initial sync, listens for new START/STOP events and updates the sheet +""" + +import os +import sys +import time +import logging +import argparse +import socket +from datetime import datetime +from typing import Optional, List, Tuple, Dict, Any + +from watchdog.observers import Observer +from watchdog.events import FileSystemEventHandler +from pathlib import Path + +# Get computer name from environment variable or user input +COMPUTER_NAME = os.getenv("COMPUTER_NAME") +if not COMPUTER_NAME: + print("COMPUTER_NAME environment variable not set.") + print("You must run 'bash ida-devices/scripts/set-computer-name.sh' to set it.") + exit(1) + +# Ensure libs/ is on the path +PROJECT_ROOT = Path(__file__).resolve().parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +# Import our sheet utilities from libs/ +from libs.google_sheet_utils import GoogleSheet + +# Import our settings_extras from libs/ +from libs.settings_extras import extract_digitizer_info, find_matching_config_files + +# Import our settings_validator from libs/ +from libs.settings_validator import report_parameter_diffs + +# Import CAENpy for power supply interaction +from CAENpy.CAENDesktopHighVoltagePowerSupply import CAENDesktopHighVoltagePowerSupply + +# Magic string constants +SETTINGS_FILENAME = 'settings.xml' +END_FILE_SUFFIX = '.txt' +CONFIG_REF_DIR_NAME = 'CONFIG' + +def clear_line() -> None: + """ + Clear the current terminal line in the console. + Used for updating status output in place. + """ + sys.stdout.write('\r' + ' ' * 80 + '\r') + sys.stdout.flush() + +def is_settings_file(path: Path) -> bool: + """ + Check if the given path is a settings.xml file. + + Parameters: + path (Path): The file path to check. + + Returns: + bool: True if the file is named 'settings.xml', False otherwise. + """ + return path.name.lower() == SETTINGS_FILENAME + +def is_end_file(path: Path) -> bool: + """ + Check if the given path is the expected end file (END_FILE_SUFFIX) for a run. + The end file must be named .txt (case-insensitive). + + Parameters: + path (Path): The file path to check. + + Returns: + bool: True if the file is the expected end file, False otherwise. + """ + return ( + path.suffix.lower() == END_FILE_SUFFIX + and path.stem.lower() == path.parent.name.lower() + ) + +def get_file_modification_time(path: Path) -> Optional[datetime]: + """ + Get the modification time of a file as a datetime object. + + Parameters: + path (Path): The file path. + + Returns: + Optional[datetime]: The modification time as a datetime object, + or None if the file is inaccessible or does not exist. + """ + try: + return datetime.fromtimestamp(path.stat().st_mtime) + except Exception: + return None + + +def ensure_run_row_exists( + sheet: 'GoogleSheet', + run_name: str, + start_dt: Optional[datetime], + stop_dt: Optional[datetime], + refresh: bool = True +) -> None: + """ + Ensure a row for the run exists in the sheet, appending if necessary. + + Parameters: + sheet (GoogleSheet): The GoogleSheet instance. + run_name (str): Name of the run. + start_dt (Optional[datetime]): Start time of the run. + stop_dt (Optional[datetime]): Stop time of the run. + refresh (bool): Whether to refresh the cache before checking/appending. + """ + row = sheet.find_run_row(run_name, refresh=refresh) + if row is None: + sheet.append_run(run_name, start_dt, stop_dt, refresh=refresh) + +def format_config_files(matches: List[str]) -> str: + """ + Format the list of matching config files as a comma-separated string. + + Parameters: + matches (List[str]): List of matching config filenames. + + Returns: + str: Comma-separated config filenames, or empty string if none. + """ + return ','.join(matches) if matches else '' + +def prepare_update_values( + sheet: 'GoogleSheet', + start_dt: Optional[datetime], + stop_dt: Optional[datetime], + digitizer: Optional[str], + config_files: str +) -> Dict[int, Any]: + """ + Prepare the dictionary of values to update in the sheet. + + Parameters: + sheet (GoogleSheet): The GoogleSheet instance. + start_dt (Optional[datetime]): Start time of the run. + stop_dt (Optional[datetime]): Stop time of the run. + digitizer (Optional[str]): Digitizer info string. + config_files (str): Comma-separated config filenames. + + Returns: + Dict[int, Any]: Mapping of column indices to values. + """ + return { + sheet.COL_SETUP: start_dt, + sheet.COL_END: stop_dt, + sheet.COL_DAQ_PC: COMPUTER_NAME, + sheet.COL_DIGITIZER: digitizer, + sheet.COL_CONFIG: config_files + } + +def warn_if_no_config_matches(matches: List[str], run_name: str) -> None: + """ + Log a warning if no matching config files are found. + + Parameters: + matches (List[str]): List of matching config filenames. + run_name (str): Name of the run. + """ + if not matches: + logging.warning(f"⚠️ No matching config files found for {run_name}") + +def process_run_folder( + run_name: str, + run_folder: Path, + sheet: 'GoogleSheet', + config_dir: Path, + start_dt: Optional[datetime] = None, + stop_dt: Optional[datetime] = None, + refresh: bool = True +) -> None: + """ + Process a run folder and atomically update the Google Sheet with all run info. + + Parameters: + run_name (str): Name of the run. + run_folder (Path): Path to the run folder. + sheet (GoogleSheet): GoogleSheet instance for updating the sheet. + config_dir (Path): Path to the config directory. + start_dt (Optional[datetime]): Start time of the run. + stop_dt (Optional[datetime]): Stop time of the run. + refresh (bool): Whether to refresh the cache before updating. + """ + settings_path = run_folder / SETTINGS_FILENAME + ensure_run_row_exists(sheet, run_name, start_dt, stop_dt, refresh=refresh) + digitizer = extract_digitizer_info(str(settings_path)) + # TEMPORARY OVERRIDE: Map DT5730S (31050) to 'caen8ch' for sheet entry + if digitizer == "DT5730S (31050)": + digitizer = "caen8ch" + matches = find_matching_config_files(str(settings_path), str(config_dir)) + config_files = format_config_files(matches) + values = prepare_update_values(sheet, start_dt, stop_dt, digitizer, config_files) + sheet.update_run_row(run_name, values, refresh=refresh) + report_parameter_diffs(str(settings_path), str(config_dir)) + warn_if_no_config_matches(matches, run_name) + +def initial_scan(root_folder: Path) -> List[Tuple[datetime, Optional[datetime], str, Path]]: + """ + Scan the root_folder for run directories and return a list of runs. + Each run is a tuple: (start_dt, stop_dt, run_name, run_folder). + + Parameters: + root_folder (Path): The root directory to scan. + + Returns: + List[Tuple[datetime, Optional[datetime], str, Path]]: List of detected runs. + """ + runs = [] + for dirpath, dirnames, filenames in os.walk(root_folder): + dirpath = Path(dirpath) + # Prevent recursing into hidden subfolders + dirnames[:] = [d for d in dirnames if not d.startswith('.')] + # Skip this folder if it’s hidden + if dirpath.name.startswith('.'): + continue + # Skip if no settings.xml file is present + if SETTINGS_FILENAME not in filenames: + continue + + run_name = dirpath.name + settings_pth = dirpath / SETTINGS_FILENAME + txt_pth = dirpath / f'{run_name}{END_FILE_SUFFIX}' + start_dt = get_file_modification_time(settings_pth) + stop_dt = get_file_modification_time(txt_pth) if txt_pth.exists() else None + + if start_dt: + runs.append((start_dt, stop_dt, run_name, dirpath)) + + runs.sort(key=lambda t: t[0]) + return runs + + +def find_most_recent_active_run( + runs: List[Tuple[datetime, Optional[datetime], str, Path]] +) -> Optional[Tuple[str, Path]]: + """ + Find the most recent run that is still active (no end file). + + Parameters: + runs (List[Tuple[datetime, Optional[datetime], str, Path]]): List of detected runs. + + Returns: + Optional[Tuple[str, Path]]: (run_name, run_folder) of the most recent active run, or None. + """ + active_runs = [(start_dt, run_name, run_folder) + for start_dt, stop_dt, run_name, run_folder in runs + if stop_dt is None] + if not active_runs: + return None + if len(active_runs) > 1: + logging.warning( + f"Multiple active runs detected: {[r[1] for r in active_runs]}. " + "Only the most recent will be updated with power supply info." + ) + # Sort by start_dt descending, pick the most recent + active_runs.sort(reverse=True) + _, run_name, run_folder = active_runs[0] + return run_name, run_folder + +def read_power_supply_info(port: str = '/dev/ttyACM0') -> Tuple[List[int], List[float]]: + """ + Read channel numbers and voltages from the CAEN power supply. + + Parameters: + port (str): Serial port for CAEN device. + + Returns: + Tuple[List[int], List[float]]: List of channel numbers and their voltages. + """ + try: + caen = CAENDesktopHighVoltagePowerSupply(port=port) + channels = [] + voltages = [] + for n_channel, channel in enumerate(caen.channels): + channels.append(n_channel) + voltages.append(channel.V_mon) + return channels, voltages + except Exception as e: + logging.warning(f"Could not read CAEN power supply: {e}") + return [], [] + +# ------------------------------------------------------------------- +# Event handler +# ------------------------------------------------------------------- + +class DAQHandler(FileSystemEventHandler): + """ + Handles file creation events in the DAQ directory and updates the Google Sheet accordingly. + + Specifically, this handler listens for: + - Creation of 'settings.xml' files in any run subfolder, which signals the START of a run. + - Creation of '.txt' files (where the filename matches the parent folder name), which signals the STOP of a run. + + The expected file naming conventions are: + - Run start: Each run folder contains a 'settings.xml' file (case-insensitive). + - Run stop: Each run folder receives a '.txt' file (case-insensitive), where matches the folder name. + + Only file creation events (not modifications) are processed to trigger updates in the Google Sheet. + """ + + def __init__(self, watch_folder: Path, sheet: 'GoogleSheet', config_dir: Path) -> None: + """ + Initialize the DAQHandler. + + Parameters: + watch_folder (Path): The root folder being watched. + sheet (GoogleSheet): The GoogleSheet instance. + config_dir (Path): Path to the config directory. + """ + super().__init__() + self.watch_folder = watch_folder + self.sheet = sheet + self.config_dir = config_dir + + def on_created(self, event: Any) -> None: + """ + Handles the creation of new files in the watched directory. + Delegates to the shared event handler. + + Parameters: + event (Any): The file system event. + """ + self._handle_event(event) + + # If handling file modifications is needed in the future, + # consider implementing an on_modified method similar to on_created. + + def _handle_event(self, event: Any) -> None: + """ + Shared logic for handling both creation and modification events. + + Parameters: + event (Any): The file system event. + """ + if event.is_directory: + return + + path = Path(event.src_path) + name = path.name + + # Ignore hidden files + if name.startswith('.'): + return + + run_folder = path.parent + run_name = run_folder.name + + if is_settings_file(path): + self._handle_run_start(run_name, run_folder, path) + elif is_end_file(path): + self._handle_run_end(run_name, run_folder, path) + + def _handle_run_start(self, run_name: str, run_folder: Path, path: Path) -> None: + """ + Handle the creation or modification of a settings.xml file (run start event). + + Parameters: + run_name (str): Name of the run. + run_folder (Path): Path to the run folder. + path (Path): Path to the settings.xml file. + """ + start_dt = get_file_modification_time(path) + if not start_dt: + return + clear_line() + logging.info(f"START {run_name}: {start_dt:%Y-%m-%d %H:%M:%S}") + process_run_folder( + run_name=run_name, + run_folder=run_folder, + sheet=self.sheet, + config_dir=self.config_dir, + start_dt=start_dt + ) + # Read power supply info and update the sheet + channels, voltages = read_power_supply_info() + update_power_supply_in_sheet(self.sheet, run_name, channels, voltages) + + def _handle_run_end(self, run_name: str, run_folder: Path, path: Path) -> None: + """ + Handle the creation or modification of an end file (run stop event). + + Parameters: + run_name (str): Name of the run. + run_folder (Path): Path to the run folder. + path (Path): Path to the end file. + """ + stop_dt = get_file_modification_time(path) + if not stop_dt: + return + clear_line() + logging.info(f"STOP {run_name}: {stop_dt:%Y-%m-%d %H:%M:%S}") + self.sheet.refresh() + row = self.sheet.find_run_row(run_name, refresh=False) + if row is None: + # In case we missed START + self.sheet.append_run(run_name, None, stop_dt, refresh=False) + else: + settings_path = run_folder / SETTINGS_FILENAME + digitizer = extract_digitizer_info(str(settings_path)) + values = { + self.sheet.COL_END: stop_dt, + self.sheet.COL_DAQ_PC: COMPUTER_NAME, + self.sheet.COL_DIGITIZER: digitizer + } + self.sheet.update_run_row(run_name, values, refresh=False) + +# ------------------------------------------------------------------- +# Main +# ------------------------------------------------------------------- + +def main() -> None: + """ + Main entry point for the DAQ directory watcher and Google Sheets synchronization tool. + + This function performs the following steps: + 1. Parses command-line arguments to determine the DAQ directory to monitor. + 2. Prompts the user for a valid directory if not provided via the command line. + 3. Initializes the GoogleSheet instance for recording run information. + 4. Performs an initial scan of the DAQ directory to detect existing runs and syncs them to the sheet. + 5. Starts a watchdog observer to monitor the directory for new or modified run files, + updating the Google Sheet in real time as START/STOP events are detected. + 6. Displays a spinner in the terminal to indicate the script is actively monitoring. + 7. Handles graceful shutdown on keyboard interrupt. + + This function is the main orchestrator for directory monitoring and data synchronization. + """ + logging.basicConfig(level=logging.INFO, format='%(message)s') + + parser = argparse.ArgumentParser( + description="Watch a DAQ directory and mirror START/STOP times into Google Sheets" + ) + parser.add_argument('watch_folder', nargs='?', help='Root DAQ directory to monitor') + args = parser.parse_args() + + # If not given on the CLI, prompt the user until a valid directory is entered + watch_folder = args.watch_folder + if not watch_folder: + try: + while True: + watch_folder = input("Enter the DAQ folder to monitor (e.g. "caen-master-project-1"): ").strip() + if Path(watch_folder).is_dir(): + break + logging.error(f"❌ '{watch_folder}' is not a valid directory. Please try again.") + except (EOFError, KeyboardInterrupt): + logging.info("No folder provided—exiting.") + sys.exit(1) + + # final check + watch_folder_path = Path(watch_folder) + if not watch_folder_path.is_dir(): + logging.error("Invalid directory: %s", watch_folder) + sys.exit(1) + + sheet = GoogleSheet() + + # Construct config_dir once here + config_dir = watch_folder_path / CONFIG_REF_DIR_NAME + + # Initial directory scan + runs = initial_scan(watch_folder_path) + + # Sync initial scan into the sheet + logging.info("=== Initial Scan & Sheet Sync ===") + for start_dt, stop_dt, run_name, run_folder in runs: + logging.info( + f"SYNC {run_name}: START={start_dt:%Y-%m-%d %H:%M:%S} STOP={stop_dt or '(none)'}" + ) + process_run_folder( + run_name=run_name, + run_folder=run_folder, + sheet=sheet, + config_dir=config_dir, + start_dt=start_dt, + stop_dt=stop_dt, + refresh=False # <--- Only refresh once at the start + ) + + # Find the most recent active run (if any) and update it with power supply info + active = find_most_recent_active_run(runs) + if active: + run_name, run_folder = active + logging.info(f"Active run detected at startup: {run_name}") + channels, voltages = read_power_supply_info() + sheet.insert_power_supply_rows(run_name, channels, voltages) + + # Start live monitoring + handler = DAQHandler(watch_folder_path, sheet, config_dir) + observer = Observer() + observer.schedule(handler, path=str(watch_folder_path), recursive=True) + observer.start() + logging.info(f"\nMonitoring '{watch_folder_path}' for new START/STOP events...") + + # Simple spinner to show liveness (keep using print for spinner) + spinner = ['|', '/', '-', '\\'] + idx = 0 + try: + while True: + sys.stdout.write(f"\r{spinner[idx % len(spinner)]} watching…") + sys.stdout.flush() + idx += 1 + time.sleep(0.2) + except KeyboardInterrupt: + logging.info("Stopping monitor.") + observer.stop() + observer.join() + +if __name__ == '__main__': + main() diff --git a/config/google_sheet_config.json b/config/google_sheet_config.json new file mode 100644 index 0000000..7418cd7 --- /dev/null +++ b/config/google_sheet_config.json @@ -0,0 +1,27 @@ +{ + "spreadsheet_id": "1O-jlIZhBrzKO-lwSM6uMs0TlYdRk_2khoCM9S3b9Mfg", + "sheet_name": "Sheet1", + "header_row": 1, + "columns": { + "id_header": "Experiment ID", + "description_header": "Description", + "digitizer_header": "Digitizer", + "digitizer_channel_number_header": "Digitizer channel number", + "daq_laptop_name_header": "DAQ laptop name", + "compas_config_file_header": "Compas configuration file (digitizer settings)", + "google_drive_data_folders_header": "Google drive data folder name(s)", + "detectors_header": "Detector(s) names and channel number(s)", + "power_supply_header": "Power supply", + "power_supply_channel_header": "Power suppy channel", + "bias_voltage_header": "Bias voltage", + "idle_total_counts_header": "Idle total counts", + "setup_header": "Setup", + "calibration_header": "Calibration", + "background1_header": "Background 1", + "experiment_header": "Experiment", + "background2_header": "Background 2", + "end_header": "End", + "psp_threshold_header": "psp threshold", + "background_analysis_notebook_header": "Background analysis notebook" + } +} diff --git a/libs/google_sheet_utils.py b/libs/google_sheet_utils.py new file mode 100644 index 0000000..8f5fafc --- /dev/null +++ b/libs/google_sheet_utils.py @@ -0,0 +1,357 @@ +#!/usr/bin/env python3 +""" +libs/google_sheet_utils.py + +Google Sheets utilities for run monitoring using gspread. +This version does one initial fetch of the entire sheet into memory, +then services all read operations locally. Only append/update calls +hit the API thereafter. + +Requirements: + - A service-account JSON key at GOOGLE_CREDS (default "sheets_credentials.json") + - A config file at SHEET_CONFIG_PATH (default "config/google_sheet_config.json") + defining: + * spreadsheet_id + * sheet_name + * header_row + * columns.id_header + * columns.google_drive_data_folders_header + * columns.setup_header + * columns.end_header + * columns.daq_laptop_name_header +""" + +import os +import json +import time +from typing import List, Optional, Dict, Any +from datetime import datetime +import logging + +import gspread +from google.oauth2.service_account import Credentials + +class GoogleSheet: + """ + Utility class for interacting with a Google Sheet for DAQ run monitoring. + Handles authentication, reading, appending, and updating rows. + + Note: + This class maintains an in-memory cache of sheet rows (`self.data_rows`) + that is only updated by this process. If the sheet is edited externally + (e.g., by another user or process), the cache may become stale and + inconsistencies may occur. For best results, avoid concurrent edits. + """ + def __init__(self, config_file: str = None, creds_file: str = None): + """ + Initialize the GoogleSheet object, loading config and authenticating. + + Uses google-auth for secure authentication. + + Parameters: + config_file (str, optional): Path to the sheet config JSON file. + creds_file (str, optional): Path to the Google service account credentials JSON file. + """ + config_file = config_file or os.getenv('SHEET_CONFIG_PATH', 'config/google_sheet_config.json') + creds_file = creds_file or os.getenv('GOOGLE_CREDS', 'sheets_credentials.json') + + # Explicitly specify UTF-8 encoding for clarity and future-proofing + try: + with open(config_file, 'r', encoding='utf-8') as cf: + cfg = json.load(cf) + except FileNotFoundError: + raise FileNotFoundError( + f"Config file '{config_file}' not found. Please ensure the path is correct." + ) + except json.JSONDecodeError as e: + raise ValueError( + f"Config file '{config_file}' is not valid JSON: {e}" + ) + except Exception as e: + raise RuntimeError( + f"An unexpected error occurred while reading config file '{config_file}': {e}" + ) + # Authenticate using google-auth and fetch sheet using gspread + scopes = ['https://www.googleapis.com/auth/spreadsheets'] + try: + creds = Credentials.from_service_account_file(creds_file, scopes=scopes) + except FileNotFoundError: + raise FileNotFoundError(f"Google credentials file not found: {creds_file}") + except Exception as e: + raise RuntimeError(f"Failed to load Google credentials from '{creds_file}': {e}") + gc = gspread.authorize(creds) + + # Parse config columns and headers + try: + cols = cfg['columns'] + self.id_header = cols['id_header'] + self.run_header = cols['google_drive_data_folders_header'] + self.setup_header = cols['setup_header'] + self.end_header = cols['end_header'] + self.daq_pc_header = cols['daq_laptop_name_header'] + self.digitizer_header = cols['digitizer_header'] + self.config_header = cols['compas_config_file_header'] + self.power_supply_channel_header = cols['power_supply_channel_header'] + self.bias_voltage_header = cols['bias_voltage_header'] + self.spreadsheet_id = cfg['spreadsheet_id'] + self.sheet_name = cfg['sheet_name'] + self.header_row = cfg['header_row'] + except KeyError as e: + raise KeyError( + f"Missing required key in config file '{config_file}': {e}" + ) + except Exception as e: + raise RuntimeError( + f"An error occurred while parsing config file '{config_file}': {e}" + ) + + # Fetch worksheet and all rows + self.ws = gc.open_by_key(self.spreadsheet_id).worksheet(self.sheet_name) + all_rows = self.ws.get_all_values() + + # Build header map and in-memory rows + self.headers = all_rows[self.header_row - 1] + # Google Sheets columns are 1-based (A=1, B=2, ...) + self.header_to_col = {name: idx + 1 for idx, name in enumerate(self.headers)} + self.COL_ID = self.header_to_col[self.id_header] # 1-based index + self.COL_RUN_NAME = self.header_to_col[self.run_header] # 1-based index + self.COL_SETUP = self.header_to_col[self.setup_header] # 1-based index + self.COL_END = self.header_to_col[self.end_header] # 1-based index + self.COL_DAQ_PC = self.header_to_col[self.daq_pc_header] # 1-based index + self.COL_DIGITIZER = self.header_to_col[self.digitizer_header] # 1-based index + self.COL_CONFIG = self.header_to_col[self.config_header] # 1-based index + self.COL_POWER_SUPPLY_CHANNEL = self.header_to_col.get(self.power_supply_channel_header) # 1-based index + self.COL_BIAS_VOLTAGE = self.header_to_col.get(self.bias_voltage_header) # 1-based index + self.data_rows = all_rows[self.header_row:] + + def _retry_api_call(self, fn, *args, retries: int = 3, delay: float = 1.0, **kwargs): + """ + Retry a Google Sheets API call up to 'retries' times with exponential backoff. + Logs every exception encountered during retries for better debugging. + + Parameters: + fn: The function to call. + *args: Positional arguments for the function. + retries (int): Number of retry attempts. + delay (float): Initial delay between retries in seconds. + **kwargs: Keyword arguments for the function. + + Returns: + The result of the function call if successful. + + Raises: + Exception: The last exception encountered if all retries fail. + """ + last_exc = None + for attempt in range(1, retries + 1): + try: + return fn(*args, **kwargs) + except Exception as e: + logging.warning( + f"API call failed on attempt {attempt}/{retries}: {e}" + ) + last_exc = e + time.sleep(delay) + delay *= 2 # Exponential backoff + logging.error(f"API call failed after {retries} attempts.") + raise last_exc + + def refresh(self) -> None: + """ + Refresh the in-memory cache of sheet rows from the Google Sheet. + + This method should be called to ensure the cache reflects the latest + state of the sheet, especially if external edits may have occurred. + """ + all_rows = self.ws.get_all_values() + self.headers = all_rows[self.header_row - 1] + self.header_to_col = {name: idx + 1 for idx, name in enumerate(self.headers)} + self.data_rows = all_rows[self.header_row:] + + def find_run_row(self, run_name: str, refresh: bool = True) -> Optional[int]: + """ + Find the row index for a given run name. + + Parameters: + run_name (str): The run name to search for. + refresh (bool): Whether to refresh the cache before searching. + + Returns: + Optional[int]: The row index if found, else None. + """ + if refresh: + self.refresh() + for sheet_row, row in enumerate(self.data_rows, start=self.header_row + 1): + if (row[self.COL_RUN_NAME - 1].strip() == run_name + and row[self.COL_ID - 1].strip()): + return sheet_row + return None + + def append_run(self, run_name: str, setup_dt: datetime, end_dt: Optional[datetime] = None, refresh: bool = True) -> None: + """ + Append a new run to the sheet with the given setup and end times. + + Parameters: + run_name (str): The run name. + setup_dt (datetime): The setup/start time. + end_dt (Optional[datetime]): The end time, if available. + refresh (bool): Whether to refresh the cache before appending. + """ + if refresh: + self.refresh() + next_id = self._get_next_id() + new_row = self._build_new_row(run_name, setup_dt, end_dt, next_id) + self._retry_api_call(self.ws.append_row, new_row, value_input_option='RAW') + self.data_rows.append(new_row) + + def _get_next_id(self) -> int: + """ + Get the next available integer ID for a new run. + """ + existing_ids = [ + int(r[self.COL_ID - 1]) for r in self.data_rows + if r[self.COL_ID - 1].isdigit() + ] + return max(existing_ids) + 1 if existing_ids else 1 + + def _build_new_row(self, run_name: str, setup_dt: datetime, end_dt: Optional[datetime], next_id: int) -> List[str]: + """ + Build a new row for appending to the sheet. + """ + new_row = [''] * len(self.headers) + new_row[self.COL_ID - 1] = str(next_id) + new_row[self.COL_RUN_NAME - 1] = run_name + if setup_dt: + new_row[self.COL_SETUP - 1] = setup_dt.strftime('%Y-%m-%d %H:%M:%S') + if end_dt: + new_row[self.COL_END - 1] = end_dt.strftime('%Y-%m-%d %H:%M:%S') + return new_row + + + def update_run_row(self, run_name: str, values: Dict[int, Any], refresh: bool = True) -> None: + """ + Atomically update multiple fields for a run in the sheet. + + Parameters: + run_name (str): The run name to update. + values (Dict[int, Any]): Mapping of column indices to new values. + refresh (bool): Whether to refresh the cache before updating. + """ + if refresh: + self.refresh() + row_idx = self.find_run_row(run_name, refresh=False) + if row_idx is None: + return + mem_idx = row_idx - self.header_row - 1 + row = self.data_rows[mem_idx] + updated = self._update_row_in_memory(row, values) + if updated: + self._push_row_update(row_idx, row) + + def _update_row_in_memory(self, row: List[str], values: Dict[int, Any]) -> bool: + """ + Update the in-memory row with the provided values. + + Parameters: + row (List[str]): The row to update. + values (Dict[int, Any]): Mapping of column indices to new values. + + Returns: + bool: True if any value was updated, False otherwise. + """ + updated = False + for col_idx, value in values.items(): + if value is None: + continue + if isinstance(value, datetime): + value = value.strftime('%Y-%m-%d %H:%M:%S') + elif not isinstance(value, str): + value = str(value) + if not row[col_idx - 1].strip(): + row[col_idx - 1] = value + updated = True + return updated + + @staticmethod + def col_idx_to_letter(idx: int) -> str: + """ + Convert a 1-based column index to Excel-style column letters. + + Parameters: + idx (int): 1-based column index (e.g., 1 -> 'A', 27 -> 'AA') + + Returns: + str: Corresponding column letters. + """ + letters = "" + while idx > 0: + idx, remainder = divmod(idx - 1, 26) + letters = chr(65 + remainder) + letters + return letters + + def _push_row_update(self, row_idx: int, row: List[str]) -> None: + """ + Push the updated row to the Google Sheet. + + Parameters: + row_idx (int): The row index in the sheet. + row (List[str]): The updated row data. + """ + end_col_letter = self.col_idx_to_letter(len(row)) + self._retry_api_call( + self.ws.update, + f"A{row_idx}:{end_col_letter}{row_idx}", + [row] + ) + + def insert_power_supply_rows( + self, + run_name: str, + channels: List[int], + voltages: List[float] + ) -> None: + """ + Insert power supply channel and voltage info as new rows directly below the run entry. + + Each channel/voltage pair is written in its own row, in the respective columns. + If power supply rows already exist, insert new rows between the run row and the next run entry. + + Parameters: + run_name (str): Name of the run. + channels (List[int]): List of channel numbers. + voltages (List[float]): List of voltages for each channel. + + Edge Cases: + - If columns are missing, logs a warning and skips update. + - If no run row is found, logs a warning and skips update. + """ + self.refresh() + run_row_idx = self.find_run_row(run_name, refresh=False) + if run_row_idx is None: + logging.warning(f"Run '{run_name}' not found in sheet; cannot insert power supply rows.") + return + + col_channel = self.COL_POWER_SUPPLY_CHANNEL + col_voltage = self.COL_BIAS_VOLTAGE + if col_channel is None or col_voltage is None: + logging.warning("Power supply columns not found in sheet headers.") + return + + new_rows = [] + for ch, v in zip(channels, voltages): + row = [''] * len(self.headers) + row[col_channel - 1] = str(ch) + row[col_voltage - 1] = f"{v:.2f}" + new_rows.append(row) + + insert_idx = run_row_idx + 1 + for i, r in enumerate(self.data_rows[insert_idx - self.header_row - 1:], start=insert_idx): + if r[self.COL_ID - 1].strip(): + break + insert_idx += 1 + + for offset, row in enumerate(new_rows): + self._retry_api_call(self.ws.insert_row, row, index=insert_idx + offset) + self.data_rows.insert(insert_idx - self.header_row - 1 + offset, row) + + logging.info(f"Inserted {len(new_rows)} power supply rows below run '{run_name}'.") diff --git a/libs/settings_extras.py b/libs/settings_extras.py new file mode 100644 index 0000000..5bc8156 --- /dev/null +++ b/libs/settings_extras.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +""" +settings_extras.py + +Utilities to pull extra metadata out of settings.xml. +Now prints a console warning if digitizer info can’t be extracted. +""" + +import xml.etree.ElementTree as ET +from pathlib import Path +import logging +from typing import Optional, List, Tuple, Dict +from libs.settings_validator import _load_parameter_map + +BOARD_TAG = 'board' +MODEL_NAME_TAG = 'modelName' +SERIAL_NUMBER_TAG = 'serialNumber' + + +def extract_digitizer_info(settings_path: str) -> Optional[str]: + """ + Extract digitizer model and serial number from a settings.xml file. + + Parameters: + settings_path (str): Path to the settings.xml file. + + Returns: + Optional[str]: String in the format "modelName (serialNumber)" or None if extraction fails. + """ + p = Path(settings_path) + if not p.exists(): + logging.warning(f"⚠️ Digitizer info: file not found: {settings_path}") + return None + + if p.stat().st_size == 0: + logging.warning(f"⚠️ Digitizer info: file is empty: {settings_path}") + return None + + root = _parse_settings_xml(str(p)) + if root is None: + return None + + return _extract_board_info(root, settings_path) + +def find_matching_config_files(settings_path: str, config_folder: str) -> List[str]: + """ + Find XML files in config_folder whose exactly match those in settings_path. + + Parameters: + settings_path (str): Path to the settings.xml file. + config_folder (str): Path to the folder containing reference config XMLs. + + Returns: + List[str]: List of matching config filenames. + """ + sfile = Path(settings_path) + cdir = Path(config_folder) + if not sfile.is_file() or not cdir.is_dir(): + return [] + try: + s_map, _ = _load_parameter_map(sfile) + except Exception as e: + logging.warning(f"⚠️ Cannot load parameters from {settings_path}: {e}") + return [] + + refs = _load_reference_parameter_maps(cdir) + matches = [ref.name for ref, r_map in refs if _is_exact_match(s_map, r_map)] + return matches + +def _parse_settings_xml(settings_path: str) -> Optional[ET.Element]: + """ + Parse the settings.xml file and return the root element. + + Parameters: + settings_path (str): Path to the settings.xml file. + + Returns: + Optional[ET.Element]: Root XML element or None if parsing fails. + """ + try: + tree = ET.parse(settings_path) + return tree.getroot() + except ET.ParseError: + logging.warning(f"⚠️ Digitizer info: malformed XML: {settings_path}") + except Exception as e: + logging.warning(f"⚠️ Digitizer info: error reading '{settings_path}': {e}") + return None + +def _extract_board_info(root: ET.Element, settings_path: str) -> Optional[str]: + """ + Extract model and serial number from the board element. + + Parameters: + root (ET.Element): Root XML element. + settings_path (str): Path to the settings.xml file. + + Returns: + Optional[str]: Formatted digitizer info or None if missing. + """ + board = root.find(BOARD_TAG) + if board is None: + logging.warning(f"⚠️ Digitizer info: missing <{BOARD_TAG}> element in {settings_path}") + return None + + model = (board.findtext(MODEL_NAME_TAG) or '').strip() + serial = (board.findtext(SERIAL_NUMBER_TAG) or '').strip() + if not model or not serial: + logging.warning(f"⚠️ Digitizer info: missing {MODEL_NAME_TAG} or {SERIAL_NUMBER_TAG} in {settings_path}") + return None + + return f"{model} ({serial})" + +def _load_reference_parameter_maps(config_dir: Path) -> List[Tuple[Path, Dict[str, str]]]: + """ + Load parameter maps from all reference XML files in the config directory. + + Parameters: + config_dir (Path): Path to the config directory. + + Returns: + List[Tuple[Path, Dict[str, str]]]: List of (Path, parameter map) tuples. + """ + refs = [] + for ref in sorted(config_dir.glob('*.xml')): + try: + r_map, _ = _load_parameter_map(ref) + refs.append((ref, r_map)) + except Exception as e: + # Log the exception for better traceability and debugging + logging.warning( + f"⚠️ Failed to load parameter map from '{ref}': {e}" + ) + return refs + +def _is_exact_match(s_map: Dict[str, str], r_map: Dict[str, str]) -> bool: + """ + Check if two parameter maps are an exact match. + + Parameters: + s_map (Dict[str, str]): Settings parameter map. + r_map (Dict[str, str]): Reference parameter map. + + Returns: + bool: True if maps are equal, False otherwise. + """ + return r_map == s_map \ No newline at end of file diff --git a/libs/settings_validator.py b/libs/settings_validator.py new file mode 100644 index 0000000..cfa32e7 --- /dev/null +++ b/libs/settings_validator.py @@ -0,0 +1,267 @@ +# libs/settings_validator.py + +import xml.etree.ElementTree as ET +from pathlib import Path +import logging +from typing import Dict, List, Tuple + + +IGNORE_TAG_RUN_ID = 'runId' +PARAMETERS_TAG = 'parameters' +ENTRY_TAG = 'entry' +KEY_TAG = 'key' +VALUE_TAG = 'value' + +def report_parameter_diffs( + settings_path: str, + config_folder: str, + max_diffs: int = 3 +) -> None: + """ + Compare in settings.xml against each .xml in config_folder. + Print grouped results and up to max_diffs per file. + + Parameters: + settings_path (str): Path to the settings.xml file. + config_folder (str): Path to the folder with reference XMLs. + max_diffs (int): Maximum number of diff lines to print per file. + """ + sfile = Path(settings_path) + cdir = Path(config_folder) + + # Pre-checks + if not sfile.is_file() or sfile.stat().st_size == 0: + logging.warning(f"⚠️ Skipping diff: '{settings_path}' is missing or empty.") + return + if not cdir.is_dir(): + logging.warning(f"⚠️ Config folder not found: {config_folder}") + return + + refs = sorted(cdir.glob('*.xml')) + if not refs: + logging.warning(f"⚠️ No reference XMLs in {cdir}") + return + + # Load master parameter map + s_map, s_lines = _load_parameter_map(sfile) + if not s_map: + logging.warning(f"⚠️ Skipping diff: '{settings_path}' could not be parsed or has no parameters.") + return + + # Build a reverse lookup: value -> line number(s) + value_to_lineno = {} + for i, ln in enumerate(s_lines): + # Try to extract value from line + if '' in ln: + val = ln.split('>')[-2].split('<')[0].strip() + if val: + value_to_lineno.setdefault(val, []).append(i + 1) + + matches: List[str] = [] + diffs_map: Dict[str, List[Tuple[int, str, str, str]]] = {} + + for ref in refs: + r_map, _ = _load_parameter_map(ref) + diffs: List[Tuple[int, str, str, str]] = [] + + # Report parameters present in current but different or missing in reference + for key, new_val in s_map.items(): + old_val = r_map.get(key) + if old_val is None: + # Parameter missing in reference + lineno = value_to_lineno.get(new_val, [None])[0] + if lineno: + diffs.append((lineno, key, "(missing in reference)", new_val)) + elif old_val != new_val: + lineno = value_to_lineno.get(new_val, [None])[0] + if lineno: + diffs.append((lineno, key, old_val, new_val)) + + # Report parameters present in reference but missing in current + for key, old_val in r_map.items(): + if key not in s_map: + # Parameter missing in current file + diffs.append((None, key, old_val, "(missing in current)")) + + if not diffs: + matches.append(ref.name) + else: + diffs_map[ref.name] = diffs + # Print grouped results + if matches: + logging.info(f"✅ Exact matches: {', '.join(matches)}") + if diffs_map: + logging.warning(f"⚠️ Differences detected in: {', '.join(diffs_map.keys())}") + for ref_name, diffs in diffs_map.items(): + logging.info(f"**{ref_name}**") + for diff in diffs[:max_diffs]: + lineno, key, old, new = diff + if lineno is not None: + logging.info(f" L{lineno:<4} {key:<24}: '{old}' → '{new}'") + else: + logging.info(f" {key:<24}: '{old}' → '{new}'") + more = len(diffs) - max_diffs + if more > 0: + logging.info(f" ...and {more} more differences...") + +def _extract_sections(xml_path: Path) -> Dict: + """ + Extract relevant sections from an XML file for comparison. + + Parameters: + xml_path (Path): Path to the XML file. + + Returns: + Dict: Dictionary of extracted sections, or empty dict if parsing fails. + """ + try: + tree = ET.parse(xml_path) + root = tree.getroot() + except ET.ParseError as e: + logging.warning(f"⚠️ Malformed XML in '{xml_path}': {e}") + return {} + except Exception as e: + logging.warning(f"⚠️ Error reading XML in '{xml_path}': {e}") + return {} + + out: Dict = {} + out.update(_extract_board_section(root)) + out['parameters'] = _extract_parameters_section(root) + out['channels'] = _extract_channels_section(root) + out.update(_extract_subtrees(root, ( + 'acquisitionMemento', 'timeCorrelationMemento', 'virtualChannelsMemento' + ))) + return out + +def _extract_board_section(root: ET.Element) -> Dict[str, str]: + """ + Extract board information from the XML root. + + Parameters: + root (ET.Element): The root XML element. + + Returns: + Dict[str, str]: Board fields and their values. + """ + out = {} + board = root.find('board') + if board is not None: + for tag in ('id', 'modelName', 'serialNumber', 'label'): + el = board.find(tag) + out[f'board.{tag}'] = (el.text or '').strip() + return out + +def _extract_parameters_section(root: ET.Element) -> Dict[str, str]: + """ + Extract parameters from the XML root. + + Parameters: + root (ET.Element): The root XML element. + + Returns: + Dict[str, str]: Parameter keys and values. + """ + params = {} + for entry in root.findall(f'.//{PARAMETERS_TAG}/{ENTRY_TAG}'): + key = entry.findtext(KEY_TAG, '').strip() + val_el = entry.find(VALUE_TAG) + if val_el is not None: + nested = val_el.findtext(VALUE_TAG) + val = nested if nested is not None else (val_el.text or '') + else: + val = '' + params[key] = val.strip() + return params + +def _extract_channels_section(root: ET.Element) -> Dict[str, str]: + """ + Extract channel information from the XML root. + + Parameters: + root (ET.Element): The root XML element. + + Returns: + Dict[str, str]: Channel IDs and their XML string. + """ + chans = {} + for ch in root.findall('.//virtualChannelsMemento/channel'): + ch_id = ch.get('id') + chans[ch_id] = ET.tostring(ch, encoding='unicode') + return chans + +def _extract_subtrees(root: ET.Element, sections: Tuple[str, ...]) -> Dict[str, str]: + """ + Extract raw XML for specified subtrees. + + Parameters: + root (ET.Element): The root XML element. + sections (Tuple[str, ...]): Section names to extract. + + Returns: + Dict[str, str]: Section names and their XML string. + """ + out = {} + for section in sections: + el = root.find(section) + out[section] = ET.tostring(el, encoding='unicode') if el is not None else '' + return out + + +def _extract_simple_fields_from_subtree( + xml_text: str, + wanted_fields: Tuple[str, ...] = ('runId',) +) -> Dict[str, str]: + """ + Extract simple fields from an XML subtree. + + Parameters: + xml_text (str): XML text to parse. + wanted_fields (Tuple[str, ...]): Fields to extract. + + Returns: + Dict[str, str]: Mapping of field names to values. + """ + if not xml_text: + return {} + try: + root = ET.fromstring(xml_text) + except ET.ParseError: + return {} + out = {} + for child in root: + if child.tag in wanted_fields: + out[child.tag] = (child.text or '').strip() + return out + +def _load_parameter_map(xml_path: Path) -> Tuple[Dict[str, str], List[str]]: + """ + Parse the XML at xml_path and return a parameter map and raw file lines. + + Parameters: + xml_path (Path): Path to the XML file. + + Returns: + Tuple[Dict[str, str], List[str]]: Parameter map and list of file lines. + """ + text = xml_path.read_text(encoding='utf-8').splitlines() + try: + # Use ET.parse for efficiency and reliability + tree = ET.parse(xml_path) + root = tree.getroot() + except ET.ParseError as e: + logging.warning(f"⚠️ Malformed XML in '{xml_path}': {e}") + return {}, text + except Exception as e: + logging.warning(f"⚠️ Error reading XML in '{xml_path}': {e}") + return {}, text + params: Dict[str, str] = {} + for entry in root.findall(f'.//{PARAMETERS_TAG}/{ENTRY_TAG}'): + key = entry.findtext(KEY_TAG, '').strip() + val_el = entry.find(VALUE_TAG) + if val_el is not None: + nested = val_el.findtext(VALUE_TAG) + val = (nested if nested is not None else val_el.text or '').strip() + else: + val = '' + params[key] = val + return params, text \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index 1409882..a543950 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,3 +10,6 @@ requests psycopg2 PyQt5 uproot +gspread +google-auth +CAENpy