Coverage for application / tator / routes.py: 20%
122 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-07 06:46 +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_id> [GET]
9/tator/deployments/<project_id>/<section_id> [GET]
10/tator/refresh-sections [GET]
11/tator/frame/<media_id>/<frame> [GET]
12/tator/localization-image/<localization_id> [GET]
13/tator/localization [PATCH]
14/tator/localization/good-image [PATCH]
15"""
17import base64
18import json
20import tator
21import requests
22from flask import current_app, request, session, Response
24from . import tator_bp
25from ..util.constants import TERM_YELLOW, TERM_RED, TERM_NORMAL
26from ..util.tator_localization_type import TatorLocalizationType
29# log in to tator (get token from tator)
30@tator_bp.post('/login')
31def tator_login():
32 res = requests.post(
33 url=f'{current_app.config.get("TATOR_URL")}/rest/Token',
34 headers={'Content-Type': 'application/json'},
35 data=json.dumps({
36 'username': request.values.get('username'),
37 'password': request.values.get('password'),
38 'refresh': True,
39 }),
40 )
41 if res.status_code == 201:
42 session['tator_token'] = res.json()['token']
43 return {'username': request.values.get('username')}, 200
44 return {}, res.status_code
47# check if stored tator token is valid
48@tator_bp.get('/token')
49def check_tator_token():
50 if 'tator_token' not in session.keys():
51 return {}, 400
52 try:
53 api = tator.get_api(
54 host=current_app.config.get('TATOR_URL'),
55 token=session['tator_token'],
56 )
57 print(f'Your Tator token: {session["tator_token"]}')
58 return {'username': api.whoami().username}, 200
59 except tator.openapi.tator_openapi.exceptions.ApiException:
60 return {}, 400
63# clears stored tator token
64@tator_bp.get('/logout')
65def tator_logout():
66 session.pop('tator_token', None)
67 return {}, 200
70# get a list of sections associated with a project from tator
71@tator_bp.get('/sections/<project_id>')
72def tator_sections(project_id):
73 def should_skip(section_path):
74 section_path_lower = section_path.lower()
75 return 'test' in section_path_lower or 'toplevelsectionname' in section_path_lower
77 try:
78 sections = {}
79 section_list = tator.get_api(
80 host=current_app.config.get('TATOR_URL'),
81 token=session.get('tator_token'),
82 ).get_section_list(project_id)
83 # just doing two passes to simplify the logic
84 for section in section_list: # first pass - get top-level sections
85 if should_skip(section.path):
86 print(f'Skipping section with test path: "{section.path}" and name: "{section.name}"')
87 continue
88 path_parts = section.path.split('.')
89 if len(path_parts) != 1:
90 continue # not a top-level section
91 section_path_name = path_parts[0]
92 if section_path_name == 'None':
93 # handle case where top-level section is named "None" 🙄 (PNG DOEX010, Solomons DOEX010, etc?)
94 section_path_name = section.name
95 if sections.get(section_path_name):
96 print(f'{TERM_YELLOW}WARNING: duplicate expedition-level section name "{section_path_name}" detected{TERM_NORMAL}')
97 sections[section_path_name] = {
98 'id': section.id,
99 'name': section.name,
100 'folders': {},
101 }
102 for section in section_list: # second pass - get subsections
103 if should_skip(section.path):
104 continue
105 path_parts = section.path.split('.')
106 if len(path_parts) == 1:
107 continue
108 if len(path_parts) != 3:
109 if path_parts[1] == 'dscm' or path_parts[1] == 'sub':
110 continue
111 print(f'Skipping section with unexpected path format: "{section.path}"')
112 continue
113 parent_name, folder_name, _ = path_parts
114 if sections.get(parent_name) is None:
115 print(f'{TERM_YELLOW}WARNING: Skipping sub-section "{section.name}" because parent section "{parent_name}" was not found{TERM_NORMAL}')
116 continue
117 if sections[parent_name]['folders'].get(folder_name) is None:
118 sections[parent_name]['folders'][folder_name] = []
119 sections[parent_name]['folders'][folder_name].append({
120 'id': section.id,
121 'name': section.name,
122 })
123 return list(sections.values()), 200
124 except tator.openapi.tator_openapi.exceptions.ApiException as e:
125 print(f'{TERM_RED}ERROR: Unable to fetch Tator sections:{TERM_NORMAL} {e}')
126 return {'500': 'Error fetching Tator sections'}, 500
129# view tator video frame (not cropped)
130@tator_bp.get('/frame/<media_id>/<frame>')
131def tator_frame(media_id, frame):
132 if 'tator_token' in session.keys():
133 token = session['tator_token']
134 else:
135 token = request.args.get('token')
136 url = f'{current_app.config.get("TATOR_URL")}/rest/GetFrame/{media_id}?frames={frame}'
137 if request.values.get('preview'):
138 url += '&quality=650'
139 res = requests.get(
140 url=url,
141 headers={'Authorization': f'Token {token}'}
142 )
143 if res.status_code == 200:
144 base64_image = base64.b64encode(res.content).decode('utf-8')
145 return Response(base64.b64decode(base64_image), content_type='image/png'), 200
146 return '', 500
149# view tator localization image (cropped)
150@tator_bp.get('/localization-image/<localization_id>')
151def tator_image(localization_id):
152 if not session.get('tator_token'):
153 if not request.values.get('token'):
154 return {}, 400
155 token = request.values.get('token')
156 else:
157 token = session["tator_token"]
158 res = requests.get(
159 url=f'{current_app.config.get("TATOR_URL")}/rest/LocalizationGraphic/{localization_id}',
160 headers={'Authorization': f'Token {token}'}
161 )
162 if res.status_code == 200:
163 base64_image = base64.b64encode(res.content).decode('utf-8')
164 return Response(base64.b64decode(base64_image), content_type='image/png'), 200
165 return '', 500
168# update tator localization
169@tator_bp.patch('/localization')
170def update_tator_localization():
171 localization_id_types = json.loads(request.values.get('localization_id_types'))
172 attributes = {
173 'Scientific Name': request.values.get('scientific_name'),
174 'Qualifier': request.values.get('qualifier'),
175 'Reason': request.values.get('reason'),
176 'Tentative ID': request.values.get('tentative_id'),
177 'IdentificationRemarks': request.values.get('identification_remarks'),
178 'Morphospecies': request.values.get('morphospecies'),
179 'Identified By': request.values.get('identified_by'),
180 'Notes': request.values.get('notes'),
181 'Attracted': request.values.get('attracted'),
182 }
183 try:
184 for localization in localization_id_types:
185 this_attributes = attributes.copy()
186 if localization['type'] == TatorLocalizationType.DOT.value:
187 this_attributes['Categorical Abundance'] = request.values.get('categorical_abundance') if request.values.get('categorical_abundance') else '--'
188 api = tator.get_api(
189 host=current_app.config.get('TATOR_URL'),
190 token=session.get('tator_token'),
191 )
192 api.update_localization_by_elemental_id(
193 version=localization['version'],
194 elemental_id=localization['elemental_id'],
195 localization_update=tator.models.LocalizationUpdate(
196 attributes=this_attributes,
197 )
198 )
199 except tator.openapi.tator_openapi.exceptions.ApiException:
200 return {}, 500
201 return {}, 200
204# update tator localization 'good image'
205@tator_bp.patch('/localization/good-image')
206def update_tator_localization_image():
207 localization_elemental_ids = request.values.getlist('localization_elemental_ids')
208 version = request.values.get('version')
209 try:
210 for elemental_id in localization_elemental_ids:
211 api = tator.get_api(
212 host=current_app.config.get('TATOR_URL'),
213 token=session.get('tator_token'),
214 )
215 api.update_localization_by_elemental_id(
216 version=version,
217 elemental_id=elemental_id,
218 localization_update=tator.models.LocalizationUpdate(
219 attributes={
220 'Good Image': True if request.values.get('good_image') == 'true' else False,
221 },
222 )
223 )
224 except tator.openapi.tator_openapi.exceptions.ApiException:
225 return {}, 500
226 return {}, 200