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.
 
 
 
 

547 lines
18 KiB

from typing import TypedDict, Dict, List
from flask import Flask, redirect, render_template, current_app, g, jsonify, request
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
from flask_wtf.csrf import CSRFProtect # type: ignore
DATABASE_URL = os.environ.get("DATABASE_URL")
SECRET_KEY = os.environ.get("SECRET_KEY")
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)
else:
csrf = CSRFProtect(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):
with get_db() as db:
with db.cursor() as cur:
try:
cur.execute("""
SELECT id FROM create_keys WHERE id = %(id)s;
""", {
'id': key,
})
if cur.rowcount == 1:
return render_template('submit.html')
except Exception as e:
print(e)
return render_template('404.html')
@app.route('/data/colonies')
def get_colonies():
with get_db() as db:
with db.cursor() as cur:
cur.execute("""SELECT id, name FROM colony""")
rows = cur.fetchall()
return {
"colonies": [{"name":i.name, "id":i.id} for i in rows],
}
@app.route('/data/observers')
def get_observers():
return {
"observers": [
"Gabbie Burns",
"Deanna de Castro",
],
}
@app.route('/data/species')
def get_species():
return {
}
class SurveySummary(TypedDict, total=False):
totalNests: int
totalAdults: int
totalYoung: int
totalPossibleNests: int
class HepNestData(TypedDict, total=False):
number: int
isFocal: bool
speciesCode: str
active: str
stage: int
adultCount: int
chickCount: int
chickConfidence: bool
comments: str
class GuteNestData(TypedDict, total=False):
subcolony: str
speciesCode: str
totalAdults: int
stage0Adults: int
stage1Nests: int
stage2Chicks: int
stage3Chicks: int
stage4Chicks: int
comments: str
class GuteNestLocationData(TypedDict, total=False):
subcolony: str
location: str
includedOnMap: bool
class DataSheetData(TypedDict, total=False):
id: str
submitterName: str
colonyId: int
date: str
startTime: str
endTime: 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]
guteNestLocationData: List[GuteNestLocationData]
@app.route('/data/sheet/save', methods=['POST'])
def save_sheet():
data: DataSheetData = request.get_json()
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'])
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']
gute_nest_location_data = data['guteNestLocationData']
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 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 = int(summary['totalPossibleNests'])
cur.execute("""
INSERT INTO survey_summary (
data_sheet_id,
species_code,
total_nests,
total_adults,
total_young,
total_possible_nests
) VALUES (
%(data_sheet_id)s,
%(species_code)s,
%(total_nests)s,
%(total_adults)s,
%(total_young)s,
%(total_possible_nests)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,
})
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'])
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,
})
for idx, gute_nest in enumerate(gute_nest_data):
subcolony = str(gute_nest['subcolony'])
species_code = str(gute_nest['speciesCode'])
total_adults = int(gute_nest['totalAdults'])
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'])
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
) 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
)
""", {
'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,
})
for idx, gute_nest_location in enumerate(gute_nest_location_data):
subcolony = str(gute_nest_location['subcolony'])
location = str(gute_nest_location['location'])
included_on_map = str(gute_nest_location['includedOnMap'])
cur.execute("""
INSERT INTO gute_nest_location_data (
data_sheet_id,
index,
subcolony,
location,
included_on_map
) VALUES (
%(data_sheet_id)s,
%(index)s,
%(subcolony)s,
%(location)s,
%(included_on_map)s
)
""", {
'data_sheet_id': data_sheet_id,
'index': idx,
'subcolony': subcolony,
'location': location,
'included_on_map': included_on_map,
})
return jsonify(data)
@app.route('/data/sheet/load', methods=['POST'])
def load_sheet():
data_sheet_id = request.form.get("data_sheet_id")
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
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 = {
"id": row.id,
"colonyId": row.colony_id,
"date": row.date, # TODO: convert to just a date (w/o time)
# "startTime": row.start_time,
# "endTime": row.end_time,
"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,
}
# SURVEY SUMMARY DATA
cur.execute("""
SELECT
id,
species_code,
total_nests,
total_adults,
total_young,
total_possible_nests
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,
}
survey_summary[row.species_code] = summary
data_sheet['surveySummary'] = survey_summary
# HEP NEST DATA
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
# GUTE NEST DATA
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
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,
}
gute_nest_data.append(gute)
data_sheet['guteNestData'] = gute_nest_data
# Gute Nest Location Data
gute_nest_location_data: List[GuteNestLocationData] = []
cur.execute("""
SELECT
index,
subcolony,
location,
included_on_map
FROM gute_nest_location_data
WHERE data_sheet_id = %(data_sheet_id)s
ORDER BY index
""", {
"data_sheet_id": data_sheet_id,
})
for row in cur:
gute_nest_location: GuteNestLocationData = {
"subcolony": row.subcolony,
"location": row.location,
"includedOnMap": row.included_on_map,
}
gute_nest_location_data.append(gute_nest_location)
data_sheet['guteNestLocationData'] = gute_nest_location_data
return data_sheet