import logging
import os
import zipfile
from dataclasses import dataclass, field
import pandas as pd
from . import utils
from .config import ENCODING, MEASUREMENTS_DIR, XRD_DATA_COLUMNS, ZIP_COMPRESS_LEVEL
from .measurement import Measurement
from .meta import Meta
from .paths import MeasurementPaths
logger = logging.getLogger(__name__)
[docs]def create_zip_archive(file_path: str, archive_path: str) -> None:
"""Create a zip archive containing the specified file."""
with zipfile.ZipFile(
archive_path, "w", zipfile.ZIP_DEFLATED, compresslevel=ZIP_COMPRESS_LEVEL
) as zf:
zf.write(file_path, arcname=os.path.basename(file_path))
logger.debug(f"Created archive: {archive_path}")
[docs]def read_from_ascii(
file_path: str,
delimiter: str = " ",
col_angle: str = XRD_DATA_COLUMNS["angle"],
col_intensity: str = XRD_DATA_COLUMNS["int_abs"],
encoding: str = ENCODING,
) -> pd.Series:
"""Read XRD data from an ASCII file and return as Pandas Series."""
df = pd.read_csv(
file_path,
delimiter=delimiter,
header=None,
names=[col_angle, col_intensity],
skipinitialspace=True,
index_col=col_angle,
encoding=encoding,
)
return df[col_intensity]
[docs]@dataclass
class MeasurementManager:
"""Measurement manager.
:encoding: Encoding used to read from and write to files.
"""
measurements_dir: str = MEASUREMENTS_DIR
encoding: str = ENCODING
def __post_init__(self):
self._create_measurements_dir()
def _create_measurements_dir(self) -> None:
"""Create measurements_dir if not existing."""
if not os.path.isdir(self.measurements_dir):
os.makedirs(self.measurements_dir, exist_ok=False)
logging.info(f"Created measurement directory: '{self.measurements_dir}'")
[docs] def list_measurements(self, sort_reverse=True) -> list[str]:
"""Returns list of measurements in measurements_dir (sorted)."""
return sorted(
[
os.path.basename(item.path)
for item in os.scandir(self.measurements_dir)
if item.is_dir()
],
reverse=sort_reverse,
)
[docs] def get_measurement(self, measurement_id: str) -> Measurement:
"""Returns Measurement object for measurement_id."""
return Measurement.from_id(measurement_id, self.measurements_dir)
[docs] def import_measurement(
self,
meta_obj: Meta,
source_file: str = None,
source_type: str = "ascii",
to_file: bool = True,
zip_source: bool = True,
remove_source: bool = True,
force: bool = False,
encoding=None,
) -> Measurement:
"""Import measurement from ASCII data file.
Parameter:
:meta_obj: Meta object for the measurement to be imported.
:source_file: Path to XRD data source file of the measurement.
:source_type: Type of data source, currently supported: [`ascii`].
:to_files: Write imported data and metadata to files (`csv`, `json`).
:zip_source: Add an archive with the source data to data subdirectory.
:remove_source: Delete the source data file if set to True.
:force: Replace the archive if it exists already in data subdirectory.
"""
# Read the XRD data from the source file
if encoding is None:
encoding = self.encoding
if source_file is None:
data = None
elif source_type == "ascii":
data = read_from_ascii(source_file, encoding=encoding)
logger.info(
f"Importing XRD data for {meta_obj.measurement_id!r} from: {source_file!r}"
)
else:
raise KeyError(f"Source type {source_type!r} unknown.")
paths = MeasurementPaths(self.measurements_dir, meta_obj.measurement_id)
measurement = Measurement(paths=paths, meta=meta_obj, data=data)
# Write the XRD data to csv file
if to_file and (source_file is not None):
print(f"Data file: {source_file}")
if not force:
utils.raise_file_exists(paths.file_data, force=True)
measurement._data_to_csv()
# Create a zip archive of the source file if zip_source is True
if zip_source and source_file is not None:
if not force:
utils.raise_file_exists(paths.file_source_data, force=True)
create_zip_archive(source_file, paths.file_source_data)
# Delete source file if remove_source is True
if remove_source and source_file is not None:
os.remove(source_file)
return measurement