Data Loader

data_loader

Módulo de carga de datos XPS

Este módulo maneja la importación de datos desde diferentes instrumentos XPS y formatos de archivo.

Classes

XPSDataset

Bases: XPSBaseModel

Representa un archivo XPS completo.

Methods:

validate_filename classmethod

validate_filename(v: str) -> str

Validar que el nombre de archivo no esté vacío.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("filename")
@classmethod
def validate_filename(cls, v: str) -> str:
    """Validar que el nombre de archivo no esté vacío."""
    if not v or not v.strip():
        raise ValueError("filename no puede estar vacío")
    return v

validate_spectra classmethod

validate_spectra(v: dict[str, XPSSpectrum]) -> dict[str, XPSSpectrum]

Validar que el diccionario de espectros no esté vacío.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("spectra")
@classmethod
def validate_spectra(cls, v: dict[str, XPSSpectrum]) -> dict[str, XPSSpectrum]:
    """Validar que el diccionario de espectros no esté vacío."""
    if not v:
        raise ValueError(
            "spectra no puede estar vacío - debe contener al menos un espectro"
        )
    return v

get_spectrum

get_spectrum(region_name: str) -> XPSSpectrum | None

Obtiene un espectro específico.

Source code in src/xps_analyzer/data_loader/core.py

def get_spectrum(self, region_name: str) -> XPSSpectrum | None:
    """Obtiene un espectro específico."""
    return self.spectra.get(region_name)

list_regions

list_regions() -> list

Lista todas las regiones disponibles.

Source code in src/xps_analyzer/data_loader/core.py

def list_regions(self) -> list:
    """Lista todas las regiones disponibles."""
    return list(self.spectra.keys())

XPSSample

Bases: XPSBaseModel

Representa una muestra XPS que puede contener múltiples archivos.

Methods:

validate_sample_name classmethod

validate_sample_name(v: str) -> str

Validar que el nombre de muestra no esté vacío.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("sample_name")
@classmethod
def validate_sample_name(cls, v: str) -> str:
    """Validar que el nombre de muestra no esté vacío."""
    if not v or not v.strip():
        raise ValueError("sample_name no puede estar vacío")
    return v

validate_datasets classmethod

validate_datasets(v: dict[str, XPSDataset]) -> dict[str, XPSDataset]

Validar que el diccionario de datasets no esté vacío.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("datasets")
@classmethod
def validate_datasets(cls, v: dict[str, XPSDataset]) -> dict[str, XPSDataset]:
    """Validar que el diccionario de datasets no esté vacío."""
    if not v:
        raise ValueError(
            "datasets no puede estar vacío - debe contener al menos un dataset"
        )
    return v

get_dataset

get_dataset(filename: str) -> XPSDataset | None

Obtiene un dataset específico.

Parameters

filename : str Nombre del archivo del dataset a obtener. Returns

---

XPSDataset | None El dataset correspondiente o None si no existe.

Source code in src/xps_analyzer/data_loader/core.py

def get_dataset(self, filename: str) -> XPSDataset | None:
    """Obtiene un dataset específico.

    Parameters
    ----------
    filename : str
        Nombre del archivo del dataset a obtener.
    Returns
    -------
    XPSDataset | None
        El dataset correspondiente o None si no existe.
    """
    return self.datasets.get(filename)

list_datasets

list_datasets() -> list

Lista todos los datasets disponibles.

Returns

list Lista de nombres de archivos de los datasets.

Source code in src/xps_analyzer/data_loader/core.py

def list_datasets(self) -> list:
    """Lista todos los datasets disponibles.

    Returns
    -------
    list
        Lista de nombres de archivos de los datasets.
    """
    return list(self.datasets.keys())

XPSSpectrum

Bases: XPSBaseModel

Representa un espectro XPS individual.

Attributes

data property

data: DataFrame

Retorna los datos como DataFrame.

Methods:

validate_arrays classmethod

validate_arrays(v: ndarray) -> np.ndarray

Validar que los campos sean arrays de NumPy y no estén vacíos.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("binding_energy", "intensity")
@classmethod
def validate_arrays(cls, v: np.ndarray) -> np.ndarray:
    """Validar que los campos sean arrays de NumPy y no estén vacíos."""
    if not isinstance(v, np.ndarray):
        raise TypeError(
            f"El campo debe ser un numpy.ndarray, no {type(v).__name__}"
        )
    if len(v) == 0:
        raise ValueError("Los arrays no pueden estar vacíos")
    return v

