Skip to content

airtable

upsert_image #

upsert_image(at_api, ds_name, image_name, image_path, image_title, image_type, institution='HHMI / Janelia Research Campus', challenge=False, dry_run=False, image_path_s3=None)

Upsert a record to airtable image table.

An image can have a filesystem copy, an S3 copy, or both. Each copy that is present gets its own field: location for the filesystem copy, and location_s3 for the S3 copy. A copy that is absent leaves its field unset. The array metadata comes from the filesystem copy when it exists, because that copy reads faster.

Parameters:

Name Type Description Default
image_table Table

image airtable object to create references.

 required
ds_name str

name of the dataset.

 required
image_name str

name of the image to upsert.

 required
image_path str | None

filesystem path of the image. Pass None if the image has no filesystem copy. An s3:// path is not accepted here.

 required
image_title str

image title on openorganelle.com.

 required
image_type Literal['human_segmentation', 'em']

image type

 required
collection_table Table

collation airtable object to create references.

 required
fibsem_table Table

fibsem_imaging airtable object to create references.

 required
annotation_table Table

annotation airtable object to create references.

 required
dry_run bool

if True, compute the record that would be created/updated, but do not call Airtable's create/update. Defaults to False.

 False
image_path_s3 str | None

s3:// path of the image. Pass None if the image has no S3 copy. Defaults to None.

 None

Raises:

Type Description
ValueError

raise value error if image_path holds an s3:// path.
ValueError

raise value error if neither path holds a copy of the image.
ValueError

raise value error if multiple records with the same location and name are found in the image table.

Returns:

Name Type Description
dict dict

the record that was upserted, in the same shape returned by pyairtable ({'id': ..., 'fields': ...}). When dry_run is True, no record is actually created/updated, so 'id' is None.
Source code in src/cellmap_utils/airtable/upsert/image.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def upsert_image(
    at_api: api,
    ds_name: str,
    image_name: str,
    image_path: str | None,
    image_title: str,
    image_type: Literal["human_segmentation", "em"],
    institution: str = "HHMI / Janelia Research Campus",
    challenge : bool = False,
    dry_run : bool = False,
    image_path_s3: str | None = None,
) -> dict:
    """Upsert a record to airtable image table.

    An image can have a filesystem copy, an S3 copy, or both. Each copy that is
    present gets its own field: ``location`` for the filesystem copy, and
    ``location_s3`` for the S3 copy. A copy that is absent leaves its field
    unset. The array metadata comes from the filesystem copy when it exists,
    because that copy reads faster.

    Args:
        image_table (api.table.Table): image airtable object to create references.
        ds_name (str): name of the dataset.
        image_name (str): name of the image to upsert.
        image_path (str | None): filesystem path of the image. Pass None if the
            image has no filesystem copy. An ``s3://`` path is not accepted here.
        image_title (str): image title on openorganelle.com.
        image_type (Literal['human_segmentation', 'em']): image type
        collection_table (api.table.Table): collation airtable object to create references.
        fibsem_table (api.table.Table): fibsem_imaging airtable object to create references.
        annotation_table (api.table.Table): annotation airtable object to create references.
        dry_run (bool, optional): if True, compute the record that would be
            created/updated, but do not call Airtable's create/update. Defaults
            to False.
        image_path_s3 (str | None, optional): ``s3://`` path of the image. Pass
            None if the image has no S3 copy. Defaults to None.

    Raises:
        ValueError: raise value error if image_path holds an s3:// path.
        ValueError: raise value error if neither path holds a copy of the image.
        ValueError: raise value error if multiple records with the same location and name are found in the image table.

    Returns:
        dict: the record that was upserted, in the same shape returned by
            pyairtable (``{'id': ..., 'fields': ...}``). When dry_run is True, no
            record is actually created/updated, so 'id' is None.
    """

    if image_path is not None and _is_s3(image_path):
        raise ValueError(
            "image_path must be a filesystem path. "
            "Pass an s3:// path in image_path_s3."
        )

    if image_path is None and image_path_s3 is None:
        raise ValueError("Pass image_path, image_path_s3, or both.")

    locations = {}
    for field, path in (("location", image_path), ("location_s3", image_path_s3)):
        if path is None:
            continue
        path = path.rstrip("/")
        if _path_exists(path):
            locations[field] = path
        else:
            print(f"No copy found at {path}")

    if not locations:
        raise ValueError("No copy found at the given paths.")

    metadata_path = locations.get("location", locations.get("location_s3"))

    image_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["IMAGE_TABLE_ID"]
    )
    collection_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["COLLECTION_TABLE_ID"]
    )
    fibsem_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["FIBSEM_TABLE_ID"]
    )
    annotation_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["ANNOTATION_TABLE_ID"]
    )
    institution_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["INSTITUTION_TABLE_ID"]
    )

    # a record already in airtable can carry only one of the two locations
    existing_records = []
    seen_ids = set()
    for field, path in locations.items():
        for record in image_table.all(formula=match({"name": image_name, field: path})):
            if record["id"] not in seen_ids:
                seen_ids.add(record["id"])
                existing_records.append(record)

    if image_type in ["human_segmentation", "ml_segmentation"]:
        value_type = "label"
    else:
        value_type = "scalar"

    zg, z_arr_name = _read_multiscale_group(metadata_path)
    scale, offset = get_s0_level(zg)
    shape = zg[z_arr_name].shape

    try:
        fibsem_imaging = [fibsem_table.all(formula=match({"name": ds_name}))[0]["id"]]
    except:
        fibsem_imaging = []

    try:
        annotation = [
            annotation_table.all(formula=match({"name": image_name}))[0]["id"]
        ]
    except:
        annotation = []

    record_to_upsert = {
        "name": image_name,
        "collection": [collection_table.all(formula=match({"id": ds_name}))[0]["id"]],
        **locations,
        "format": "zarr",
        "title": image_title,
        "institution": [
            institution_table.all(formula=match({"name": institution}))[0]["id"]
        ],
        "image_type": image_type,
        "value_type": value_type,
        "size_x_pix": shape[2],
        "size_y_pix": shape[1],
        "size_z_pix": shape[0],
        "resolution_x_nm": scale[2],
        "resolution_y_nm": scale[1],
        "resolution_z_nm": scale[0],
        "offset_x_nm": offset[2],
        "offset_y_nm": offset[1],
        "offset_z_nm": offset[0],
        "fibsem_imaging": fibsem_imaging,
        "challenge" : challenge,
        "annotation": annotation,
    }

    if len(existing_records) > 2:
        raise ValueError("Multiple records with matching input image name found")

    if dry_run:
        action = "update" if existing_records else "create"
        print(f"[dry_run] would {action} image record: {record_to_upsert}")
        return {"id": None, "fields": record_to_upsert}

    if not existing_records:
        return image_table.create(record_to_upsert)
    elif len(existing_records) == 1:
        return image_table.update(existing_records[0]["id"], record_to_upsert)

