212 lines
10 KiB
Python
212 lines
10 KiB
Python
from wtforms import SubmitField, StringField, HiddenField, validators, Form
|
|
from flask_wtf import FlaskForm
|
|
from flask import request, render_template, redirect, url_for
|
|
from main import db, app, ma
|
|
from settings import Settings, AIModel
|
|
from sqlalchemy import Sequence
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from status import st, Status
|
|
from flask_login import login_required, current_user
|
|
from werkzeug.utils import secure_filename
|
|
from shared import GenFace, GenThumb
|
|
from face import FaceRefimgLink
|
|
import os
|
|
import json
|
|
|
|
|
|
# pylint: disable=no-member
|
|
|
|
################################################################################
|
|
# Class describing Refimg in DB via sqlalchemy
|
|
# fname: original file name of refimg
|
|
# face: actual binary of numpy data for this refimg's face (always only 1)
|
|
# orig*: width/height of original image, because when we show in person, it get scaled
|
|
# face_locn: location of ace - we need to know to draw green box around face locn (with orig* above)
|
|
# thumbnail: image data of actual img. once we load refimg, we only store this data, not the orig file
|
|
# model_used: which AI model (cnn or hog) used to create face
|
|
# person: read-only convenience field not in DB, just used in html
|
|
################################################################################
|
|
class Refimg(db.Model):
|
|
id = db.Column(db.Integer, db.Sequence('refimg_id_seq'), primary_key=True )
|
|
fname = db.Column(db.String(256), unique=True, nullable=False)
|
|
face = db.Column(db.LargeBinary, unique=True, nullable=False)
|
|
orig_w = db.Column(db.Integer)
|
|
orig_h = db.Column(db.Integer)
|
|
face_locn = db.Column(db.String)
|
|
thumbnail = db.Column(db.String, unique=True, nullable=False)
|
|
created_on = db.Column(db.Float)
|
|
model_used = db.Column(db.Integer, db.ForeignKey("ai_model.id") )
|
|
person = db.relationship( 'Person', secondary="person_refimg_link", uselist=False, viewonly=True )
|
|
|
|
def __repr__(self):
|
|
return "<id: {}, fname: {}>".format(self.id, self.fname )
|
|
|
|
################################################################################
|
|
# Class describing Person to Refimg link in DB via sqlalchemy
|
|
################################################################################
|
|
class PersonRefimgLink(db.Model):
|
|
__tablename__ = "person_refimg_link"
|
|
person_id = db.Column(db.Integer, db.ForeignKey('person.id'), unique=True, nullable=False, primary_key=True)
|
|
refimg_id = db.Column(db.Integer, db.ForeignKey('refimg.id'), unique=True, nullable=False, primary_key=True)
|
|
|
|
def __repr__(self):
|
|
return f"<person_id: {self.person_id}, refimg_id: {self.refimg_id}>"
|
|
|
|
|
|
################################################################################
|
|
# Class describing Person in DB via sqlalchemy
|
|
################################################################################
|
|
class Person(db.Model):
|
|
id = db.Column(db.Integer, db.Sequence('person_id_seq'), primary_key=True )
|
|
tag = db.Column(db.String(48), unique=False, nullable=False)
|
|
surname = db.Column(db.String(48), unique=False, nullable=False)
|
|
firstname = db.Column(db.String(48), unique=False, nullable=False)
|
|
refimg = db.relationship('Refimg', secondary=PersonRefimgLink.__table__, order_by=Refimg.id)
|
|
|
|
def __repr__(self):
|
|
return "<tag: {}, firstname: {}, surname: {}, refimg: {}>".format(self.tag,self.firstname, self.surname, self.refimg)
|
|
|
|
################################################################################
|
|
# Helper class that inherits a .dump() method to turn class Person into json / useful in jinja2
|
|
################################################################################
|
|
class PersonSchema(ma.SQLAlchemyAutoSchema):
|
|
class Meta:
|
|
model = Person
|
|
ordered = True
|
|
|
|
################################################################################
|
|
# Helper class that defines a form for person, used to make html <form>, with field validation (via wtforms)
|
|
################################################################################
|
|
class PersonForm(FlaskForm):
|
|
id = HiddenField()
|
|
tag = StringField('Tag (searchable name):', [validators.DataRequired()])
|
|
firstname = StringField('FirstName(s):', [validators.DataRequired()])
|
|
surname = StringField('Surname:', [validators.DataRequired()])
|
|
save = SubmitField('Save' )
|
|
delete = SubmitField('Delete' )
|
|
|
|
################################################################################
|
|
# Routes for person data
|
|
#
|
|
# /persons -> GET only -> prints out list of all persons
|
|
################################################################################
|
|
@app.route("/persons", methods=["GET"])
|
|
@login_required
|
|
def persons():
|
|
persons = Person.query.all()
|
|
return render_template("persons.html", persons=persons)
|
|
|
|
|
|
################################################################################
|
|
# /person -> GET/POST -> creates a new person and shows new creation
|
|
################################################################################
|
|
@app.route("/person", methods=["GET", "POST"])
|
|
@login_required
|
|
def new_person():
|
|
form = PersonForm(request.form)
|
|
page_title='Create new Person'
|
|
|
|
if 'surname' not in request.form:
|
|
return render_template("person.html", person=None, form=form, page_title=page_title )
|
|
else:
|
|
person = Person( tag=request.form["tag"], surname=request.form["surname"], firstname=request.form["firstname"] )
|
|
try:
|
|
db.session.add(person)
|
|
db.session.commit()
|
|
st.SetMessage( "Created new Person ({})".format(person.tag) )
|
|
return redirect( f'/person/{person.id}' )
|
|
except SQLAlchemyError as e:
|
|
st.SetAlert( "danger" )
|
|
st.SetMessage( "<b>Failed to add Person:</b> {}".format(e.orig) )
|
|
return render_template("person.html", person=person, form=form, page_title = page_title)
|
|
|
|
################################################################################
|
|
# /person/<id> -> GET/POST(save or delete) -> shows/edits/delets a single person
|
|
################################################################################
|
|
@app.route("/person/<id>", methods=["GET", "POST"])
|
|
@login_required
|
|
def person(id):
|
|
form = PersonForm(request.form)
|
|
page_title='Edit Person'
|
|
|
|
if request.method == 'POST':
|
|
try:
|
|
person = Person.query.get(id)
|
|
if 'delete' in request.form:
|
|
st.SetMessage("Successfully deleted Person: ({})".format( person.tag ) )
|
|
# do the linkage tables by hand
|
|
db.session.execute( f"delete from face_refimg_link frl where refimg_id in ( select refimg_id from person_refimg_link where person_id = {id} )" )
|
|
db.session.execute( f"delete from person_refimg_link where person_id = {id}" )
|
|
Person.query.filter(Person.id==id).delete()
|
|
db.session.commit()
|
|
return redirect( f'/persons' )
|
|
elif request.form and form.validate():
|
|
new_refs=[]
|
|
for ref_img in person.refimg:
|
|
if "ref-img-id-{}".format(ref_img.id) in request.form:
|
|
new_refs.append(ref_img)
|
|
if new_refs != person.refimg:
|
|
deld = list(set(person.refimg) - set(new_refs));
|
|
person.refimg = new_refs
|
|
Refimg.query.filter(Refimg.id==deld[0].id).delete()
|
|
st.SetMessage( f"Successfully Updated Person: removed reference image {deld[0].fname}" )
|
|
else:
|
|
st.SetMessage("Successfully Updated Person: (From: {}, {}, {})".format(person.tag, person.firstname, person.surname) )
|
|
person.tag = request.form['tag']
|
|
person.surname = request.form['surname']
|
|
person.firstname = request.form['firstname']
|
|
st.AppendMessage(" To: ({}, {}, {})".format(person.tag, person.firstname, person.surname) )
|
|
db.session.add(person)
|
|
db.session.commit()
|
|
return redirect( f'/person/{person.id}' )
|
|
except SQLAlchemyError as e:
|
|
st.SetAlert( "danger" )
|
|
st.SetMessage( "<b>Failed to modify Person:</b> {}".format(e) )
|
|
return render_template("person.html", form=form, page_title=page_title)
|
|
else:
|
|
person = Person.query.get(id)
|
|
for r in person.refimg:
|
|
r.face_locn=json.loads(r.face_locn)
|
|
form = PersonForm(request.values, obj=person)
|
|
return render_template("person.html", person=person, form=form, page_title = page_title)
|
|
|
|
################################################################################
|
|
# /add_refimg -> POST(add new refimg to a person)
|
|
################################################################################
|
|
@app.route("/add_refimg", methods=["POST"])
|
|
@login_required
|
|
def add_refimg():
|
|
# now save into the DB
|
|
person = Person.query.get(request.form['person_id']);
|
|
if not person:
|
|
raise Exception("could not find person to add reference image too!")
|
|
f=request.files['refimg_file']
|
|
refimg = Refimg( fname=f.filename )
|
|
try:
|
|
# save the actual uploaded image to reference_images/
|
|
fname=secure_filename(f.filename)
|
|
if fname == "":
|
|
raise Exception("invalid filename")
|
|
|
|
fname = f"/tmp/{fname}"
|
|
f.save( fname )
|
|
refimg.thumbnail, refimg.orig_w, refimg.orig_h = GenThumb( fname )
|
|
settings = Settings.query.first()
|
|
model=AIModel.query.get(settings.default_refimg_model)
|
|
refimg.face, face_locn = GenFace( fname, model=model.name )
|
|
refimg.face_locn = json.dumps(face_locn)
|
|
refimg.model_used = settings.default_refimg_model
|
|
os.remove(fname)
|
|
person.refimg.append(refimg)
|
|
db.session.add(person)
|
|
db.session.add(refimg)
|
|
db.session.commit()
|
|
st.SetMessage( f"Associated new Refimg ({refimg.fname}) with person: {person.tag}" )
|
|
except SQLAlchemyError as e:
|
|
st.SetAlert( "danger" )
|
|
st.SetMessage( f"<b>Failed to add Refimg:</b> {e.orig}" )
|
|
except Exception as e:
|
|
st.SetAlert( "danger" )
|
|
st.SetMessage( f"<b>Failed to modify Refimg:</b> {e}" )
|
|
return redirect( url_for( 'person', id=person.id) )
|