validate_positive_energies classmethod

validate_positive_energies(v: ndarray) -> np.ndarray

Validar que las energías sean positivas.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("binding_energy")
@classmethod
def validate_positive_energies(cls, v: np.ndarray) -> np.ndarray:
    """Validar que las energías sean positivas."""
    if np.any(v < 0):
        raise ValueError("Los valores de binding_energy deben ser positivos")
    return v

validate_region_name classmethod

validate_region_name(v: str) -> str

Validar que el nombre de la región no esté vacío.

Source code in src/xps_analyzer/data_loader/core.py

@field_validator("region_name")
@classmethod
def validate_region_name(cls, v: str) -> str:
    """Validar que el nombre de la región no esté vacío."""
    if not v or not v.strip():
        raise ValueError("region_name no puede estar vacío")
    return v

validate_matching_lengths

validate_matching_lengths() -> XPSSpectrum

Validar que los arrays tengan la misma longitud.

Source code in src/xps_analyzer/data_loader/core.py

@model_validator(mode="after")
def validate_matching_lengths(self) -> XPSSpectrum:
    """Validar que los arrays tengan la misma longitud."""
    if len(self.binding_energy) != len(self.intensity):
        raise ValueError(
            f"binding_energy ({len(self.binding_energy)} puntos) e intensity "
            f"({len(self.intensity)} puntos) deben tener la misma longitud"
        )
    return self

Functions:

detect_file_format

detect_file_format(filepath: str | Path) -> str | None

Detecta automáticamente el formato del archivo XPS.

Parameters

filepath : str or Path Ruta al archivo a analizar.

Returns

str | None Tipo de formato detectado: - "multiplex": Formato multiplex propietario - "survey": Formato survey simple - "text": Formato de texto genérico - None: Formato no reconocido

Raises

FileNotFoundError Si el archivo no existe.

Examples

fmt = detect_file_format("data/sample_multiplex.txt") print(f"Formato detectado: {fmt}") Formato detectado: multiplex

Source code in src/xps_analyzer/data_loader/core.py

def detect_file_format(filepath: str | Path) -> str | None:
    """
    Detecta automáticamente el formato del archivo XPS.

    Parameters
    ----------
    filepath : str or Path
        Ruta al archivo a analizar.

    Returns
    -------
    str | None
        Tipo de formato detectado:
        - "multiplex": Formato multiplex propietario
        - "survey": Formato survey simple
        - "text": Formato de texto genérico
        - None: Formato no reconocido

    Raises
    ------
    FileNotFoundError
        Si el archivo no existe.

    Examples
    --------
    >>> fmt = detect_file_format("data/sample_multiplex.txt")
    >>> print(f"Formato detectado: {fmt}")
    Formato detectado: multiplex
    """
    filepath = Path(filepath)

    # Validar que el archivo existe
    if not filepath.exists():
        raise FileNotFoundError(f"Archivo no encontrado: {filepath}")

    try:
        # Leer primeras 10 líneas para análisis
        with open(filepath, encoding="utf-8", errors="ignore") as f:
            first_lines = [f.readline().strip() for _ in range(10)]

        # Concatenar para búsqueda
        content = "\n".join(first_lines)

        # Detección por nombre de archivo
        filename_lower = filepath.name.lower()
        if "multiplex" in filename_lower:
            return "multiplex"

        # Detección por estructura de contenido
        # Formato multiplex tiene múltiples secciones "Element"
        element_count = content.count("Element")
        if element_count >= 2:
            return "multiplex"

        # Si tiene separadores ";" típicos del formato propietario
        if any(";" in line for line in first_lines):
            # Si solo hay una sección, es survey
            if element_count <= 1:
                return "survey"
            return "text"

        # Formato no reconocido
        return None

    except UnicodeDecodeError:
        # Archivo binario - formato no soportado
        return None
    except Exception:
        return None

load_all_data

load_all_data(data_path: str | Path, recursive: bool = True) -> dict[str, XPSDataset]

Carga todos los archivos de datos XPS desde un directorio.

Parameters

data_path : str or Path Ruta al directorio que contiene los datos XPS. recursive : bool, default=True Si True, busca archivos recursivamente en subdirectorios.

Returns