upsert_doi #

upsert_doi(at_api, image_path, doi_name, dry_run=False)

Upsert a record to airtable doi table, linked to an image record.

Segmentation images (image_type "human_segmentation") do not belong in the doi table. They belong in the doi_crop table instead, so this method rejects them. Images with image_type "ml_segmentation" are still accepted here.

The doi record's dataset field is set to "{collection id} {image title}", matching the naming convention already used for other doi records. The image's linked collection record is also set on the doi record's collection field, which drives several of the doi table's lookup fields. The collection's linked sample record, if any, is set on the doi record's sample field. When image_type is "em" (raw FIB-SEM data), the image's linked fibsem_imaging record is also carried forward onto the doi record's fibsem_imaging field. Derived images (predictions, segmentations) do not get this field set.

Parameters:

Name Type Description Default
at_api api

pyairtable Api object used to reach the base.

 required
image_path str

filesystem path or s3:// path of the image record to link to the doi record. The image record's location or location_s3 field must hold this exact path. Plain image names are not accepted here, because images in different datasets often share the same name.

 required
doi_name str

value for the doi record's doi_name field.

 required
dry_run bool

if True, compute the record that would be created/updated, but do not call Airtable's create/update. Defaults to False.

 False

Raises:

Type Description
ValueError

raise value error if no image record matches image_path.
ValueError

raise value error if multiple image records match image_path.
ValueError

raise value error if the image's image_type is "human_segmentation".
ValueError

raise value error if multiple doi records already link to the image.
ValueError

raise value error if the image has no linked collection record.

Returns:

Name Type Description
dict dict

