Skip to content

Commit

Permalink
fix(shared-data): handle bad files in pipette defs (#15847)
Browse files Browse the repository at this point in the history
We are iterating through pipette definitions on disk but not really
handlign the possibility that you might have random files or dirs in
there. Now we do, and you can drop .DS_Store to your heart's content.

Closes RQA-2726
  • Loading branch information
sfoster1 authored Jul 31, 2024
1 parent 36a7501 commit ec7298a
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 13 deletions.
39 changes: 26 additions & 13 deletions shared-data/python/opentrons_shared_data/pipette/load_data.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
import json
import os
from pathlib import Path
from logging import getLogger

from typing import Dict, Any, Union, Optional, List
from typing import Dict, Any, Union, Optional, List, Iterator
from typing_extensions import Literal
from functools import lru_cache

Expand All @@ -26,6 +27,8 @@

LoadedConfiguration = Dict[str, Union[str, Dict[str, Any]]]

LOG = getLogger(__name__)


def _get_configuration_dictionary(
config_type: Literal["general", "geometry", "liquid"],
Expand Down Expand Up @@ -96,6 +99,12 @@ def _physical(
return _get_configuration_dictionary("general", channels, model, version)


def _dirs_in(path: Path) -> Iterator[Path]:
for child in path.iterdir():
if child.is_dir():
yield child


@lru_cache(maxsize=None)
def load_serial_lookup_table() -> Dict[str, str]:
"""Load a serial abbreviation lookup table mapped to model name."""
Expand All @@ -112,23 +121,27 @@ def load_serial_lookup_table() -> Dict[str, str]:
"eight_channel": "multi",
}
_model_shorthand = {"p1000": "p1k", "p300": "p3h"}
for channel_dir in os.listdir(config_path):
for model_dir in os.listdir(config_path / channel_dir):
for version_file in os.listdir(config_path / channel_dir / model_dir):
version_list = version_file.split(".json")[0].split("_")
built_model = f"{model_dir}_{_channel_model_str[channel_dir]}_v{version_list[0]}.{version_list[1]}"

model_shorthand = _model_shorthand.get(model_dir, model_dir)

for channel_dir in _dirs_in(config_path):
for model_dir in _dirs_in(channel_dir):
for version_file in model_dir.iterdir():
if version_file.suffix != ".json":
continue
try:
version_list = version_file.stem.split("_")
built_model = f"{model_dir.stem}_{_channel_model_str[channel_dir.stem]}_v{version_list[0]}.{version_list[1]}"
except IndexError:
LOG.warning(f"Pipette def with bad name {version_file} ignored")
continue
model_shorthand = _model_shorthand.get(model_dir.stem, model_dir.stem)
if (
model_dir == "p300"
model_dir.stem == "p300"
and int(version_list[0]) == 1
and int(version_list[1]) == 0
):
# Well apparently, we decided to switch the shorthand of the p300 depending
# on whether it's a "V1" model or not...so...here is the lovely workaround.
model_shorthand = model_dir
serial_shorthand = f"{model_shorthand.upper()}{_channel_shorthand[channel_dir]}V{version_list[0]}{version_list[1]}"
model_shorthand = model_dir.stem
serial_shorthand = f"{model_shorthand.upper()}{_channel_shorthand[channel_dir.stem]}V{version_list[0]}{version_list[1]}"
_lookup_table[serial_shorthand] = built_model
return _lookup_table

Expand Down
1 change: 1 addition & 0 deletions shared-data/python/tests/pipette/test_validate_schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ def test_pick_up_configs_configuration_by_nozzle_map_keys() -> None:
for channel_dir in os.listdir(paths_to_validate):
for model_dir in os.listdir(paths_to_validate / channel_dir):
for version_file in os.listdir(paths_to_validate / channel_dir / model_dir):
print(version_file)
version_list = version_file.split(".json")[0].split("_")
built_model: PipetteModel = PipetteModel(
f"{model_dir}_{_channel_model_str[channel_dir]}_v{version_list[0]}.{version_list[1]}"
Expand Down

0 comments on commit ec7298a

Please sign in to comment.