dict[str, XPSDataset] Diccionario con los datos cargados. Las claves son los nombres de archivo y los valores son los XPSDataset procesados.

Raises

FileNotFoundError Si el directorio no existe. ValueError Si la ruta no es un directorio.

Examples

data = load_all_data("data/raw/samples/") print(f"Cargados {len(data)} archivos") for filename, dataset in data.items(): ... print(f"{filename}: {len(dataset.spectra)} espectros")

Source code in src/xps_analyzer/data_loader/core.py

def load_all_data(
    data_path: str | Path, recursive: bool = True
) -> dict[str, XPSDataset]:
    """
    Carga todos los archivos de datos XPS desde un directorio.

    Parameters
    ----------
    data_path : str or Path
        Ruta al directorio que contiene los datos XPS.
    recursive : bool, default=True
        Si True, busca archivos recursivamente en subdirectorios.

    Returns
    -------
    dict[str, XPSDataset]
        Diccionario con los datos cargados. Las claves son los nombres
        de archivo y los valores son los XPSDataset procesados.

    Raises
    ------
    FileNotFoundError
        Si el directorio no existe.
    ValueError
        Si la ruta no es un directorio.

    Examples
    --------
    >>> data = load_all_data("data/raw/samples/")
    >>> print(f"Cargados {len(data)} archivos")
    >>> for filename, dataset in data.items():
    ...     print(f"{filename}: {len(dataset.spectra)} espectros")
    """
    directory = Path(data_path)

    # Validar que el directorio existe
    if not directory.exists():
        raise FileNotFoundError(f"Directorio no encontrado: {directory}")

    if not directory.is_dir():
        raise ValueError(f"La ruta no es un directorio: {directory}")

    datasets = {}
    errors = []

    # Buscar archivos recursivamente o no
    if recursive:
        pattern = directory.rglob("*.txt")
    else:
        pattern = directory.glob("*.txt")

    # Cargar cada archivo encontrado
    for filepath in pattern:
        try:
            dataset = load_single_file(filepath)
            datasets[filepath.name] = dataset
        except Exception as e:
            # Guardar errores pero continuar con otros archivos
            errors.append((filepath.name, str(e)))

    # Reportar errores al usuario si hubo alguno
    if errors:
        print(f"Advertencia: {len(errors)} archivo(s) no pudieron cargarse:")
        for filename, error in errors[:5]:  # Mostrar solo primeros 5
            print(f"  - {filename}: {error}")
        if len(errors) > 5:
            print(f"  ... y {len(errors) - 5} más")

    return datasets

load_single_file

load_single_file(filepath: str | Path) -> XPSDataset

Carga un solo archivo de datos XPS. Parameters

---

filepath : str or Path Ruta al archivo de datos XPS. Returns

---

XPSDataset Objeto XPSDataset con los datos cargados.

Source code in src/xps_analyzer/data_loader/core.py

def load_single_file(filepath: str | Path) -> XPSDataset:
    """
    Carga un solo archivo de datos XPS.
    Parameters
    ----------
    filepath : str or Path
        Ruta al archivo de datos XPS.
    Returns
    -------
    XPSDataset
        Objeto XPSDataset con los datos cargados.
    """
    survey = True
    header = {}

    if isinstance(filepath, str):
        filepath = Path(filepath)

    with open(filepath, encoding="latin-1") as file:
        data = file.readlines()
        data = [line.strip() for line in data if line.strip()]

        if "multiplex" in filepath.name.lower():
            survey = False
            print("Archivo multiplex detectado")
            header = parse_metadata(data[:3], header=True)
            data = data[3:]

        if survey:
            spectrum = get_spectrum_data(data)
            dataset = XPSDataset(
                filename=filepath.name, header=header, spectra={"survey": spectrum}
            )
        else:
            # Procesar múltiples regiones
            spectra = {}
            i = 0
            while i < len(data):
                if data[i].startswith("Element"):
                    region_lines = []
                    # Recolectar todas las líneas hasta la siguiente región o EOF
                    while i < len(data) and (
                        not data[i].startswith("Element") or len(region_lines) == 0
                    ):
                        region_lines.append(data[i])
                        i += 1
                    spectrum = get_spectrum_data(region_lines)
                    spectra[spectrum.region_name] = spectrum
                else:
                    i += 1
            dataset = XPSDataset(filename=filepath.name, header=header, spectra=spectra)

    return dataset