the record that was upserted, in the same shape returned by pyairtable ({'id': ..., 'fields': ...}). When dry_run is True, no record is actually created/updated, so 'id' is None.
Source code in src/cellmap_utils/airtable/upsert/doi.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def upsert_doi(
    at_api: api,
    image_path: str,
    doi_name: str,
    dry_run: bool = False,
) -> dict:
    """Upsert a record to airtable doi table, linked to an image record.

    Segmentation images (image_type "human_segmentation") do not belong in the
    doi table. They belong in the doi_crop table instead, so this method
    rejects them. Images with image_type "ml_segmentation" are still accepted
    here.

    The doi record's ``dataset`` field is set to ``"{collection id} {image
    title}"``, matching the naming convention already used for other doi
    records. The image's linked
    collection record is also set on the doi record's ``collection`` field,
    which drives several of the doi table's lookup fields. The collection's
    linked sample record, if any, is set on the doi record's ``sample`` field.
    When image_type is "em" (raw FIB-SEM data), the image's linked
    fibsem_imaging record is also carried forward onto the doi record's
    ``fibsem_imaging`` field. Derived images (predictions, segmentations) do
    not get this field set.

    Args:
        at_api (api): pyairtable Api object used to reach the base.
        image_path (str): filesystem path or s3:// path of the image record to
            link to the doi record. The image record's ``location`` or
            ``location_s3`` field must hold this exact path. Plain image
            names are not accepted here, because images in different
            datasets often share the same name.
        doi_name (str): value for the doi record's doi_name field.
        dry_run (bool, optional): if True, compute the record that would be
            created/updated, but do not call Airtable's create/update. Defaults
            to False.

    Raises:
        ValueError: raise value error if no image record matches image_path.
        ValueError: raise value error if multiple image records match image_path.
        ValueError: raise value error if the image's image_type is
            "human_segmentation".
        ValueError: raise value error if multiple doi records already link to
            the image.
        ValueError: raise value error if the image has no linked collection
            record.

    Returns:
        dict: the record that was upserted, in the same shape returned by
            pyairtable (``{'id': ..., 'fields': ...}``). When dry_run is True, no
            record is actually created/updated, so 'id' is None.
    """

    image_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["IMAGE_TABLE_ID"]
    )
    doi_table = at_api.table(os.environ["AIRTABLE_BASE_ID"], os.environ["DOI_TABLE_ID"])
    collection_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["COLLECTION_TABLE_ID"]
    )

    image_path = image_path.rstrip("/")
    image_records = image_table.all(
        formula=match({"location": image_path})
    ) or image_table.all(formula=match({"location_s3": image_path}))
    if not image_records:
        raise ValueError(f"No image record found with location {image_path!r}")
    if len(image_records) > 1:
        raise ValueError(f"Multiple image records found with location {image_path!r}")
    image_record = image_records[0]
    image_name = image_record["fields"]["name"]

    image_type = image_record["fields"].get("image_type")
    if image_type == "human_segmentation":
        raise ValueError(
            "image_type 'human_segmentation' belongs in the doi_crop table, "
            "not doi. Use a doi_crop upsert method instead."
        )

    # narrow by the image table's primary field value first (cheap), then
    # confirm by exact linked record id, because a link field compares
    # against the linked record's primary field rather than its name
    image_primary_value = image_record["fields"].get("collection/name", image_name)
    candidate_records = doi_table.all(formula=match({"image": image_primary_value}))
    existing_records = [
        record
        for record in candidate_records
        if image_record["id"] in record["fields"].get("image", [])
    ]
    if len(existing_records) > 1:
        raise ValueError(
            f"Multiple doi records already link to image {image_path!r}"
        )

    collection_ids = image_record["fields"].get("collection", [])
    if not collection_ids:
        raise ValueError(f"Image {image_path!r} has no linked collection record")
    collection_record = collection_table.get(collection_ids[0])
    ds_name = collection_record["fields"]["id"]
    title = image_record["fields"].get("title", "")
    dataset = f"{ds_name} {title}".strip()

    # a collection links to a single sample, but a sample can have several
    # collections, so this is a 1-to-many relationship from sample to collection
    sample_ids = collection_record["fields"].get("sample", [])
    if not sample_ids:
        print(f"No sample linked to collection {ds_name!r}")

    # only raw FIB-SEM data (image_type "em") carries its own fibsem_imaging
    # link forward onto the doi record; derived images (predictions,
    # segmentations) do not.
    fibsem_imaging_ids = []
    if image_type == "em":
        fibsem_imaging_ids = image_record["fields"].get("fibsem_imaging", [])
        if not fibsem_imaging_ids:
            print(f"No fibsem_imaging linked to image {image_path!r}")

    record_to_upsert = {
        "image": [image_record["id"]],
        "doi_name": doi_name,
        "dataset": dataset,
        "collection": [collection_ids[0]],
        **({"sample": sample_ids} if sample_ids else {}),
        **({"fibsem_imaging": fibsem_imaging_ids} if fibsem_imaging_ids else {}),
    }

    if dry_run:
        action = "update" if existing_records else "create"
        print(f"[dry_run] would {action} doi record: {record_to_upsert}")
        return {"id": None, "fields": record_to_upsert}

    if not existing_records:
        return doi_table.create(record_to_upsert)
    else:
        return doi_table.update(existing_records[0]["id"], record_to_upsert)

