Coverage for application / tator / routes.py: 18%
143 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-23 05:22 +0000
1"""
2General endpoints for Tator that are used throughout the application.
4/tator/login [POST]
5/tator/token [GET]
6/tator/logout [GET]
7/tator/projects [GET]
8/tator/sections?project=<project_id> [GET]
9/tator/refresh-sections [GET]
10/tator/frame/<media_id>/<frame> [GET]
11/tator/localization-image/<localization_id> [GET]
12/tator/localization [PATCH]
13/tator/localization/good-image [PATCH]
14"""
16import json
18import tator
19import requests
20from flask import current_app, request, session, Response
22from . import tator_bp
23from ..util.constants import TERM_YELLOW, TERM_RED, TERM_NORMAL
24from application.tator.tator_type import TatorLocalizationType
25from application.tator.tator_rest_client import TatorRestClient
28# log in to tator (get token from tator)
29@tator_bp.post('/login')
30def tator_login():
31 try:
32 token = TatorRestClient.login(
33 tator_url=current_app.config.get('TATOR_URL'),
34 username=request.values.get('username'),
35 password=request.values.get('password'),
36 )
37 session['tator_token'] = token
38 return {'username': request.values.get('username')}, 200
39 except requests.HTTPError as e:
40 return {}, e.response.status_code
43# check if stored tator token is valid
44@tator_bp.get('/token')
45def check_tator_token():
46 if 'tator_token' not in session.keys():
47 return {}, 400
48 try:
49 api = tator.get_api(
50 host=current_app.config.get('TATOR_URL'),
51 token=session['tator_token'],
52 )
53 print(f'Your Tator token: {session["tator_token"]}')
54 return {'username': api.whoami().username}, 200
55 except tator.openapi.tator_openapi.exceptions.ApiException:
56 return {}, 400
59# clears stored tator token
60@tator_bp.get('/logout')
61def tator_logout():
62 session.pop('tator_token', None)
63 return {}, 200
66# get a list of sections associated with a project from tator
67@tator_bp.get('/sections')
68def tator_sections():
69 def should_skip(section_path):
70 section_path_lower = section_path.lower()
71 return 'test' in section_path_lower or 'toplevelsectionname' in section_path_lower
73 project_id = request.args.get('project')
74 try:
75 sections = {}
76 section_list = tator.get_api(
77 host=current_app.config.get('TATOR_URL'),
78 token=session.get('tator_token'),
79 ).get_section_list(project_id)
80 # just doing two passes to simplify the logic
81 for section in section_list: # first pass - get top-level sections
82 if should_skip(section.path):
83 print(f'Skipping section with test path: "{section.path}" and name: "{section.name}"')
84 continue
85 path_parts = section.path.split('.')
86 if len(path_parts) != 1:
87 continue # not a top-level section
88 section_path_name = path_parts[0]
89 if section_path_name == 'None':
90 # handle case where top-level section is named "None" 🙄 (PNG DOEX010, Solomons DOEX010, etc?)
91 section_path_name = section.name
92 if sections.get(section_path_name):
93 print(f'{TERM_YELLOW}WARNING: duplicate expedition-level section name "{section_path_name}" detected{TERM_NORMAL}')
94 sections[section_path_name] = {
95 'id': section.id,
96 'name': section.name,
97 'folders': {},
98 }
99 for section in section_list: # second pass - get subsections
100 if should_skip(section.path):
101 continue
102 path_parts = section.path.split('.')
103 if len(path_parts) == 1:
104 continue
105 if len(path_parts) != 3:
106 if path_parts[1] == 'dscm' or path_parts[1] == 'sub':
107 continue
108 print(f'Skipping section with unexpected path format: "{section.path}"')
109 continue
110 parent_name, folder_name, _ = path_parts
111 if sections.get(parent_name) is None:
112 print(f'{TERM_YELLOW}WARNING: Skipping sub-section "{section.name}" because parent section "{parent_name}" was not found{TERM_NORMAL}')
113 continue
114 if sections[parent_name]['folders'].get(folder_name) is None:
115 sections[parent_name]['folders'][folder_name] = []
116 sections[parent_name]['folders'][folder_name].append({
117 'id': section.id,
118 'name': section.name,
119 })
120 return list(sections.values()), 200
121 except tator.openapi.tator_openapi.exceptions.ApiException as e:
122 print(f'{TERM_RED}ERROR: Unable to fetch Tator sections:{TERM_NORMAL} {e}')
123 return {'500': 'Error fetching Tator sections'}, 500
126# get a list of media ids marked as "transect" given a section id
127@tator_bp.get('/transects')
128def transects():
129 project_id = request.values.get('project')
130 section_ids = request.values.getlist('section')
131 try:
132 tator_api = tator.get_api(
133 host=current_app.config.get('TATOR_URL'),
134 token=session.get('tator_token'),
135 )
136 transect_media_ids = set()
137 state_list = tator_api.get_state_list(project_id, multi_section=section_ids)
138 for state in state_list:
139 if state.attributes.get('Mode') == 'Transect':
140 transect_media_ids.update(state.media)
141 if len(transect_media_ids) == 0:
142 print(f'{TERM_YELLOW}WARNING: No media ids marked as "transect" found for sections {section_ids}{TERM_NORMAL}')
143 return [], 200
144 media_name_id_list = []
145 media_list = tator_api.get_media_list(project_id, media_id=list(transect_media_ids))
146 for media in media_list:
147 media_name_id_list.append({'name': media.name, 'id': media.id})
148 return media_name_id_list, 200
149 except tator.openapi.tator_openapi.exceptions.ApiException as e:
150 print(f'{TERM_RED}ERROR: Unable to fetch transects list from Tator:{TERM_NORMAL} {e}')
151 return {'500': 'Error fetching Tator transects'}, 500
154# view tator video frame (not cropped)
155@tator_bp.get('/frame/<media_id>/<frame>')
156def tator_frame(media_id, frame):
157 token = session.get('tator_token') or request.args.get('token')
158 if not token:
159 return {}, 400
160 tator_client = TatorRestClient(current_app.config.get('TATOR_URL'), token)
161 quality = 650 if request.values.get('preview') else None
162 image = tator_client.get_frame(int(media_id), frame=int(frame), quality=quality)
163 return Response(image, content_type='image/png'), 200
166# view tator localization image (cropped)
167@tator_bp.get('/localization-image/<localization_id>')
168def tator_image(localization_id):
169 token = session.get('tator_token') or request.values.get('token')
170 if not token:
171 return {}, 400
172 tator_client = TatorRestClient(current_app.config.get('TATOR_URL'), token)
173 image = tator_client.get_localization_graphic(int(localization_id))
174 return Response(image, content_type='image/png'), 200
177# update tator localization
178@tator_bp.patch('/localization')
179def update_tator_localization():
180 localization_id_types = json.loads(request.values.get('localization_id_types'))
181 attributes = {
182 'Scientific Name': request.values.get('scientific_name'),
183 'Qualifier': request.values.get('qualifier'),
184 'Reason': request.values.get('reason'),
185 'Tentative ID': request.values.get('tentative_id'),
186 'IdentificationRemarks': request.values.get('identification_remarks'),
187 'Morphospecies': request.values.get('morphospecies'),
188 'Identified By': request.values.get('identified_by'),
189 'Notes': request.values.get('notes'),
190 }
191 if attracted := request.values.get('attracted'):
192 attributes['Attracted'] = attracted
193 if upon := request.values.get('upon'):
194 attributes['Upon'] = upon
195 try:
196 for localization in localization_id_types:
197 this_attributes = attributes.copy()
198 if TatorLocalizationType.is_dot(localization['type']):
199 this_attributes['Categorical Abundance'] = request.values.get('categorical_abundance') if request.values.get('categorical_abundance') else '--'
200 api = tator.get_api(
201 host=current_app.config.get('TATOR_URL'),
202 token=session.get('tator_token'),
203 )
204 api.update_localization_by_elemental_id(
205 version=localization['version'],
206 elemental_id=localization['elemental_id'],
207 localization_update=tator.models.LocalizationUpdate(
208 attributes=this_attributes,
209 )
210 )
211 except tator.openapi.tator_openapi.exceptions.ApiException as e:
212 print(f'{TERM_RED}ERROR: Unable to update Tator localization:{TERM_NORMAL} {e.body}')
213 return {}, 500
214 return {}, 200
217# update tator localization 'good image'
218@tator_bp.patch('/localization/good-image')
219def update_tator_localization_image():
220 localization_elemental_ids = request.values.getlist('localization_elemental_ids')
221 version = request.values.get('version')
222 try:
223 for elemental_id in localization_elemental_ids:
224 api = tator.get_api(
225 host=current_app.config.get('TATOR_URL'),
226 token=session.get('tator_token'),
227 )
228 api.update_localization_by_elemental_id(
229 version=version,
230 elemental_id=elemental_id,
231 localization_update=tator.models.LocalizationUpdate(
232 attributes={
233 'Good Image': True if request.values.get('good_image') == 'true' else False,
234 },
235 )
236 )
237 except tator.openapi.tator_openapi.exceptions.ApiException:
238 return {}, 500
239 return {}, 200