|
3 | 3 | from abc import ABC |
4 | 4 | from collections.abc import Iterator |
5 | 5 | from dataclasses import dataclass, field |
6 | | -from datetime import datetime |
7 | 6 | from enum import Enum |
8 | 7 | import math |
9 | 8 | import re |
@@ -65,14 +64,25 @@ def num_wells_to_n_columns(well_count: int) -> int: |
65 | 64 |
|
66 | 65 |
|
67 | 66 | def time_to_seconds(time: str) -> float: |
68 | | - """Transforms HH:MM:SS formatted time into seconds""" |
69 | | - try: |
70 | | - pt = datetime.strptime(time, "%H:%M:%S").astimezone() |
71 | | - except ValueError as e: |
72 | | - msg = "Bad time formatting, expected HH:MM:SS, got {time}" |
73 | | - raise AllotropeConversionError(msg) from e |
74 | | - |
75 | | - return pt.hour * 3600 + pt.minute * 60 + pt.second |
| 67 | + """Transforms time string into seconds. Supports H:MM:SS, HH:MM:SS, and M:SS formats.""" |
| 68 | + parts = time.split(":") |
| 69 | + if len(parts) == 3: |
| 70 | + try: |
| 71 | + h, m, s = int(parts[0]), int(parts[1]), int(parts[2]) |
| 72 | + return h * 3600 + m * 60 + s |
| 73 | + except ValueError as e: |
| 74 | + msg = f"Bad time formatting, expected HH:MM:SS, got {time}" |
| 75 | + raise AllotropeConversionError(msg) from e |
| 76 | + elif len(parts) == 2: |
| 77 | + try: |
| 78 | + m, s = int(parts[0]), int(parts[1]) |
| 79 | + return m * 60 + s |
| 80 | + except ValueError as e: |
| 81 | + msg = f"Bad time formatting, expected MM:SS, got {time}" |
| 82 | + raise AllotropeConversionError(msg) from e |
| 83 | + else: |
| 84 | + msg = f"Bad time formatting, expected HH:MM:SS or MM:SS, got {time}" |
| 85 | + raise AllotropeConversionError(msg) |
76 | 86 |
|
77 | 87 |
|
78 | 88 | class ReadType(Enum): |
@@ -256,10 +266,8 @@ def create(reader: CsvReader) -> GroupData: |
256 | 266 | # We are doing 1 for now, but we should check |
257 | 267 | data = data.dropna(axis=1, how="all") |
258 | 268 |
|
259 | | - assert_not_none( |
260 | | - data.get("Sample"), |
261 | | - msg=f"Unable to find sample identifier column in group data {name}", |
262 | | - ) |
| 269 | + if data.empty or data.get("Sample") is None: |
| 270 | + return GroupData(name=name, sample_data=[]) |
263 | 271 |
|
264 | 272 | calc_data_cols = GroupData.get_calculated_data_columns(data) |
265 | 273 |
|
@@ -296,18 +304,10 @@ class GroupColumns: |
296 | 304 |
|
297 | 305 | @staticmethod |
298 | 306 | def create(reader: CsvReader) -> GroupColumns: |
299 | | - data = assert_not_none( |
300 | | - reader.pop_csv_block_as_df(sep="\t", header=0), |
301 | | - msg="Unable to find group block columns.", |
302 | | - ) |
303 | | - |
304 | | - if "Formula Name" not in data: |
305 | | - msg = "Unable to find 'Formula Name' in group block columns." |
306 | | - raise AllotropeConversionError(msg) |
| 307 | + data = reader.pop_csv_block_as_df(sep="\t", header=0) |
307 | 308 |
|
308 | | - if "Formula" not in data: |
309 | | - msg = "Unable to find 'Formula' in group block columns." |
310 | | - raise AllotropeConversionError(msg) |
| 309 | + if data is None or "Formula Name" not in data or "Formula" not in data: |
| 310 | + return GroupColumns(data={}) |
311 | 311 |
|
312 | 312 | return GroupColumns( |
313 | 313 | data=dict(zip(data["Formula Name"], data["Formula"], strict=True)), |
@@ -430,8 +430,7 @@ def create( |
430 | 430 | for position, raw_value in data.items(): |
431 | 431 | value = try_non_nan_float_or_none(raw_value) |
432 | 432 | if value is None and elapsed_time is not None: |
433 | | - msg = f"Missing kinetic measurement for well position {position} at {elapsed_time}s." |
434 | | - raise AllotropeConversionError(msg) |
| 433 | + continue |
435 | 434 | data_elements[str(position)] = DataElement( |
436 | 435 | uuid=random_uuid_str(), |
437 | 436 | plate=header.name, |
@@ -459,6 +458,8 @@ def update_kinetic_data_elements( |
459 | 458 | for col, raw_value in zip(df_data.columns, row_data, strict=True) |
460 | 459 | } |
461 | 460 | for position, value in data.items(): |
| 461 | + if position not in self.data_elements: |
| 462 | + continue |
462 | 463 | if value is None: |
463 | 464 | msg = f"Missing kinetic measurement for well position {position} at {elapsed_time}s." |
464 | 465 | raise AllotropeConversionError(msg) |
|
0 commit comments