You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

2237 lines
73 KiB

from typing import TypedDict, Dict, List
from flask import Flask, redirect, render_template, current_app, g, jsonify, request, abort
from flask_cors import cross_origin, CORS # type: ignore
import os
from pathlib import Path
import psycopg2 # type: ignore
from psycopg2.extras import NamedTupleCursor # type: ignore
DATABASE_URL = os.environ.get("DATABASE_URL")
SECRET_KEY = os.environ.get("SECRET_KEY")
FLASK_ENV = os.environ.get("FLASK_ENV")
if (FLASK_ENV == "development"):
app = Flask(__name__, template_folder='dist/', static_folder='dist/')
else:
app = Flask(__name__, template_folder='dist/')
app.config['SECRET_KEY'] = SECRET_KEY
if app.debug:
app.config['SEND_FILE_MAX_AGE_DEFAULT'] = 0
CORS(app)
Path('/tmp/app-initialized').touch()
def get_db():
if 'db' not in g:
g.db = psycopg2.connect(DATABASE_URL, sslmode='require', cursor_factory=NamedTupleCursor)
return g.db
def close_db(e=None):
db = g.pop('db', None)
if db is not None:
db.close()
app.teardown_appcontext(close_db)
@app.route('/')
def index():
return render_template('index.html')
@app.route('/submit/<key>')
def submit(key):
if is_create_key_valid(key):
return render_template('submit.html', version='current')
return render_template('404.html')
@app.route('/submit2020/<key>')
def submit2020(key):
if is_create_key_valid(key):
return render_template('submit.html', version='2020')
return render_template('404.html')
@app.route('/submit2021/<key>')
def submit2021(key):
if is_create_key_valid(key):
return render_template('submit.html', version='2021')
return render_template('404.html')
@app.route('/proof/<key>')
def proof(key):
if is_proof_key_valid(key):
return render_template('proof.html')
return render_template('404.html')
def is_proof_key_valid(key):
with get_db() as db:
with db.cursor() as cur:
try:
cur.execute("""
SELECT id
FROM proof_keys
WHERE id = %(id)s AND is_enabled = true
""", {
'id': key,
})
if cur.rowcount == 1:
return True
except Exception as e:
print(e)
return False
def is_create_key_valid(key):
with get_db() as db:
with db.cursor() as cur:
try:
cur.execute("""
SELECT id
FROM create_keys
WHERE id = %(id)s AND is_enabled = true
""", {
'id': key,
})
if cur.rowcount == 1:
return True
except Exception as e:
print(e)
return False
@app.route('/data/colonies')
def get_colonies():
with get_db() as db:
with db.cursor() as cur:
cur.execute("""SELECT id, name FROM colony ORDER BY name""")
rows = cur.fetchall()
return {
"colonies": [{"name":i.name, "id":i.id} for i in rows],
}
@app.route('/data/proof/<key>/inbox', methods=['POST'])
def get_proof_inbox(key):
if not is_proof_key_valid(key):
abort(404)
data = request.get_json()
filterField = data["filterBy"]
filterValue = ""
if data["filterValue"]:
filterValue = int(data["filterValue"])
# Parse filter criterion
where_by = ""
if filterField:
if filterField == "colony":
where_by = "data_sheet.colony_id "
elif filterField == "year":
where_by = "extract(year from data_sheet.date) "
if filterValue:
if filterField == "year":
filterValue = int(filterValue)
where_by += "= %(filterValue)s"
else:
where_by += "IS NULL"
query_string = """
SELECT
data_sheet.id,
data_sheet.date,
data_sheet.entered_by_name,
colony.name
FROM
data_sheet,
colony
WHERE
"""
if where_by:
query_string += where_by
query_string += " AND "
query_string += """data_sheet.colony_id = colony.id
AND
data_sheet.proofed_by IS NULL
ORDER BY
data_sheet.date DESC, data_sheet.start_time DESC
"""
with get_db() as db:
with db.cursor() as cur:
cur.execute(query_string, {
'filterValue': filterValue,
})
data_sheets = []
for row in cur:
ds = {
"id": row.id,
"date": row.date.isoformat(),
"enteredByName": row.entered_by_name,
"colonyName": row.name,
}
data_sheets.append(ds)
return {
"dataSheets": data_sheets,
}
@app.route('/data/proof/<key>/completed', methods=['POST'])
def get_proof_completed(key):
if not is_proof_key_valid(key):
abort(404)
data = request.get_json()
filterField = data["filterBy"]
filterValue = ""
if data["filterValue"]:
filterValue = int(data["filterValue"])
# Parse filter criterion
where_by = ""
if filterField:
if filterField == "colony":
where_by = "data_sheet.colony_id "
elif filterField == "year":
where_by = "extract(year from data_sheet.date) "
if filterValue:
if filterField == "year":
filterValue = int(filterValue)
where_by += "= %(filterValue)s"
else:
where_by += "IS NULL"
query_string = """
SELECT
data_sheet.id,
data_sheet.date,
data_sheet.entered_by_name,
colony.name
FROM
data_sheet,
colony
WHERE
"""
if where_by:
query_string += where_by
query_string += " AND "
query_string += """data_sheet.colony_id = colony.id
AND
data_sheet.proofed_by IS NOT NULL
ORDER BY
data_sheet.date DESC, data_sheet.start_time DESC
"""
with get_db() as db:
with db.cursor() as cur:
cur.execute(query_string, {
'filterValue': filterValue,
})
data_sheets = []
for row in cur:
ds = {
"id": row.id,
"date": row.date.isoformat(),
"enteredByName": row.entered_by_name,
"colonyName": row.name,
}
data_sheets.append(ds)
return {
"dataSheets": data_sheets,
}
@app.route('/data/observers')
def get_observers():
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
SELECT name
FROM observers
ORDER BY name
""")
observers = []
for row in cur:
observers.append(row.name)
return {
"observers": observers,
}
@app.route('/data/duplicate', methods=['POST'])
def check_duplicate_survey():
data = request.get_json()
survey_date = data["date"]
colony_id = data["colonyId"]
with get_db() as db:
with db.cursor() as cur:
cur.execute("""SELECT
id
FROM
data_sheet
WHERE colony_id = %(colony_id)s AND date = %(survey_date)s
""", {
"colony_id": colony_id,
"survey_date": survey_date,
})
return jsonify(cur.rowcount > 0)
class SurveySummary(TypedDict, total=False):
totalNests: int
totalAdults: int
totalYoung: int
totalPossibleNests: int
speciesType: str
class HepNestData(TypedDict, total=False):
number: int
isFocal: bool
speciesCode: str
active: str
stage: int
adultCount: int
chickCount: int
chickConfidence: bool
comments: str
class HepNestStageData(TypedDict, total=False):
speciesCode: str
stage0: int
stage1: int
stage2: int
stage3: int
stage4: int
stage5: int
unknownStage: int
class HepS4BroodData(TypedDict, total=False):
speciesCode: str
oneChickNests: int
twoChickNests: int
threeChickNests: int
fourChickNests: int
fiveChickNests: int
chicksNoNest: int
class GuteNestData(TypedDict, total=False):
subcolony: str
speciesCode: str
totalAdults: int
stage0Adults: int
stage1Nests: int
stage2Chicks: int
stage3Chicks: int
stage4Chicks: int
comments: str
location: str
includedOnMap: bool
class DisturbanceData(TypedDict, total=False):
observedInferred: str
distType: str
distResult: str
distDescription: str
class DataSheetData(TypedDict, total=False):
authKey: str
id: str
submitterName: str
colonyId: int
date: str
startTime: str
endTime: str
observers: List[str]
nestVisibility: str
visibilityComments: str
surveySummary: Dict[str, SurveySummary]
hepNestStageData: List[HepNestStageData]
hepS4BroodData: List[HepS4BroodData]
guteNestData: List[GuteNestData]
disturbanceData: List[DisturbanceData]
colonyAdditionalObservations: str
proofedBy: str
class DataSheetData2021(TypedDict, total=False):
authKey: str
id: str
submitterName: str
colonyId: int
date: str
startTime: str
endTime: str
observers: List[str]
nestVisibility: str
visibilityComments: str
surveySummary: Dict[str, SurveySummary]
hepNestData: List[HepNestData]
guteNestData: List[GuteNestData]
disturbanceData: List[DisturbanceData]
proofedBy: str
class DataSheetData2020(TypedDict, total=False):
authKey: str
id: str
submitterName: str
colonyId: int
date: str
startTime: str
endTime: str
observers: List[str]
nestVisibility: str
visibilityComments: str
surveySummary: Dict[str, SurveySummary]
colonySignificantChanges: bool
colonySignificantChangesNotes: str
colonyHumanDisturbance: bool
colonyHumanDisturbanceNotes: str
colonyAdditionalObservations: str
hepNestData: List[HepNestData]
guteNestData: List[GuteNestData]
proofedBy: str
def save_observers_helper(cur, observers, data_sheet_id):
# make sure observer records exist for all observers
for observer in observers:
cur.execute("""
INSERT INTO observers (
name
) VALUES (
%(name)s
)
ON CONFLICT (name) DO NOTHING
""", {
"name": observer,
})
# get the ids for all observers
cur.execute("""
SELECT
id
FROM observers
WHERE name IN %(names)s
""", {
"names": tuple(observers),
})
# collect all those ids in to a nice list
observer_ids = []
for o_row in cur:
observer_ids.append(o_row.id)
# create the observer records for this datasheet
for id in observer_ids:
cur.execute("""
INSERT INTO data_sheet_observers (
data_sheet_id,
observer_id
) VALUES (
%(data_sheet_id)s,
%(observer_id)s
)
""", {
"data_sheet_id": data_sheet_id,
"observer_id": id,
})
def save_summary_helper(cur, survey_summary, data_sheet_id):
# Save the survey summary data: these are the overall nest counts.
for species_code, summary in survey_summary.items():
total_nests = int(summary['totalNests'])
total_adults = int(summary['totalAdults'])
total_young = int(summary['totalYoung'])
total_possible_nests = None
if 'totalPossibleNests' in summary:
total_possible_nests = int(summary['totalPossibleNests'])
species_type = str(summary['speciesType'])
cur.execute("""
INSERT INTO survey_summary (
data_sheet_id,
species_code,
total_nests,
total_adults,
total_young,
total_possible_nests,
species_type
) VALUES (
%(data_sheet_id)s,
%(species_code)s,
%(total_nests)s,
%(total_adults)s,
%(total_young)s,
%(total_possible_nests)s,
%(species_type)s
)
""", {
'data_sheet_id': data_sheet_id,
'species_code': species_code,
'total_nests': total_nests,
'total_adults': total_adults,
'total_young': total_young,
'total_possible_nests': total_possible_nests,
'species_type': species_type,
})
def save_hep_nests_helper(cur, hep_nest_data, data_sheet_id):
for hep_nest in hep_nest_data:
number = int(hep_nest['number'])
is_focal = bool(hep_nest['isFocal'])
species_code = str(hep_nest['speciesCode'])
active = str(hep_nest['active'])
stage = int(hep_nest['stage']) if hep_nest['stage'] is not None else None
adult_count = int(hep_nest['adultCount'])
chick_count = int(hep_nest['chickCount'])
chick_confidence = bool(hep_nest['chickConfidence'])
comments = str(hep_nest['comments'])
cur.execute("""
INSERT INTO hep_nest_data (
data_sheet_id,
nest_number,
focal,
species_code,
active,
stage,
adult_count,
chick_count,
chick_confidence,
comments
) VALUES (
%(data_sheet_id)s,
%(nest_number)s,
%(focal)s,
%(species_code)s,
%(active)s,
%(stage)s,
%(adult_count)s,
%(chick_count)s,
%(chick_confidence)s,
%(comments)s
)
""", {
'data_sheet_id': data_sheet_id,
'nest_number': number,
'focal': is_focal,
'species_code': species_code,
'active': active,
'stage': stage,
'adult_count': adult_count,
'chick_count': chick_count,
'chick_confidence': chick_confidence,
'comments': comments,
})
def save_hep_nest_stage_helper(cur, hep_nest_stage_data, data_sheet_id):
for idx, hep_stage_row in enumerate(hep_nest_stage_data):
species_code = str(hep_stage_row['speciesCode'])
stage0 = int(hep_stage_row['stage0'])
stage1 = int(hep_stage_row['stage1'])
stage2 = int(hep_stage_row['stage2'])
stage3 = int(hep_stage_row['stage3'])
stage4 = int(hep_stage_row['stage4'])
stage5 = int(hep_stage_row['stage5'])
unknown_stage = int(hep_stage_row['unknownStage'])
cur.execute("""
INSERT INTO hep_stage_counts (
data_sheet_id,
species_code,
stage_0,
stage_1,
stage_2,
stage_3,
stage_4,
stage_5,
stage_unknown,
index
) VALUES (
%(data_sheet_id)s,
%(species_code)s,
%(stage_0)s,
%(stage_1)s,
%(stage_2)s,
%(stage_3)s,
%(stage_4)s,
%(stage_5)s,
%(stage_unknown)s,
%(index)s
)
""", {
'data_sheet_id': data_sheet_id,
'species_code': species_code,
'stage_0': stage0,
'stage_1': stage1,
'stage_2': stage2,
'stage_3': stage3,
'stage_4': stage4,
'stage_5': stage5,
'stage_unknown': unknown_stage,
'index': idx,
})
def save_hep_s4_brood_helper(cur, hep_s4_brood_data, data_sheet_id):
for idx, hep_s4_brood_row in enumerate(hep_s4_brood_data):
species_code = str(hep_s4_brood_row['speciesCode'])
one_chick_nests = int(hep_s4_brood_row['oneChickNests'])
two_chick_nests = int(hep_s4_brood_row['twoChickNests'])
three_chick_nests = int(hep_s4_brood_row['threeChickNests'])
four_chick_nests = int(hep_s4_brood_row['fourChickNests'])
five_chick_nests = int(hep_s4_brood_row['fiveChickNests'])
chicks_no_nest = int(hep_s4_brood_row['chicksNoNest'])
cur.execute("""
INSERT INTO hep_stage_4_brood_size (
data_sheet_id,
species_code,
one_chick_nests,
two_chick_nests,
three_chick_nests,
four_chick_nests,
five_chick_nests,
chicks_no_nest,
index
) VALUES (
%(data_sheet_id)s,
%(species_code)s,
%(one_chick_nests)s,
%(two_chick_nests)s,
%(three_chick_nests)s,
%(four_chick_nests)s,
%(five_chick_nests)s,
%(chicks_no_nest)s,
%(index)s
)
""", {
'data_sheet_id': data_sheet_id,
'species_code': species_code,
'one_chick_nests': one_chick_nests,
'two_chick_nests': two_chick_nests,
'three_chick_nests': three_chick_nests,
'four_chick_nests': four_chick_nests,
'five_chick_nests': five_chick_nests,
'chicks_no_nest': chicks_no_nest,
'index': idx,
})
def save_gute_nests_helper(cur, gute_nest_data, data_sheet_id):
for idx, gute_nest in enumerate(gute_nest_data):
subcolony = None
if 'subcolony' in gute_nest_data:
subcolony = str(gute_nest_data['gute_nest_data'])
species_code = str(gute_nest['speciesCode'])
total_adults = int(gute_nest['totalAdults'])
if gute_nest['stage0Adults'] is None:
stage_0_adults = None
else:
stage_0_adults = int(gute_nest['stage0Adults'])
stage_1_nests = int(gute_nest['stage1Nests'])
stage_2_chicks = int(gute_nest['stage2Chicks'])
stage_3_chicks = int(gute_nest['stage3Chicks'])
stage_4_chicks = int(gute_nest['stage4Chicks'])
comments = str(gute_nest['comments'])
location = None
if 'location' in gute_nest_data:
location = str(gute_nest['location'])
included_on_map = None
if 'included_on_map' in gute_nest_data:
included_on_map = str(gute_nest['includedOnMap'])
cur.execute("""
INSERT INTO gute_nest_data (
data_sheet_id,
index,
subcolony,
species_code,
total_adults,
stage_0_adults,
stage_1_nests,
stage_2_chicks,
stage_3_chicks,
stage_4_chicks,
comments,
location,
included_on_map
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(subcolony)s,
%(species_code)s,
%(total_adults)s,
%(stage_0_adults)s,
%(stage_1_nests)s,
%(stage_2_chicks)s,
%(stage_3_chicks)s,
%(stage_4_chicks)s,
%(comments)s,
%(location)s,
%(included_on_map)s
)
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'subcolony': subcolony,
'species_code': species_code,
'total_adults': total_adults,
'stage_0_adults': stage_0_adults,
'stage_1_nests': stage_1_nests,
'stage_2_chicks': stage_2_chicks,
'stage_3_chicks': stage_3_chicks,
'stage_4_chicks': stage_4_chicks,
'comments': comments,
'location': location,
'included_on_map': included_on_map,
})
@app.route('/data/sheet/save', methods=['POST'])
def save_sheet():
data: DataSheetData = request.get_json()
auth_key = str(data['authKey'])
entered_by_name = str(data['submitterName'])
colony_id = int(data['colonyId'])
date = str(data['date'])
start_time = str(data['startTime'])
end_time = str(data['endTime'])
nest_visibility = str(data['nestVisibility'])
observers = list(data['observers'])
visibility_comments = str(data['visibilityComments'])
survey_summary = data['surveySummary']
hep_nest_stage_data = data['hepNestStageData']
hep_s4_brood_data = data['hepS4BroodData']
gute_nest_data = data['guteNestData']
disturbance_data = data['disturbanceData']
additional_observations = str(data['colonyAdditionalObservations'])
if not is_create_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
INSERT INTO data_sheet (
colony_id,
date,
start_time,
end_time,
nest_visibility,
visibility_comments,
entered_by_name,
additional_observations
) VALUES (
%(colony_id)s,
%(date)s,
%(start_time)s,
%(end_time)s,
%(nest_visibility)s,
%(visibility_comments)s,
%(entered_by_name)s,
%(additional_observations)s
)
RETURNING id
""", {
'entered_by_name': entered_by_name,
'colony_id': colony_id,
'date': date,
'start_time': start_time,
'end_time': end_time,
'nest_visibility': nest_visibility,
'visibility_comments': visibility_comments,
'additional_observations': additional_observations,
})
row = cur.fetchone()
data_sheet_id = row.id
# Save the observer data
save_observers_helper(cur, observers, data_sheet_id)
# Save the summary data (totals for each species)
save_summary_helper(cur, survey_summary, data_sheet_id)
# Save the HEP nest stage data
save_hep_nest_stage_helper(cur, hep_nest_stage_data, data_sheet_id)
# Save the HEP s4 brood size data
save_hep_s4_brood_helper(cur, hep_s4_brood_data, data_sheet_id)
# Save the GUTE subcolony data
save_gute_nests_helper(cur, gute_nest_data, data_sheet_id)
for idx, disturbance in enumerate(disturbance_data):
observed_inferred = str(disturbance['observedInferred'])
dist_type = str(disturbance['distType'])
dist_result = str(disturbance['distResult'])
dist_description = str(disturbance['distDescription'])
cur.execute("""
INSERT INTO disturbance_data (
data_sheet_id,
index,
observed_inferred,
dist_type,
dist_result,
dist_description
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(observed_inferred)s,
%(dist_type)s,
%(dist_result)s,
%(dist_description)s
)
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'observed_inferred': observed_inferred,
'dist_type': dist_type,
'dist_result': dist_result,
'dist_description': dist_description,
})
return jsonify(data)
@app.route('/data/sheet/save2021', methods=['POST'])
def save_sheet_2021():
data: DataSheetData2021 = request.get_json()
auth_key = str(data['authKey'])
entered_by_name = str(data['submitterName'])
colony_id = int(data['colonyId'])
date = str(data['date'])
start_time = str(data['startTime'])
end_time = str(data['endTime'])
nest_visibility = str(data['nestVisibility'])
observers = list(data['observers'])
visibility_comments = str(data['visibilityComments'])
survey_summary = data['surveySummary']
hep_nest_data = data['hepNestData']
gute_nest_data = data['guteNestData']
disturbance_data = data['disturbanceData']
if not is_create_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
INSERT INTO data_sheet (
colony_id,
date,
start_time,
end_time,
nest_visibility,
visibility_comments,
entered_by_name
) VALUES (
%(colony_id)s,
%(date)s,
%(start_time)s,
%(end_time)s,
%(nest_visibility)s,
%(visibility_comments)s,
%(entered_by_name)s
)
RETURNING id
""", {
'entered_by_name': entered_by_name,
'colony_id': colony_id,
'date': date,
'start_time': start_time,
'end_time': end_time,
'nest_visibility': nest_visibility,
'visibility_comments': visibility_comments,
})
row = cur.fetchone()
data_sheet_id = row.id
# Save the observer data
save_observers_helper(cur, observers, data_sheet_id)
# Save the summary data (totals for each species)
save_summary_helper(cur, survey_summary, data_sheet_id)
# Save the HEP nest data
save_hep_nests_helper(cur, hep_nest_data, data_sheet_id)
# Save the GUTE subcolony data
save_gute_nests_helper(cur, gute_nest_data, data_sheet_id)
for idx, disturbance in enumerate(disturbance_data):
observed_inferred = str(disturbance['observedInferred'])
dist_type = str(disturbance['distType'])
dist_result = str(disturbance['distResult'])
dist_description = str(disturbance['distDescription'])
cur.execute("""
INSERT INTO disturbance_data (
data_sheet_id,
index,
observed_inferred,
dist_type,
dist_result,
dist_description
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(observed_inferred)s,
%(dist_type)s,
%(dist_result)s,
%(dist_description)s
)
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'observed_inferred': observed_inferred,
'dist_type': dist_type,
'dist_result': dist_result,
'dist_description': dist_description,
})
return jsonify(data)
@app.route('/data/sheet/save2020', methods=['POST'])
def save_sheet_2020():
data: DataSheetData2020 = request.get_json()
auth_key = str(data['authKey'])
entered_by_name = str(data['submitterName'])
colony_id = int(data['colonyId'])
date = str(data['date'])
start_time = str(data['startTime'])
end_time = str(data['endTime'])
nest_visibility = str(data['nestVisibility'])
observers = list(data['observers'])
visibility_comments = str(data['visibilityComments'])
survey_summary = data['surveySummary']
significant_changes = bool(data['colonySignificantChanges'])
significant_changes_notes = str(data['colonySignificantChangesNotes'])
human_disturbance = bool(data['colonyHumanDisturbance'])
human_disturbance_notes = str(data['colonyHumanDisturbanceNotes'])
additional_observations = str(data['colonyAdditionalObservations'])
hep_nest_data = data['hepNestData']
gute_nest_data = data['guteNestData']
if not is_create_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
INSERT INTO data_sheet (
colony_id,
date,
start_time,
end_time,
nest_visibility,
visibility_comments,
significant_changes,
significant_change_notes,
human_disturbance,
human_disturbance_notes,
additional_observations,
entered_by_name
) VALUES (
%(colony_id)s,
%(date)s,
%(start_time)s,
%(end_time)s,
%(nest_visibility)s,
%(visibility_comments)s,
%(significant_changes)s,
%(significant_change_notes)s,
%(human_disturbance)s,
%(human_disturbance_notes)s,
%(additional_observations)s,
%(entered_by_name)s
)
RETURNING id
""", {
'entered_by_name': entered_by_name,
'colony_id': colony_id,
'date': date,
'start_time': start_time,
'end_time': end_time,
'nest_visibility': nest_visibility,
'visibility_comments': visibility_comments,
'significant_changes': significant_changes,
'significant_change_notes': significant_changes_notes,
'human_disturbance': human_disturbance,
'human_disturbance_notes': human_disturbance_notes,
'additional_observations': additional_observations,
})
row = cur.fetchone()
data_sheet_id = row.id
# Save the observer data
save_observers_helper(cur, observers, data_sheet_id)
# Save the summary data (totals for each species)
save_summary_helper(cur, survey_summary, data_sheet_id)
# Save the HEP nest data
save_hep_nests_helper(cur, hep_nest_data, data_sheet_id)
# Save the GUTE subcolony data
save_gute_nests_helper(cur, gute_nest_data, data_sheet_id)
return jsonify(data)
def load_observers_helper(cur, data_sheet, data_sheet_id):
cur.execute("""
SELECT
observers.name
FROM
observers,
data_sheet_observers
WHERE
observers.id = data_sheet_observers.observer_id
AND
data_sheet_observers.data_sheet_id = %(data_sheet_id)s
""", {
"data_sheet_id": data_sheet_id,
})
observers: List[str] = []
for row in cur:
observers.append(row.name)
data_sheet["observers"] = observers
def load_summary_helper(cur, data_sheet, data_sheet_id):
cur.execute("""
SELECT
id,
species_code,
total_nests,
total_adults,
total_young,
total_possible_nests,
species_type
FROM survey_summary
WHERE data_sheet_id = %(data_sheet_id)s
""", {
"data_sheet_id": data_sheet_id,
})
survey_summary: Dict[str, SurveySummary] = {}
for row in cur:
summary: SurveySummary = {
"totalNests": row.total_nests,
"totalAdults": row.total_adults,
"totalYoung": row.total_young,
"totalPossibleNests": row.total_possible_nests,
"speciesType": row.species_type,
}
survey_summary[row.species_code] = summary
data_sheet['surveySummary'] = survey_summary
def load_hep_nests_helper(cur, data_sheet, data_sheet_id):
cur.execute("""
SELECT
nest_number,
focal,
species_code,
active,
stage,
adult_count,
chick_count,
chick_confidence,
comments
FROM hep_nest_data
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY nest_number
""", {
"data_sheet_id": data_sheet_id,
})
hep_nest_data: List[HepNestData] = []
for row in cur:
hep_data: HepNestData = {
"number": row.nest_number,
"isFocal": row.focal,
"speciesCode": row.species_code,
"active": row.active,
"stage": row.stage,
"adultCount": row.adult_count,
"chickCount": row.chick_count,
"chickConfidence": row.chick_confidence,
"comments": row.comments,
}
hep_nest_data.append(hep_data)
data_sheet['hepNestData'] = hep_nest_data
def load_hep_nest_stage_helper(cur, data_sheet, data_sheet_id):
cur.execute("""
SELECT
index,
species_code,
stage_0,
stage_1,
stage_2,
stage_3,
stage_4,
stage_5,
stage_unknown
FROM hep_stage_counts
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY index
""", {
"data_sheet_id": data_sheet_id,
})
hep_nest_stage_data: List[HepNestStageData] = []
for row in cur:
hep_stage: HepNestStageData = {
"speciesCode": row.species_code,
"stage0": row.stage_0,
"stage1": row.stage_1,
"stage2": row.stage_2,
"stage3": row.stage_3,
"stage4": row.stage_4,
"stage5": row.stage_5,
"unknownStage": row.stage_unknown,
}
hep_nest_stage_data.append(hep_stage)
data_sheet['hepNestStageData'] = hep_nest_stage_data
def load_hep_s4_brood_helper(cur, data_sheet, data_sheet_id):
cur.execute("""
SELECT
species_code,
one_chick_nests,
two_chick_nests,
three_chick_nests,
four_chick_nests,
five_chick_nests,
chicks_no_nest,
index
FROM hep_stage_4_brood_size
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY index
""", {
"data_sheet_id": data_sheet_id,
})
hep_s4_brood_data: List[HepS4BroodData] = []
for row in cur:
hep_s4_row: HepS4BroodData = {
"speciesCode": row.species_code,
"oneChickNests": row.one_chick_nests,
"twoChickNests": row.two_chick_nests,
"threeChickNests": row.three_chick_nests,
"fourChickNests": row.four_chick_nests,
"fiveChickNests": row.five_chick_nests,
"chicksNoNest": row.chicks_no_nest,
}
hep_s4_brood_data.append(hep_s4_row)
data_sheet['hepS4BroodData'] = hep_s4_brood_data
def load_gute_nests_helper(cur, data_sheet, data_sheet_id):
cur.execute("""
SELECT
index,
subcolony,
species_code,
total_adults,
stage_0_adults,
stage_1_nests,
stage_2_chicks,
stage_3_chicks,
stage_4_chicks,
comments,
location,
included_on_map
FROM gute_nest_data
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY index
""", {
"data_sheet_id": data_sheet_id,
})
gute_nest_data: List[GuteNestData] = []
for row in cur:
gute: GuteNestData = {
"subcolony": row.subcolony,
"speciesCode": row.species_code,
"totalAdults": row.total_adults,
"stage0Adults": row.stage_0_adults,
"stage1Nests": row.stage_1_nests,
"stage2Chicks": row.stage_2_chicks,
"stage3Chicks": row.stage_3_chicks,
"stage4Chicks": row.stage_4_chicks,
"comments": row.comments,
"location": row.location,
"includedOnMap": row.included_on_map,
}
gute_nest_data.append(gute)
data_sheet['guteNestData'] = gute_nest_data
@app.route('/data/sheet/load', methods=['POST'])
def load_sheet():
data_sheet_id = request.form.get("data_sheet_id")
auth_key = request.form.get("auth_key")
if not is_proof_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
SELECT
id,
colony_id,
date,
start_time,
end_time,
nest_visibility,
visibility_comments,
additional_observations,
entered_by_name,
proofed_by
FROM data_sheet
WHERE id = %(data_sheet_id)s
""", {
"data_sheet_id": data_sheet_id,
})
if cur.rowcount == 0:
raise ValueError("data_sheet_id doesn't exist")
row = cur.fetchone()
data_sheet: DataSheetData = {
"version": "current",
"id": row.id,
"proofedBy": row.proofed_by,
"submitterName": row.entered_by_name,
"colonyId": row.colony_id,
"date": row.date.isoformat(),
"startTime": row.start_time.isoformat('minutes'),
"endTime": row.end_time.isoformat('minutes'),
"nestVisibility": row.nest_visibility,
"colonyAdditionalObservations": row.additional_observations,
"visibilityComments": row.visibility_comments,
}
# Load the observer data
load_observers_helper(cur, data_sheet, data_sheet_id)
# Load the summary data
load_summary_helper(cur, data_sheet, data_sheet_id)
# Load the HEP nest data
load_hep_nest_stage_helper(cur, data_sheet, data_sheet_id)
# Load the HEP S4 brood size data
load_hep_s4_brood_helper(cur, data_sheet, data_sheet_id)
# Load the GUTE nest data
load_gute_nests_helper(cur, data_sheet, data_sheet_id)
# DISTURBANCE DATA
cur.execute("""
SELECT
index,
observed_inferred,
dist_type,
dist_description,
dist_result
FROM disturbance_data
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY index
""", {
"data_sheet_id": data_sheet_id,
})
disturbance_data: List[DisturbanceData] = []
for row in cur:
disturbance: DisturbanceData = {
"observedInferred": row.observed_inferred,
"distType": row.dist_type,
"distResult": row.dist_result,
"distDescription": row.dist_description,
}
disturbance_data.append(disturbance)
data_sheet['disturbanceData'] = disturbance_data
return data_sheet
@app.route('/data/sheet/load2021', methods=['POST'])
def load_sheet_2021():
data_sheet_id = request.form.get("data_sheet_id")
auth_key = request.form.get("auth_key")
if not is_proof_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
SELECT
id,
colony_id,
date,
start_time,
end_time,
nest_visibility,
visibility_comments,
entered_by_name,
proofed_by
FROM data_sheet
WHERE id = %(data_sheet_id)s
""", {
"data_sheet_id": data_sheet_id,
})
if cur.rowcount == 0:
raise ValueError("data_sheet_id doesn't exist")
row = cur.fetchone()
data_sheet: DataSheetData2021 = {
"version": "2021",
"id": row.id,
"proofedBy": row.proofed_by,
"submitterName": row.entered_by_name,
"colonyId": row.colony_id,
"date": row.date.isoformat(),
"startTime": row.start_time.isoformat('minutes'),
"endTime": row.end_time.isoformat('minutes'),
"nestVisibility": row.nest_visibility,
"visibilityComments": row.visibility_comments,
}
# Load the observer data
load_observers_helper(cur, data_sheet, data_sheet_id)
# Load the summary data
load_summary_helper(cur, data_sheet, data_sheet_id)
# Load the HEP nest data
load_hep_nests_helper(cur, data_sheet, data_sheet_id)
# Load the GUTE nest data
load_gute_nests_helper(cur, data_sheet, data_sheet_id)
# DISTURBANCE DATA
cur.execute("""
SELECT
index,
observed_inferred,
dist_type,
dist_description,
dist_result
FROM disturbance_data
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY index
""", {
"data_sheet_id": data_sheet_id,
})
disturbance_data: List[DisturbanceData] = []
for row in cur:
disturbance: DisturbanceData = {
"observedInferred": row.observed_inferred,
"distType": row.dist_type,
"distResult": row.dist_result,
"distDescription": row.dist_description,
}
disturbance_data.append(disturbance)
data_sheet['disturbanceData'] = disturbance_data
return data_sheet
@app.route('/data/sheet/load2020', methods=['POST'])
def load_sheet_2020():
data_sheet_id = request.form.get("data_sheet_id")
auth_key = request.form.get("auth_key")
if not is_proof_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
SELECT
id,
colony_id,
date,
start_time,
end_time,
nest_visibility,
visibility_comments,
significant_changes,
significant_change_notes,
human_disturbance,
human_disturbance_notes,
additional_observations,
entered_by_name,
proofed_by
FROM data_sheet
WHERE id = %(data_sheet_id)s
""", {
"data_sheet_id": data_sheet_id,
})
if cur.rowcount == 0:
raise ValueError("data_sheet_id doesn't exist")
row = cur.fetchone()
data_sheet: DataSheetData2020 = {
"version": "2020",
"id": row.id,
"proofedBy": row.proofed_by,
"submitterName": row.entered_by_name,
"colonyId": row.colony_id,
"date": row.date.isoformat(),
"startTime": row.start_time.isoformat('minutes'),
"endTime": row.end_time.isoformat('minutes'),
"nestVisibility": row.nest_visibility,
"visibilityComments": row.visibility_comments,
"colonySignificantChanges": row.significant_changes,
"colonySignificantChangesNotes": row.significant_change_notes,
"colonyHumanDisturbance": row.human_disturbance,
"colonyHumanDisturbanceNotes": row.human_disturbance_notes,
"colonyAdditionalObservations": row.additional_observations,
}
# Load the observer data
load_observers_helper(cur, data_sheet, data_sheet_id)
# Load the summary data
load_summary_helper(cur, data_sheet, data_sheet_id)
# Load the HEP nest data
load_hep_nests_helper(cur, data_sheet, data_sheet_id)
# Load the GUTE nest data
load_gute_nests_helper(cur, data_sheet, data_sheet_id)
return data_sheet
def proof_observers_helper(cur, observers, data_sheet_id):
# make sure observer records exist for all observers
for observer in observers:
cur.execute("""
INSERT INTO observers (
name
) VALUES (
%(name)s
)
ON CONFLICT (name) DO NOTHING
""", {
"name": observer,
})
# get the ids for all observers
cur.execute("""
SELECT
id
FROM observers
WHERE name IN %(names)s
""", {
"names": tuple(observers),
})
# collect all those ids in to a nice list
observer_ids = []
for o_row in cur:
observer_ids.append(o_row.id)
# create the observer records for this datasheet
for id in observer_ids:
cur.execute("""
INSERT INTO data_sheet_observers (
data_sheet_id,
observer_id
) VALUES (
%(data_sheet_id)s,
%(observer_id)s
)
ON CONFLICT (data_sheet_id, observer_id) DO NOTHING
""", {
"data_sheet_id": data_sheet_id,
"observer_id": id,
})
cur.execute("""
DELETE FROM data_sheet_observers
WHERE
data_sheet_id = %(data_sheet_id)s
AND
observer_id NOT IN %(observer_ids)s
""", {
"data_sheet_id": data_sheet_id,
"observer_ids": tuple(observer_ids),
})
def proof_summary_helper(cur, survey_summary, data_sheet_id):
# Save the survey summary data: these are the overall nest counts.
print("Debugging")
print(survey_summary)
represented_species = []
for species_code, summary in survey_summary.items():
total_nests = int(summary['totalNests'])
total_adults = int(summary['totalAdults'])
total_young = int(summary['totalYoung'])
total_possible_nests = None
if 'totalPossibleNests' in summary:
temp = summary['totalPossibleNests']
if temp:
total_possible_nests = int(temp)
species_type = str(summary['speciesType'])
represented_species.append(species_code)
cur.execute("""
INSERT INTO survey_summary (
data_sheet_id,
species_code,
total_nests,
total_adults,
total_young,
total_possible_nests,
species_type
) VALUES (
%(data_sheet_id)s,
%(species_code)s,
%(total_nests)s,
%(total_adults)s,
%(total_young)s,
%(total_possible_nests)s,
%(species_type)s
)
ON CONFLICT (data_sheet_id, species_code) DO UPDATE SET
total_nests = %(total_nests)s,
total_adults = %(total_adults)s,
total_young = %(total_young)s,
total_possible_nests = %(total_possible_nests)s,
species_type = %(species_type)s
""", {
'data_sheet_id': data_sheet_id,
'species_code': species_code,
'total_nests': total_nests,
'total_adults': total_adults,
'total_young': total_young,
'total_possible_nests': total_possible_nests,
'species_type': species_type,
})
# clean up any species that are no longer represented in the data.
if len(represented_species) > 0:
cur.execute("""
DELETE FROM survey_summary
WHERE
data_sheet_id = %(data_sheet_id)s
AND
species_code NOT IN %(species_codes)s
""", {
'data_sheet_id': data_sheet_id,
'species_codes': tuple(represented_species),
})
def proof_hep_nests_helper(cur, hep_nest_data, data_sheet_id):
hep_nest_numbers = []
for hep_nest in hep_nest_data:
number = int(hep_nest['number'])
is_focal = bool(hep_nest['isFocal'])
species_code = str(hep_nest['speciesCode'])
active = str(hep_nest['active'])
stage = int(hep_nest['stage']) if hep_nest['stage'] is not None else None
adult_count = int(hep_nest['adultCount'])
chick_count = int(hep_nest['chickCount'])
chick_confidence = bool(hep_nest['chickConfidence'])
comments = str(hep_nest['comments'])
hep_nest_numbers.append(number)
cur.execute("""
INSERT INTO hep_nest_data (
data_sheet_id,
nest_number,
focal,
species_code,
active,
stage,
adult_count,
chick_count,
chick_confidence,
comments
) VALUES (
%(data_sheet_id)s,
%(nest_number)s,
%(focal)s,
%(species_code)s,
%(active)s,
%(stage)s,
%(adult_count)s,
%(chick_count)s,
%(chick_confidence)s,
%(comments)s
)
ON CONFLICT (data_sheet_id, nest_number) DO UPDATE SET
focal = %(focal)s,
species_code = %(species_code)s,
active = %(active)s,
stage = %(stage)s,
adult_count = %(adult_count)s,
chick_count = %(chick_count)s,
chick_confidence = %(chick_confidence)s,
comments = %(comments)s
""", {
'data_sheet_id': data_sheet_id,
'nest_number': number,
'focal': is_focal,
'species_code': species_code,
'active': active,
'stage': stage,
'adult_count': adult_count,
'chick_count': chick_count,
'chick_confidence': chick_confidence,
'comments': comments,
})
if len(hep_nest_numbers) > 0:
cur.execute("""
DELETE FROM hep_nest_data
WHERE
data_sheet_id = %(data_sheet_id)s
AND
nest_number NOT IN %(hep_nest_numbers)s
""", {
'data_sheet_id': data_sheet_id,
'hep_nest_numbers': tuple(hep_nest_numbers),
})
def proof_hep_nest_stage_helper(cur, hep_nest_stage_data, data_sheet_id):
hep_stage_indexes = []
for idx, hep_stage_row in enumerate(hep_nest_stage_data):
species_code = str(hep_stage_row['speciesCode'])
stage0 = int(hep_stage_row['stage0'])
stage1 = int(hep_stage_row['stage1'])
stage2 = int(hep_stage_row['stage2'])
stage3 = int(hep_stage_row['stage3'])
stage4 = int(hep_stage_row['stage4'])
stage5 = int(hep_stage_row['stage5'])
unknown_stage = int(hep_stage_row['unknownStage'])
hep_stage_indexes.append(idx)
cur.execute("""
INSERT INTO hep_stage_counts (
data_sheet_id,
index,
species_code,
stage_0,
stage_1,
stage_2,
stage_3,
stage_4,
stage_5,
stage_unknown
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(species_code)s,
%(stage_0)s,
%(stage_1)s,
%(stage_2)s,
%(stage_3)s,
%(stage_4)s,
%(stage_5)s,
%(stage_unknown)s
)
ON CONFLICT (data_sheet_id, index) DO UPDATE SET
species_code = %(species_code)s,
stage_0 = %(stage_0)s,
stage_1 = %(stage_1)s,
stage_2 = %(stage_2)s,
stage_3 = %(stage_3)s,
stage_4 = %(stage_4)s,
stage_5 = %(stage_5)s,
stage_unknown = %(stage_unknown)s
""", {
'data_sheet_id': data_sheet_id,
'species_code': species_code,
'stage_0': stage0,
'stage_1': stage1,
'stage_2': stage2,
'stage_3': stage3,
'stage_4': stage4,
'stage_5': stage5,
'stage_unknown': unknown_stage,
'index': idx,
})
if len(hep_stage_indexes) > 0:
cur.execute("""
DELETE FROM hep_stage_counts
WHERE
data_sheet_id = %(data_sheet_id)s
AND
index NOT IN %(hep_stage_indexes)s
""", {
'data_sheet_id': data_sheet_id,
'hep_stage_indexes': tuple(hep_stage_indexes),
})
else:
cur.execute("""
DELETE FROM hep_stage_counts
WHERE
data_sheet_id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
})
def proof_hep_s4_brood_helper(cur, hep_s4_brood_data, data_sheet_id):
hep_s4_brood_indexes = []
for idx, hep_s4_brood_row in enumerate(hep_s4_brood_data):
species_code = str(hep_s4_brood_row['speciesCode'])
one_chick_nests = int(hep_s4_brood_row['oneChickNests'])
two_chick_nests = int(hep_s4_brood_row['twoChickNests'])
three_chick_nests = int(hep_s4_brood_row['threeChickNests'])
four_chick_nests = int(hep_s4_brood_row['fourChickNests'])
five_chick_nests = int(hep_s4_brood_row['fiveChickNests'])
chicks_no_nest = int(hep_s4_brood_row['chicksNoNest'])
hep_s4_brood_indexes.append(idx)
cur.execute("""
INSERT INTO hep_stage_4_brood_size (
data_sheet_id,
species_code,
one_chick_nests,
two_chick_nests,
three_chick_nests,
four_chick_nests,
five_chick_nests,
chicks_no_nest,
index
) VALUES (
%(data_sheet_id)s,
%(species_code)s,
%(one_chick_nests)s,
%(two_chick_nests)s,
%(three_chick_nests)s,
%(four_chick_nests)s,
%(five_chick_nests)s,
%(chicks_no_nest)s,
%(index)s
)
ON CONFLICT (data_sheet_id, index) DO UPDATE SET
species_code = %(species_code)s,
one_chick_nests = %(one_chick_nests)s,
two_chick_nests = %(two_chick_nests)s,
three_chick_nests = %(three_chick_nests)s,
four_chick_nests = %(four_chick_nests)s,
five_chick_nests = %(five_chick_nests)s,
chicks_no_nest = %(chicks_no_nest)s
""", {
'data_sheet_id': data_sheet_id,
'species_code': species_code,
'one_chick_nests': one_chick_nests,
'two_chick_nests': two_chick_nests,
'three_chick_nests': three_chick_nests,
'four_chick_nests': four_chick_nests,
'five_chick_nests': five_chick_nests,
'chicks_no_nest': chicks_no_nest,
'index': idx,
})
if len(hep_s4_brood_indexes) > 0:
cur.execute("""
DELETE FROM hep_stage_4_brood_size
WHERE
data_sheet_id = %(data_sheet_id)s
AND
index NOT IN %(hep_s4_brood_indexes)s
""", {
'data_sheet_id': data_sheet_id,
'hep_s4_brood_indexes': tuple(hep_s4_brood_indexes),
})
else:
cur.execute("""
DELETE FROM hep_stage_4_brood_size
WHERE
data_sheet_id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
})
def proof_gute_nests_helper(cur, gute_nest_data, data_sheet_id):
gute_indexes = []
for idx, gute_nest in enumerate(gute_nest_data):
subcolony = None
if 'subcolony' in gute_nest_data:
subcolony = str(gute_nest_data['gute_nest_data'])
species_code = str(gute_nest['speciesCode'])
total_adults = int(gute_nest['totalAdults'])
if gute_nest['stage0Adults'] is None:
stage_0_adults = None
else:
stage_0_adults = int(gute_nest['stage0Adults'])
stage_1_nests = int(gute_nest['stage1Nests'])
stage_2_chicks = int(gute_nest['stage2Chicks'])
stage_3_chicks = int(gute_nest['stage3Chicks'])
stage_4_chicks = int(gute_nest['stage4Chicks'])
comments = str(gute_nest['comments'])
location = None
if 'location' in gute_nest_data:
location = str(gute_nest['location'])
included_on_map = None
if 'included_on_map' in gute_nest_data:
included_on_map = str(gute_nest['includedOnMap'])
gute_indexes.append(idx)
cur.execute("""
INSERT INTO gute_nest_data (
data_sheet_id,
index,
subcolony,
species_code,
total_adults,
stage_0_adults,
stage_1_nests,
stage_2_chicks,
stage_3_chicks,
stage_4_chicks,
comments,
location,
included_on_map
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(subcolony)s,
%(species_code)s,
%(total_adults)s,
%(stage_0_adults)s,
%(stage_1_nests)s,
%(stage_2_chicks)s,
%(stage_3_chicks)s,
%(stage_4_chicks)s,
%(comments)s,
%(location)s,
%(included_on_map)s
)
ON CONFLICT (data_sheet_id, index) DO UPDATE SET
subcolony = %(subcolony)s,
species_code = %(species_code)s,
total_adults = %(total_adults)s,
stage_0_adults = %(stage_0_adults)s,
stage_1_nests = %(stage_1_nests)s,
stage_2_chicks = %(stage_2_chicks)s,
stage_3_chicks = %(stage_3_chicks)s,
stage_4_chicks = %(stage_4_chicks)s,
comments = %(comments)s,
location = %(location)s,
included_on_map = %(included_on_map)s
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'subcolony': subcolony,
'species_code': species_code,
'total_adults': total_adults,
'stage_0_adults': stage_0_adults,
'stage_1_nests': stage_1_nests,
'stage_2_chicks': stage_2_chicks,
'stage_3_chicks': stage_3_chicks,
'stage_4_chicks': stage_4_chicks,
'comments': comments,
'location': location,
'included_on_map': included_on_map,
})
if len(gute_indexes) > 0:
cur.execute("""
DELETE FROM gute_nest_data
WHERE
data_sheet_id = %(data_sheet_id)s
AND
index NOT IN %(gute_indexes)s
""", {
'data_sheet_id': data_sheet_id,
'gute_indexes': tuple(gute_indexes),
})
else:
cur.execute("""
DELETE FROM gute_nest_data
WHERE
data_sheet_id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
})
@app.route('/data/sheet/proof', methods=['POST'])
def proof_sheet():
data: DataSheetData = request.get_json()
data_sheet_id = str(data['id'])
auth_key = str(data['authKey'])
proofed_by = str(data['proofedBy'])
entered_by_name = str(data['submitterName'])
colony_id = int(data['colonyId'])
date = str(data['date'])
start_time = str(data['startTime'])
end_time = str(data['endTime'])
observers = list(data['observers'])
nest_visibility = str(data['nestVisibility'])
visibility_comments = str(data['visibilityComments'])
survey_summary = data['surveySummary']
hep_nest_stage_data = data['hepNestStageData']
hep_s4_brood_data = data['hepS4BroodData']
gute_nest_data = data['guteNestData']
disturbance_data = data['disturbanceData']
additional_observations = data['colonyAdditionalObservations']
if not is_proof_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
UPDATE data_sheet
SET
proofed_by = %(proofed_by)s,
proofed_timestamp = now(),
colony_id = %(colony_id)s,
date = %(date)s,
start_time = %(start_time)s,
end_time = %(end_time)s,
nest_visibility = %(nest_visibility)s,
visibility_comments = %(visibility_comments)s,
entered_by_name = %(entered_by_name)s,
additional_observations = %(additional_observations)s
WHERE
id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
'proofed_by': proofed_by,
'entered_by_name': entered_by_name,
'colony_id': colony_id,
'date': date,
'start_time': start_time,
'end_time': end_time,
'nest_visibility': nest_visibility,
'visibility_comments': visibility_comments,
'additional_observations': additional_observations,
})
# Save the observer data
proof_observers_helper(cur, observers, data_sheet_id)
# Save the survey summary data
proof_summary_helper(cur, survey_summary, data_sheet_id)
# Save the HEP nest stage data
proof_hep_nest_stage_helper(cur, hep_nest_stage_data, data_sheet_id)
# Save the HEP s4 brood info
proof_hep_s4_brood_helper(cur, hep_s4_brood_data, data_sheet_id)
# Save the GUTE subcolony data
proof_gute_nests_helper(cur, gute_nest_data, data_sheet_id)
# Save the disturbance data
disturbance_indexes = []
for idx, disturbance in enumerate(disturbance_data):
observed_inferred = str(disturbance['observedInferred'])
dist_type = str(disturbance['distType'])
dist_result = str(disturbance['distResult'])
dist_description = str(disturbance['distDescription'])
disturbance_indexes.append(idx)
cur.execute("""
INSERT INTO disturbance_data (
data_sheet_id,
index,
observed_inferred,
dist_type,
dist_result,
dist_description
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(observed_inferred)s,
%(dist_type)s,
%(dist_result)s,
%(dist_description)s
)
ON CONFLICT (data_sheet_id, index) DO UPDATE SET
observed_inferred = %(observed_inferred)s,
dist_type = %(dist_type)s,
dist_result = %(dist_result)s,
dist_description = %(dist_description)s
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'observed_inferred': observed_inferred,
'dist_type': dist_type,
'dist_result': dist_result,
'dist_description': dist_description,
})
if len(disturbance_indexes) > 0:
cur.execute("""
DELETE FROM disturbance_data
WHERE
data_sheet_id = %(data_sheet_id)s
AND
index NOT IN %(disturbance_indexes)s
""", {
'data_sheet_id': data_sheet_id,
'disturbance_indexes': tuple(disturbance_indexes),
})
else:
cur.execute("""
DELETE FROM disturbance_data
WHERE
data_sheet_id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
})
return jsonify(data)
@app.route('/data/sheet/proof2021', methods=['POST'])
def proof_sheet_2021():
data: DataSheetData2021 = request.get_json()
data_sheet_id = str(data['id'])
auth_key = str(data['authKey'])
proofed_by = str(data['proofedBy'])
entered_by_name = str(data['submitterName'])
colony_id = int(data['colonyId'])
date = str(data['date'])
start_time = str(data['startTime'])
end_time = str(data['endTime'])
observers = list(data['observers'])
nest_visibility = str(data['nestVisibility'])
visibility_comments = str(data['visibilityComments'])
survey_summary = data['surveySummary']
hep_nest_data = data['hepNestData']
gute_nest_data = data['guteNestData']
disturbance_data = data['disturbanceData']
if not is_proof_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
UPDATE data_sheet
SET
proofed_by = %(proofed_by)s,
proofed_timestamp = now(),
colony_id = %(colony_id)s,
date = %(date)s,
start_time = %(start_time)s,
end_time = %(end_time)s,
nest_visibility = %(nest_visibility)s,
visibility_comments = %(visibility_comments)s,
entered_by_name = %(entered_by_name)s
WHERE
id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
'proofed_by': proofed_by,
'entered_by_name': entered_by_name,
'colony_id': colony_id,
'date': date,
'start_time': start_time,
'end_time': end_time,
'nest_visibility': nest_visibility,
'visibility_comments': visibility_comments,
})
# Save the observer data
proof_observers_helper(cur, observers, data_sheet_id)
# Save the survey summary data
proof_summary_helper(cur, survey_summary, data_sheet_id)
# Save the HEP nest data
proof_hep_nests_helper(cur, hep_nest_data, data_sheet_id)
# Save the GUTE subcolony data
proof_gute_nests_helper(cur, gute_nest_data, data_sheet_id)
# Save the disturbance data
disturbance_indexes = []
for idx, disturbance in enumerate(disturbance_data):
observed_inferred = str(disturbance['observedInferred'])
dist_type = str(disturbance['distType'])
dist_result = str(disturbance['distResult'])
dist_description = str(disturbance['distDescription'])
disturbance_indexes.append(idx)
cur.execute("""
INSERT INTO disturbance_data (
data_sheet_id,
index,
observed_inferred,
dist_type,
dist_result,
dist_description
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(observed_inferred)s,
%(dist_type)s,
%(dist_result)s,
%(dist_description)s
)
ON CONFLICT (data_sheet_id, index) DO UPDATE SET
observed_inferred = %(observed_inferred)s,
dist_type = %(dist_type)s,
dist_result = %(dist_result)s,
dist_description = %(dist_description)s
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'observed_inferred': observed_inferred,
'dist_type': dist_type,
'dist_result': dist_result,
'dist_description': dist_description,
})
if len(disturbance_indexes) > 0:
cur.execute("""
DELETE FROM disturbance_data
WHERE
data_sheet_id = %(data_sheet_id)s
AND
index NOT IN %(disturbance_indexes)s
""", {
'data_sheet_id': data_sheet_id,
'disturbance_indexes': tuple(disturbance_indexes),
})
return jsonify(data)
@app.route('/data/sheet/proof2020', methods=['POST'])
def proof_sheet_2020():
data: DataSheetData2020 = request.get_json()
data_sheet_id = str(data['id'])
auth_key = str(data['authKey'])
proofed_by = str(data['proofedBy'])
entered_by_name = str(data['submitterName'])
colony_id = int(data['colonyId'])
date = str(data['date'])
start_time = str(data['startTime'])
end_time = str(data['endTime'])
observers = list(data['observers'])
nest_visibility = str(data['nestVisibility'])
visibility_comments = str(data['visibilityComments'])
survey_summary = data['surveySummary']
significant_changes = bool(data['colonySignificantChanges'])
significant_changes_notes = str(data['colonySignificantChangesNotes'])
human_disturbance = bool(data['colonyHumanDisturbance'])
human_disturbance_notes = str(data['colonyHumanDisturbanceNotes'])
additional_observations = str(data['colonyAdditionalObservations'])
hep_nest_data = data['hepNestData']
gute_nest_data = data['guteNestData']
if not is_proof_key_valid(auth_key):
abort(401)
with get_db() as db:
with db.cursor() as cur:
cur.execute("""
UPDATE data_sheet
SET
proofed_by = %(proofed_by)s,
proofed_timestamp = now(),
colony_id = %(colony_id)s,
date = %(date)s,
start_time = %(start_time)s,
end_time = %(end_time)s,
nest_visibility = %(nest_visibility)s,
visibility_comments = %(visibility_comments)s,
significant_changes = %(significant_changes)s,
significant_change_notes = %(significant_change_notes)s,
human_disturbance = %(human_disturbance)s,
human_disturbance_notes = %(human_disturbance_notes)s,
additional_observations = %(additional_observations)s,
entered_by_name = %(entered_by_name)s
WHERE
id = %(data_sheet_id)s
""", {
'data_sheet_id': data_sheet_id,
'proofed_by': proofed_by,
'entered_by_name': entered_by_name,
'colony_id': colony_id,
'date': date,
'start_time': start_time,
'end_time': end_time,
'nest_visibility': nest_visibility,
'visibility_comments': visibility_comments,
'significant_changes': significant_changes,
'significant_change_notes': significant_changes_notes,
'human_disturbance': human_disturbance,
'human_disturbance_notes': human_disturbance_notes,
'additional_observations': additional_observations,
})
# Save the observer data
proof_observers_helper(cur, observers, data_sheet_id)
# Save the survey summary data
proof_summary_helper(cur, survey_summary, data_sheet_id)
# Save the HEP nest data
proof_hep_nests_helper(cur, hep_nest_data, data_sheet_id)
# Save the GUTE subcolony data
proof_gute_nests_helper(cur, gute_nest_data, data_sheet_id)
return jsonify(data)