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
Original file line number Diff line number Diff line change
Expand Up @@ -249,7 +249,10 @@ def _create_measurements(plate_block: PlateBlock, position: str) -> list[Measure


def _create_measurement_group(
plate_block: PlateBlock, position: str, date_last_saved: str | None
plate_block: PlateBlock,
position: str,
date_last_saved: str | None,
structure_data: StructureData,
) -> MeasurementGroup | None:

if not (measurements := _create_measurements(plate_block, position)):
Expand All @@ -260,10 +263,41 @@ def _create_measurement_group(
delta = datetime.timedelta(seconds=plate_block.header.read_time or 0)
measurement_time = (parser.parse(date_last_saved) + delta).isoformat()

# Gather unread group fields for this plate/position
unread_rows: list[dict[str, float | str | None]] = []
single_plate_name: str | None = None
if len(structure_data.block_list.plate_blocks) == 1:
single_plate_name = next(iter(structure_data.block_list.plate_blocks.keys()))

for group_block in structure_data.block_list.group_blocks:
for group_sample_data in group_block.group_data.sample_data:
for (
plate_key,
positions_map,
) in group_sample_data.unread_by_plate_and_position.items():
effective_plate = (
plate_key if plate_key is not None else single_plate_name
)
# filters out unread entries that do not belong to the current plate
if effective_plate != plate_block.header.name:
continue
if position in positions_map:
unread_rows.extend(positions_map[position])

# Flatten unread rows into a single custom_info dict
custom_info = None
if unread_rows:
merged = {}
for row_dict in unread_rows:
# Only include keys with non-None values
merged.update({k: v for k, v in row_dict.items() if v is not None})
custom_info = merged

return MeasurementGroup(
measurements=measurements,
plate_well_count=plate_block.header.num_wells,
measurement_time=measurement_time,
custom_info=custom_info,
)


Expand All @@ -274,7 +308,7 @@ def create_measurement_groups(data: StructureData) -> list[MeasurementGroup]:
for position in plate_block.iter_wells()
if (
measurement_group := _create_measurement_group(
plate_block, position, data.date_last_saved
plate_block, position, data.date_last_saved, data
)
)
]
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -128,6 +128,10 @@ class GroupDataElement:
class GroupSampleData:
identifier: str
data_elements: list[GroupDataElement]
# Capture unread fields per plate and well position. The None key is used when plate is unspecified.
unread_by_plate_and_position: dict[
str | None, dict[str, list[dict[str, float | str | None]]]
]

@classmethod
def create(cls, data: pd.DataFrame, calc_data_cols: list[str]) -> GroupSampleData:
Expand All @@ -141,6 +145,9 @@ def create(cls, data: pd.DataFrame, calc_data_cols: list[str]) -> GroupSampleDat
data_elements: dict[str, list[GroupDataElement]] = {
column: [] for column in data_columns
}
unread_by_plate_and_position: dict[
str | None, dict[str, list[dict[str, float | str | None]]]
] = {}
for row in row_data:
row_results = cls._get_row_results(row, data_columns)
position = row[str, ["Well", "Wells"]]
Expand Down Expand Up @@ -170,12 +177,17 @@ def create(cls, data: pd.DataFrame, calc_data_cols: list[str]) -> GroupSampleDat
)
elif data_elements[column]:
data_elements[column][-1].positions.append(position)
unread = row.get_unread(skip={"Sample"})
if unread:
by_position = unread_by_plate_and_position.setdefault(plate, {})
by_position.setdefault(str(position), []).append(unread)

return GroupSampleData(
identifier=identifier,
data_elements=[
elem for elements in data_elements.values() for elem in elements
],
unread_by_plate_and_position=unread_by_plate_and_position,
)

@classmethod
Expand Down
Loading