upsert_record_scene_tables #

upsert_record_scene_tables(scene_table, scene_to_image_table, image_table, image_location, scene_data={}, scene_to_image_data={}, dry_run=False)

This method upserts records into scene and scene_to_image table at the same time.

Parameters:

Name Type Description Default
scene_table Table

output table to upsert

 required
scene_to_image_table Table

output table to upsert

 required
image_table Table

input table from where the record is being taken

 required
image_location str

provides value for location filter paramater for the image record

 required
scene_update dict

custom data to upsert into "scene" table

 required
scene_to_image_update dict

custom data to upsert in "scene_to_image" table

 required
dry_run bool

if True, compute the records that would be created/updated, but do not call Airtable's create/update. Since the scene record links to the scene_to_image record's id, that link is None in dry_run mode (no real record was created to link to). Defaults to False.

 False

Returns:

Type Description
Tuple[dict]

Tuple[dict]: output records that were upserted, in the same shape returned by pyairtable ({'id': ..., 'fields': ...}). When dry_run is True, no records are actually created/updated, so 'id' is None in both.
Source code in src/cellmap_utils/airtable/upsert/scene.py 
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def upsert_record_scene_tables(
    scene_table: api.table.Table,
    scene_to_image_table: api.table.Table,
    image_table: api.table.Table,
    image_location: str,
    scene_data: dict = {},
    scene_to_image_data: dict = {},
    dry_run: bool = False,
) -> Tuple[dict]:
    """This method upserts records into scene and scene_to_image table at the same time.

    Args:
        scene_table (pyairtable.api.table.Table): output table to upsert
        scene_to_image_table (pyairtable.api.table.Table): output table to upsert
        image_table (pyairtable.api.table.Table): input table from where the record is being taken
        image_location (str): provides value for location filter paramater for the image record
        scene_update (dict): custom data to upsert into "scene" table
        scene_to_image_update (dict): custom data to upsert in "scene_to_image" table
        dry_run (bool, optional): if True, compute the records that would be
            created/updated, but do not call Airtable's create/update. Since the
            scene record links to the scene_to_image record's id, that link is
            None in dry_run mode (no real record was created to link to).
            Defaults to False.

    Returns:
        Tuple[dict]: output records that were upserted, in the same shape
            returned by pyairtable (``{'id': ..., 'fields': ...}``). When
            dry_run is True, no records are actually created/updated, so 'id'
            is None in both.
    """

    # fetch a record from image table:
    image_records = image_table.all(formula=match({"location": image_location}))
    if len(image_records) > 1:
        raise ValueError(
            "Multiple images with the same path are found in airtable image table. Only one record should exist!"
        )
    else:
        image_airt = image_records[0]

    scene_to_image_ids = {}
    scene_ids = {}

    scene_to_image_insert = {
        "image": [image_airt["id"]],
        "contrast_start": 0,
        "contrast_stop": 255,
        "color": "white",
    }
    # update if input data for an upsert record exists
    scene_to_image_insert = {
        key: scene_to_image_data.get(key, val)
        for key, val in scene_to_image_insert.items()
    }

    location = image_airt["fields"]["location"]
    print(location)

    # define whether update or create a record in the scene_to_image table by looking at the value from 'location_web' column
    existing_record_scene_to_image = scene_to_image_table.first(
        formula=match({"location_web": location})
    )

    # add records to scene_to_image table
    if dry_run:
        action = "update" if existing_record_scene_to_image else "create"
        print(f"[dry_run] would {action} scene_to_image record: {scene_to_image_insert}")
        scene_to_image_ids[location] = {"id": None, "fields": scene_to_image_insert}
    elif existing_record_scene_to_image == None:
        scene_to_image_ids[location] = scene_to_image_table.create(
            scene_to_image_insert
        )
    else:
        scene_to_image_ids[location] = scene_to_image_table.update(
            existing_record_scene_to_image["id"], scene_to_image_insert
        )

    # add record to scene table
    # define whether update or create a record in the scene table by looking at the value from 'scene_to_image' column
    if existing_record_scene_to_image:
        existing_record_scene = scene_table.first(
            formula=match(
                {"scene_to_image": existing_record_scene_to_image["fields"]["name"]}
            )
        )
    else:
        existing_record_scene = None

    scene_insert = {
        "name": "Default view",
        "scene_to_image": [scene_to_image_ids[location]["id"]],
        "description": "The default view of the data.",
        "crossection_scale": 10.0,
        "projection_scale": 1000.0,
    }
    # update if input data for an upsert record exists
    scene_insert = {key: scene_data.get(key, val) for key, val in scene_insert.items()}

    if dry_run:
        action = "update" if existing_record_scene else "create"
        print(f"[dry_run] would {action} scene record: {scene_insert}")
        scene_ids[location] = {"id": None, "fields": scene_insert}
    elif existing_record_scene == None:
        scene_ids[location] = scene_table.create(scene_insert)
    else:
        scene_ids[location] = scene_table.update(
            existing_record_scene["id"], scene_insert
        )

    return (scene_to_image_ids, scene_ids)

