| 20 | |
| 21 | |
| 22 | class Upload(): |
| 23 | |
| 24 | def __init__(self, |
| 25 | session, |
| 26 | project, |
| 27 | request, |
| 28 | member): |
| 29 | |
| 30 | self.session = session |
| 31 | self.project = project |
| 32 | self.request = request |
| 33 | self.member = member |
| 34 | self.log = regular_log.default() |
| 35 | |
| 36 | def get_project_labels(self): |
| 37 | result = [] |
| 38 | directory = self.project.directory_default |
| 39 | |
| 40 | working_dir_sub_query = self.session.query(WorkingDirFileLink).filter( |
| 41 | WorkingDirFileLink.working_dir_id == directory.id, |
| 42 | WorkingDirFileLink.type == "label").subquery('working_dir_sub_query') |
| 43 | |
| 44 | working_dir_file_list = self.session.query(File).filter( |
| 45 | File.id == working_dir_sub_query.c.file_id).all() |
| 46 | |
| 47 | for file in working_dir_file_list: |
| 48 | if file.state != "removed": |
| 49 | result.append(file.serialize_with_label_and_colour(session = self.session)) |
| 50 | return result |
| 51 | |
| 52 | def extract_metadata_from_batch(self, input, input_batch_id, file_name): |
| 53 | input_batch = InputBatch.get_by_id(self.session, id = input_batch_id) |
| 54 | pre_labels = input_batch.pre_labeled_data |
| 55 | if pre_labels is None: |
| 56 | return input |
| 57 | uuid = None |
| 58 | file_data = None |
| 59 | if self.request: |
| 60 | uuid = self.request.form.get('uuid') |
| 61 | file_data = pre_labels.get(uuid) |
| 62 | if file_data is None: |
| 63 | # Try finding the pre_labels with the file_name as a backup |
| 64 | file_data = pre_labels.get(file_name) |
| 65 | if file_data is None: |
| 66 | logger.warning(f"Input: {input.id} File {file_name} has no pre_labeled data associated") |
| 67 | return |
| 68 | |
| 69 | if file_data.get('file_metadata'): |
| 70 | file_metadata = file_data.get('file_metadata') |
| 71 | input.file_metadata = file_metadata |
| 72 | |
| 73 | return input |
| 74 | |
| 75 | def extract_instance_list_from_batch(self, input, input_batch_id, file_name): |
| 76 | if input_batch_id is None: |
| 77 | return |
| 78 | input_batch = InputBatch.get_by_id(self.session, id = input_batch_id) |
| 79 | if not input_batch: |
no outgoing calls
no test coverage detected