Skip to content
Snippets Groups Projects

F awi sams

Merged Florian Spreckelsen requested to merge f-awi-sams into main
1 unresolved thread
2 files
+ 38
23
Compare changes
  • Side-by-side
  • Inline
Files
2
#
# Copyright (C) 2025 Indiscale GmbH <info@indiscale.com>
# Copyright (C) 2025 Florian Spreckelsen <f.spreckelsen@indiscale.com>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
import linkahead as db
import pandas as pd
from caosadvancedtools.datainconsistency import DataInconsistencyError
from .sample_upload_get_container import get_container
from .sample_upload_get_device import get_device
from .sample_upload_get_person import get_person
from .utils import get_column_header_name, get_entity_name, update_property, return_value_if_not_none
def add_special_properties(sample: db.Record, data: pd.Series) -> db.Record:
"""This function adds all the properties that can not be added
trivially from the given data row to the sample.
"""
main_user_prop = db.get_entity_by_name(get_entity_name("Main User"))
pi_prop = db.get_entity_by_name(get_entity_name("PI"))
container_rt = db.get_entity_by_name(get_entity_name("container_rt"))
device_rt = db.get_entity_by_name(get_entity_name("Device"))
parent_sample_prop = db.get_entity_by_name(get_entity_name("parent_sample_prop"))
if (get_column_header_name("PI") in data and
return_value_if_not_none(data[get_column_header_name("PI")]) is not None):
pi_user = get_person(return_value_if_not_none(data[get_column_header_name("PI")]))
sample = update_property(sample, pi_prop.id, pi_user, property_name=pi_prop.name)
if (get_column_header_name("Main User") in data and
return_value_if_not_none(data[get_column_header_name("Main User")]) is not None):
main_user = get_person(return_value_if_not_none(data[get_column_header_name("Main User")]))
sample = update_property(sample, main_user_prop.id, main_user, property_name=main_user_prop.name)
if (get_column_header_name("Storage ID") in data and
return_value_if_not_none(data[get_column_header_name("Storage ID")]) is not None):
container = get_container(return_value_if_not_none(
data[get_column_header_name("Storage ID")]))
sample = update_property(sample, container_rt.id, container, property_name=container_rt.name)
if (get_column_header_name("Device") in data and
return_value_if_not_none(data[get_column_header_name("Device")]) is not None):
device = get_device(return_value_if_not_none(
data[get_column_header_name("Device")]))
sample = update_property(sample, device_rt.id, device, property_name=device_rt.name)
if (get_column_header_name("Parent LinkAhead ID") in data and
return_value_if_not_none(data[get_column_header_name("Parent LinkAhead ID")]) is not None):
parent = return_value_if_not_none(
data[get_column_header_name("Parent LinkAhead ID")])
sample = update_property(sample, parent_sample_prop.id, parent, property_name=parent_sample_prop.name)
for name in ["Curator", "Collector"]:
prop = db.get_entity_by_name(get_entity_name(name))
if (get_column_header_name(name) in data and
return_value_if_not_none(data[get_column_header_name(name)]) is not None):
persons = [get_person(id_str) for id_str in return_value_if_not_none(
data[get_column_header_name(name)])]
sample = update_property(sample, prop.id, persons,
datatype=db.LIST(get_entity_name("Person")), property_name=prop.name)
return sample
Loading