get_image_record #

get_image_record(image_path, ds_name, at_api)

Read an image record from airtable and return supabase image, dataset, sample, and acquisition records.

Parameters:

Name Type Description Default
image_path str

path to dataset used as a filter parameter for airtable image table records

 required
at_api api

airtable api instance

 required

Returns:

Name Type Description
dict

returns a dictionary of records, with pydantic models for supabase table records as values.
Source code in src/cellmap_utils/airtable/supabase/air_to_supabase.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
def get_image_record(image_path: str, ds_name: str, at_api: api):
    """Read an image record from airtable and return supabase image, dataset, sample, and acquisition records.

    Args:
        image_path (str): path to dataset used as a filter parameter for airtable image table records
        at_api (api): airtable api instance

    Returns:
        dict: returns a dictionary of records, with pydantic models for supabase table records as values.
    """

    # image table (switch between prod and test bases in .env)
    image_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["IMAGE_TABLE_ID"]
    )
    fibsem_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["FIBSEM_TABLE_ID"]
    )
    institution_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["INSTITUTION_TABLE_ID"]
    )

    image_record = image_table.all(formula=match({"location_s3": image_path.rstrip("/")}))[0]

    name = image_record["fields"]["name"]
    display_settings = {
        "invertLUT": False,
        "contrastLimits": {"end": 255, "max": 255, "min": 0, "start": 0},
    }

    content_type_mapping = {
        "em": "em",
        "ml_segmentation": "segmentation",
        "human_segmentation": "segmentation",
    }
    content_type = content_type_mapping[image_record["fields"]["image_type"]]

    source = None
    try:

        fibsem_record = fibsem_table.get(image_record["fields"]["fibsem_imaging"][0])
        if image_record["fields"]["image_type"] == "em":
            display_settings["color"] = "white"

            format_string = "%Y-%m-%d"
            source = {
                "bias_V": fibsem_record["fields"]["bias_v"],
                "scan_hz": fibsem_record["fields"]["scan_rate_mhz"],
                "current_nA": fibsem_record["fields"]["imaging_current_nA"],
                "duration_days": (
                    datetime.strptime(
                        fibsem_record["fields"]["stop_date"], format_string
                    ).date()
                    - datetime.strptime(
                        fibsem_record["fields"]["start_date"], format_string
                    ).date()
                ).days,
                "landing_energy_eV": fibsem_record["fields"]["primary_energy_ev"],
            }
    except:
        print("NO FIBSEM RECORD", ds_name)

    if image_record["fields"]["image_type"] == "em":
        pub_name = "Reconstructed FIB-SEM data"
        image_stack = f"{ds_name}_{name}"
    elif image_record["fields"]["image_type"] == "human_segmentation":
        pub_name = "Segmentations"
        image_stack = f"{ds_name}_groundtruth"
    else:
        pub_name = "Segmentations"
        image_stack = f"{ds_name}_{name}"

    if image_record["fields"]["image_type"] == "human_segmentation":
        doi_field = "doi_link_dataset_crop"
    else:
        doi_field = "doi_link_dataset"

    print(image_record)
    supa_image = SupaImageModel(
        source=source,
        name=name,
        url=image_path,
        description=image_record["fields"]["title"],
        format="zarr",
        display_settings=display_settings,
        sample_type=image_record["fields"]["value_type"],
        content_type=content_type,
        dataset_name=ds_name,
        institution=institution_table.get( image_record['fields']['institution'][0])['fields']['name'],
        grid_dims=["z", "y", "x"],
        grid_scale=[
            image_record["fields"][f"resolution_{_}_nm"] for _ in ["z", "y", "x"]
        ],
        grid_translation=[0.0, 0.0, 0.0],
        # [
        #     image_record["fields"][f"offset_{_}_nm"] for _ in ["z", "y", "x"]
        # ],
        grid_units=["nm", "nm", "nm"],
        grid_index_order="C",
        stage="dev",
        image_stack=image_stack,
        doi={"url": image_record["fields"][doi_field][0], "name": pub_name},
    )

    return supa_image

