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
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ authors = [
{ name = "Stephen Worlow", email = "stephen.worlow@benchling.com" },
{ name = "Brian Worth", email = "brian@benchling.com" },
{ name = "Alex Reis", email = "alex@manifold.bio" },
{ name = "Jane Dance", email = "jane.dance@abcellera.com" },
]
classifiers = [
"License :: OSI Approved :: MIT License",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -726,6 +726,7 @@ def create(
msg="unable to find raw data from time block.",
)
set_columns(data, columns)

return TimeWavelengthData(
wavelength=wavelength,
measurement_data=[
Expand Down Expand Up @@ -798,15 +799,72 @@ def create(
reader: CsvReader,
header: PlateHeader,
) -> TimeData:
raw_data = None
reduced_data = 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)
# For REDUCED only, create synthetic raw data with error message
else:
raw_data = TimeData._create_synthetic_raw_data(header)

# Read reduced data if available, regardless of data_type (RAW can have reduced data!)
if reader.current_line_exists():
reduced_data = TimeReducedData.create(reader, header)

return TimeData(
raw_data=TimeRawData.create(reader, header),
reduced_data=(
TimeReducedData.create(reader, header)
if reader.current_line_exists()
else None
),
raw_data=raw_data,
reduced_data=reduced_data,
)

@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 = []

for wavelength in header.wavelengths:
# For each position in the plate, create a data element with an error document
num_cols = header.num_columns
num_rows = header.num_rows

# Create synthetic data elements for all well positions
data_elements = {}
for row in range(1, num_rows + 1):
for col in range(1, num_cols + 1):
position = f"{num_to_chars(row-1)}{col}"
data_elements[position] = DataElement(
uuid=random_uuid_str(),
plate=header.name,
temperature=None,
wavelength=wavelength,
position=position,
value=NEGATIVE_ZERO,
error_document=[
ErrorDocument("Not reported", header.read_mode)
],
elapsed_time=[
0.0
], # Add a dummy value to ensure array is not empty
kinetic_measures=[
NEGATIVE_ZERO
], # Add a dummy value to ensure array is not empty
)

# Create a single TimeMeasurementData with all positions
measurement_data = [TimeMeasurementData(data_elements=data_elements)]

# Add this wavelength data to our list
synthetic_wavelength_data.append(
TimeWavelengthData(
wavelength=wavelength,
measurement_data=measurement_data,
)
)

# Create TimeRawData with our synthetic wavelength data
return TimeRawData(wavelength_data=synthetic_wavelength_data)

def iter_data_elements(self, position: str) -> Iterator[DataElement]:
for time_wavelength_data in self.raw_data.wavelength_data:
for measurement_data in time_wavelength_data.measurement_data:
Expand Down Expand Up @@ -892,8 +950,12 @@ def get_read_mode(cls, read_type: str, read_mode_raw: str) -> str:

@classmethod
def check_data_type(cls, data_type: str) -> None:
if data_type not in (DataType.RAW.value, DataType.BOTH.value):
msg = f"The SoftMax Pro file is required to include either 'Raw' or 'Both' (Raw and Reduced) data for all plates, got {data_type}."
if data_type not in (
DataType.RAW.value,
DataType.BOTH.value,
DataType.REDUCED.value,
):
msg = f"Unexpected data type: {data_type}, supported values are RAW, REDUCED, or BOTH."
raise AllotropeConversionError(msg)

@classmethod
Expand Down
Loading
Loading