Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions src/allotropy/parsers/lines_reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,14 @@ def pop_as_series(self, sep: str = " ") -> pd.Series[str] | None:
line = self.pop()
return None if line is None else pd.Series(line.split(sep))

def pop_line_as_df(
self, sep: str = " ", empty_pat: str = EMPTY_STR_PATTERN
) -> pd.DataFrame | None:
self.drop_empty(empty_pat)
line = self.pop()
self.drop_empty(empty_pat)
return None if line is None else pd.DataFrame([line.split(sep)])

def lines_as_df(
self,
lines: list[str],
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -709,7 +709,37 @@ def create(
)
set_columns(data, columns)

measurement_data = [
TimeMeasurementData.create(header, wavelength, row)
for _, row in data.iterrows()
]

return TimeWavelengthData(
wavelength=wavelength,
measurement_data=measurement_data,
)


@dataclass(frozen=True)
class SpectrumTimeWavelengthData:
wavelength: float
measurement_data: list[TimeMeasurementData]

@staticmethod
def create(
reader: CsvReader,
header: PlateHeader,
wavelength: float,
columns: pd.Series[str],
) -> SpectrumTimeWavelengthData:
data = assert_not_none(
reader.pop_line_as_df(sep="\t"),
msg="unable to find raw data from time block for Spectrum read type.",
)

set_columns(data, columns)

return SpectrumTimeWavelengthData(
wavelength=wavelength,
measurement_data=[
TimeMeasurementData.create(header, wavelength, row)
Expand Down Expand Up @@ -742,6 +772,29 @@ def create(reader: CsvReader, header: PlateHeader) -> TimeRawData:
)


@dataclass(frozen=True)
class TimeSpectrumRawData:
wavelength_data: list[SpectrumTimeWavelengthData]

@staticmethod
def create(reader: CsvReader, header: PlateHeader) -> TimeSpectrumRawData:
columns = assert_not_none(
reader.pop_as_series(sep="\t"),
msg="unable to find data columns for time block raw data.",
)
wavelength_data = [
SpectrumTimeWavelengthData.create(
reader,
header,
wavelength,
columns,
)
for wavelength in header.wavelengths
]

return TimeSpectrumRawData(wavelength_data=wavelength_data)


@dataclass(frozen=True)
class TimeReducedData:
data: list[ReducedDataElement]
Expand Down Expand Up @@ -773,20 +826,23 @@ def create(reader: CsvReader, header: PlateHeader) -> TimeReducedData:

@dataclass(frozen=True)
class TimeData:
raw_data: TimeRawData
raw_data: TimeRawData | TimeSpectrumRawData
reduced_data: TimeReducedData | None

@staticmethod
def create(
reader: CsvReader,
header: PlateHeader,
) -> TimeData:
raw_data = None
reduced_data = None
raw_data: TimeRawData | TimeSpectrumRawData | None = None

# Read raw data if data_type is RAW or BOTH
if header.data_type in (DataType.RAW.value, DataType.BOTH.value):
raw_data = TimeRawData.create(reader, header)
if header.read_type == ReadType.SPECTRUM.value:
raw_data = TimeSpectrumRawData.create(reader, header)
else:
raw_data = TimeRawData.create(reader, header)
# For REDUCED only, create synthetic raw data with error message
else:
raw_data = TimeData._create_synthetic_raw_data(header)
Expand All @@ -803,8 +859,8 @@ def create(
@staticmethod
def _create_synthetic_raw_data(header: PlateHeader) -> TimeRawData:
"""Create synthetic raw data with error messages when only reduced data is available."""
synthetic_wavelength_data = []

synthetic_wavelength_data = []
for wavelength in header.wavelengths:
# For each position in the plate, create a data element with an error document
num_cols = header.num_columns
Expand Down
Loading
Loading