get_mesh_record #

get_mesh_record(mesh_path, image_path, ds_name, at_api)

Read a neuroglancer mesh and its airtable image record, and return a supabase mesh record.

The grid scale and the grid shift come from the transform in the mesh info file, not from a fixed value. A mesh that carries a transform other than the identity prints a warning, because such a mesh does not sit on the image.

Parameters:

Name Type Description Default
mesh_path str

s3 path of the neuroglancer precomputed mesh volume.

 required
image_path str

s3 path of the image the mesh comes from. It must match the location_s3 field of an airtable image record.

 required
ds_name str

name of the dataset.

 required
at_api api

airtable api instance.

 required

Returns:

Name Type Description
SupaMeshModel

the supabase mesh record.
Source code in src/cellmap_utils/airtable/supabase/air_to_supabase.py 
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def get_mesh_record(mesh_path: str, image_path: str, ds_name: str, at_api: api):
    """Read a neuroglancer mesh and its airtable image record, and return a supabase mesh record.

    The grid scale and the grid shift come from the transform in the mesh info
    file, not from a fixed value. A mesh that carries a transform other than the
    identity prints a warning, because such a mesh does not sit on the image.

    Args:
        mesh_path (str): s3 path of the neuroglancer precomputed mesh volume.
        image_path (str): s3 path of the image the mesh comes from. It must match
            the location_s3 field of an airtable image record.
        ds_name (str): name of the dataset.
        at_api (api): airtable api instance.

    Returns:
        SupaMeshModel: the supabase mesh record.
    """

    image_table = at_api.table(
        os.environ["AIRTABLE_BASE_ID"], os.environ["IMAGE_TABLE_ID"]
    )
    image_record = image_table.all(
        formula=match({"location_s3": image_path.rstrip("/")})
    )[0]

    info = _read_mesh_info(mesh_path)
    grid_scale, grid_translation = _grid_from_transform(info["transform"])

    if info["transform"] != _IDENTITY_TRANSFORM:
        print(f"MESH TRANSFORM IS NOT THE IDENTITY, {mesh_path}, {info['transform']}")

    supa_mesh = SupaMeshModel(
        name=_mesh_name(mesh_path),
        description=image_record["fields"]["title"],
        url=mesh_path.rstrip("/"),
        format=info["@type"],
        ids=[],
        source=None,
        grid_dims=["z", "y", "x"],
        grid_scale=grid_scale,
        grid_translation=grid_translation,
        grid_units=["nm", "nm", "nm"],
        grid_index_order="C",
        stage="dev",
        dataset_name=ds_name,
        image_name=image_record["fields"]["name"],
    )

    return supa_mesh

filter_records #

filter_records(records, filter_field, filter_value)

Custom method to filter records by a field value. all(formula=match({})) doesn't work when filtering by reference.

Parameters:

Name Type Description Default
records (dict)

list of all records in a table. Not using table.all() within a method, since it slow to fetch all the records every time when a method is called.

 required
field str

table field name

 required
field_value list[str] | str

table field value

 required

Returns:

Type Description

list[dict]: return records with filter_value for filter_field field
Source code in src/cellmap_utils/airtable/filter/filter.py 
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def filter_records(records, filter_field: str, filter_value: list[str] | str):
    """Custom method to filter records by a field value. all(formula=match({})) doesn't work when filtering by reference.

    Args:
        records (dict) : list of all records in a table. Not using table.all() within a method, since it slow to fetch all the records every time when a method is called.
        field (str): table field name
        field_value (list[str] | str): table field value

    Returns:
        list[dict]: return records with filter_value for filter_field field
    """

    matches = []
    for record in records:
        try:
            if isinstance(record["fields"][filter_field], list):
                cond = filter_value in record["fields"][filter_field]
            else:
                cond = record["fields"][filter_field] == filter_value
            if cond:
                matches.append(record)
        except:
            pass
    return matches