476 lines
21 KiB
Python
476 lines
21 KiB
Python
from wtforms import SubmitField, StringField, HiddenField, validators, Form
|
|
from flask_wtf import FlaskForm
|
|
from flask import request, render_template, redirect, url_for, make_response, jsonify
|
|
from main import db, app, ma
|
|
from settings import Settings, AIModel
|
|
from sqlalchemy import Sequence, func, select
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import joinedload
|
|
from flask_login import login_required, current_user
|
|
from werkzeug.utils import secure_filename
|
|
from shared import GenFace, GenThumb, PA
|
|
from face import Face, FaceRefimgLink, FaceOverrideType, FaceNoMatchOverride, FaceForceMatchOverride
|
|
from path import Path, PathType
|
|
from job import JobExtra, NewJob, SetFELog
|
|
|
|
import os
|
|
import time
|
|
from PIL import Image
|
|
import base64
|
|
from io import BytesIO
|
|
import os.path
|
|
|
|
|
|
# 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
|
|
# 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(PA,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_top = db.Column(db.Integer)
|
|
face_right = db.Column(db.Integer)
|
|
face_bottom = db.Column(db.Integer)
|
|
face_left = db.Column(db.Integer)
|
|
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 )
|
|
|
|
|
|
################################################################################
|
|
# Class describing Person to Refimg link in DB via sqlalchemy
|
|
################################################################################
|
|
class PersonRefimgLink(PA,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)
|
|
|
|
|
|
################################################################################
|
|
# Class describing Person in DB via sqlalchemy
|
|
################################################################################
|
|
class Person(PA,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)
|
|
|
|
################################################################################
|
|
# 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' )
|
|
|
|
################################################################################
|
|
# Helper functions
|
|
# AddRefimgToPerson( filename, person )
|
|
################################################################################
|
|
def AddRefimgToPerson( filename, person ):
|
|
refimg = Refimg( fname=os.path.basename( filename ) )
|
|
try:
|
|
#False == dont autorotate, its not needed on this image
|
|
refimg.thumbnail, refimg.orig_w, refimg.orig_h = GenThumb( filename, False )
|
|
settings = Settings.query.first()
|
|
model=db.session.get(AIModel,settings.default_refimg_model)
|
|
refimg.face, face_locn = GenFace( filename, model=model.name )
|
|
try:
|
|
os.remove(filename)
|
|
except Exception as e:
|
|
# can fail "silently" here, if the face_locn worked, great, its only
|
|
# a tmp file in /tmp - if not, the next if will send a msg to the front-end
|
|
SetFELog( message=f"Failed to delete tmp file for refimg addition: {e}", level="danger", persistent=True, cant_close=True )
|
|
|
|
if not face_locn:
|
|
SetFELog( f"<b>Failed to find face in Refimg:</b>", "danger" )
|
|
raise Exception("Could not find face in uploaded reference image" )
|
|
refimg.face_top = face_locn[0]
|
|
refimg.face_right = face_locn[1]
|
|
refimg.face_bottom = face_locn[2]
|
|
refimg.face_left = face_locn[3]
|
|
refimg.model_used = settings.default_refimg_model
|
|
refimg.created_on = time.time()
|
|
person.refimg.append(refimg)
|
|
db.session.add(person)
|
|
db.session.add(refimg)
|
|
db.session.commit()
|
|
SetFELog( f"Associated new Refimg ({refimg.fname}) with person: {person.tag}" )
|
|
except SQLAlchemyError as e:
|
|
SetFELog( f"<b>Failed to add Refimg:</b> {e.orig}", "danger" )
|
|
except Exception as e:
|
|
SetFELog( f"<b>Failed to modify Refimg:</b> {e}", "danger" )
|
|
return refimg
|
|
|
|
################################################################################
|
|
# TempRefimgFile: helper function that takes data POST'd (from dialog box to
|
|
# add face to new/existing person). Converts data into a jpg file to be used by
|
|
# wrapper funcs to AI / refimg <-> person. filename will be <tag>.jpg
|
|
################################################################################
|
|
def TempRefimgFile( data, tag ):
|
|
# undo the munging sending via http has done
|
|
data=data.replace(' ', '+' )
|
|
|
|
# convert b64 encoded to a temp file to process...
|
|
bytes_decoded = base64.b64decode(data)
|
|
img = Image.open(BytesIO(bytes_decoded))
|
|
out_jpg = img.convert("RGB")
|
|
# save file to /tmp/<tag>.jpg
|
|
fname="/tmp/" + tag + ".jpg"
|
|
out_jpg.save(fname)
|
|
return fname
|
|
|
|
################################################################################
|
|
# 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.order_by(Person.tag).all()
|
|
for p in persons:
|
|
if not p.refimg:
|
|
continue
|
|
# okay see how many faces match this person
|
|
stat=Person.query.filter(Person.id==p.id).join(PersonRefimgLink).join(Refimg).join(FaceRefimgLink).with_entities( func.count( Person.tag ) ).group_by( Person.tag ).first()
|
|
if stat:
|
|
p.num_matches=stat[0]
|
|
|
|
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 request.method=='POST':
|
|
person = Person( tag=request.form["tag"], surname=request.form["surname"], firstname=request.form["firstname"] )
|
|
try:
|
|
db.session.add(person)
|
|
db.session.commit()
|
|
SetFELog( f"Created new Person ({person.tag})" )
|
|
return redirect( url_for( 'person', id=person.id) )
|
|
except SQLAlchemyError as e:
|
|
SetFELog( f"<b>Failed to add Person:</b> {e.orig}", "danger" )
|
|
return redirect( url_for( '/persons') )
|
|
else:
|
|
return render_template("person.html", person=None, form=form, page_title=page_title )
|
|
|
|
@app.route("/match_with_create_person", methods=["POST"])
|
|
@login_required
|
|
def match_with_create_person():
|
|
p = Person( tag=request.form["tag"], surname=request.form["surname"], firstname=request.form["firstname"] )
|
|
# add this fname (of temp refimg) to person
|
|
fname=TempRefimgFile( request.form['refimg_data'], p.tag )
|
|
r=AddRefimgToPerson( fname, p )
|
|
SetFELog( f"Created person: {p.tag}" )
|
|
refimg_schema=RefimgSchema(many=False)
|
|
r_data=refimg_schema.dump(r)
|
|
|
|
return make_response( jsonify( refimg=r_data, who=p.tag, distance='0.0' ) )
|
|
|
|
################################################################################
|
|
# /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 = db.session.get(Person, id)
|
|
if 'delete' in request.form:
|
|
SetFELog( f"Successfully deleted Person: ({person.tag})" )
|
|
|
|
# delete refimgs that are associated with this person (one-by-one), not super efficient
|
|
# but simple/clearer than cascades for now
|
|
for ref in person.refimg:
|
|
FaceRefimgLink.query.filter(FaceRefimgLink.refimg_id==ref.id).delete()
|
|
PersonRefimgLink.query.filter(PersonRefimgLink.refimg_id==ref.id).delete()
|
|
Refimg.query.filter(Refimg.id==ref.id).delete()
|
|
|
|
# now can delete the person entry with no foreign key data left
|
|
Person.query.filter(Person.id==id).delete()
|
|
db.session.commit()
|
|
return redirect( url_for( '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
|
|
# delete the "match" between a face found in a file and this ref img
|
|
FaceRefimgLink.query.filter(FaceRefimgLink.refimg_id==deld[0].id).delete()
|
|
Refimg.query.filter(Refimg.id==deld[0].id).delete()
|
|
SetFELog( f"Successfully Updated Person: removed reference image {deld[0].fname}" )
|
|
else:
|
|
s=f"Successfully Updated Person: (From: {person.tag}, {person.firstname}, {person.surname})"
|
|
person.tag = request.form['tag']
|
|
person.surname = request.form['surname']
|
|
person.firstname = request.form['firstname']
|
|
SetFELog( f"{s} To: ({person.tag}, {person.firstname}, {person.surname})" )
|
|
db.session.add(person)
|
|
db.session.commit()
|
|
return redirect( url_for( 'person', id=person.id) )
|
|
except SQLAlchemyError as e:
|
|
SetFELog( f"<b>Failed to modify Person:</b> {e}", "danger" )
|
|
return redirect( url_for( 'persons' ) )
|
|
else:
|
|
person = db.session.get(Person, id)
|
|
if not person:
|
|
SetFELog( f"No such person with id: {id}", "danger" )
|
|
return redirect("/")
|
|
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():
|
|
try:
|
|
# 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!")
|
|
|
|
# save the actual uploaded image to reference_images/
|
|
f=request.files['refimg_file']
|
|
fname=secure_filename(f.filename)
|
|
if fname == "":
|
|
raise Exception("invalid filename")
|
|
|
|
fname = f"/tmp/{fname}"
|
|
f.save( fname )
|
|
except Exception as e:
|
|
SetFELog( f"<b>Failed to load reference image:</b> {e}", "danger" )
|
|
|
|
r=AddRefimgToPerson( fname, person )
|
|
return redirect( url_for( 'person', id=person.id) )
|
|
|
|
################################################################################
|
|
# /find_persons/<who> -> POST (an arbitrary string to find a person (via
|
|
# name/tag match)
|
|
################################################################################
|
|
@app.route("/find_persons/<who>", methods=["POST"])
|
|
@login_required
|
|
def find_persons(who):
|
|
resp={}
|
|
people = Person.query.filter( Person.tag.ilike(f'%{who}%') | Person.firstname.ilike(f'%{who}%') | Person.surname.ilike(f'%{who}%') ).all();
|
|
|
|
for p in people:
|
|
resp[p.id]={}
|
|
resp[p.id]['id']=p.id
|
|
resp[p.id]['tag']=p.tag
|
|
resp[p.id]['firstname']=p.firstname
|
|
resp[p.id]['surname']=p.surname
|
|
|
|
return make_response( resp )
|
|
|
|
class FaceRefimgLinkSchema(ma.SQLAlchemyAutoSchema):
|
|
class Meta: model = FaceRefimgLink
|
|
face_distance = ma.auto_field() # Explicitly include face_distance
|
|
load_instance = True
|
|
|
|
class PersonSchema(ma.SQLAlchemyAutoSchema):
|
|
class Meta: model=Person
|
|
load_instance = True
|
|
|
|
class RefimgSchema(ma.SQLAlchemyAutoSchema):
|
|
class Meta:
|
|
model = Refimg
|
|
exclude = ('face',)
|
|
load_instance = True
|
|
person = ma.Nested(PersonSchema)
|
|
|
|
class FaceSchema(ma.SQLAlchemyAutoSchema):
|
|
class Meta:
|
|
model=Face
|
|
exclude = ('face',)
|
|
load_instance = True
|
|
refimg = ma.Nested(RefimgSchema,allow_none=True)
|
|
refimg_lnk = ma.Nested(FaceRefimgLinkSchema,allow_none=True)
|
|
|
|
################################################################################
|
|
# /add_refimg_to_person/ -> POST
|
|
################################################################################
|
|
@app.route("/add_refimg_to_person", methods=["POST"])
|
|
@login_required
|
|
def add_refimg_to_person():
|
|
stmt = select(Face).options( joinedload(Face.refimg_lnk) ).where(Face.id == request.form['face_id'])
|
|
f=db.session.execute(stmt).scalars().first()
|
|
stmt = select(Person).options( joinedload(Person.refimg) ).where(Person.id == request.form['person_id'])
|
|
p=db.session.execute(stmt).scalars().first()
|
|
|
|
# add this fname (of temp refimg) to person
|
|
fname=TempRefimgFile( request.form['refimg_data'], p.tag )
|
|
r=AddRefimgToPerson( fname, p )
|
|
|
|
# connect the refimg to the face in the db, now we have added this refimg to the person
|
|
frl=FaceRefimgLink( face_id=f.id, refimg_id=r.id, face_distance=0 )
|
|
db.session.add(frl)
|
|
db.session.commit()
|
|
|
|
if request.form['search'] == "true":
|
|
jex=[]
|
|
ptype=PathType.query.filter(PathType.name=='Import').first()
|
|
jex.append( JobExtra( name=f"person", value="all" ) )
|
|
jex.append( JobExtra( name=f"path_type", value=str(ptype.id) ) )
|
|
job=NewJob( name="run_ai_on_path", num_files=0, wait_for=None, jex=jex, desc="Look for face(s) in import path(s)" )
|
|
|
|
jex=[]
|
|
ptype=PathType.query.filter(PathType.name=='Storage').first()
|
|
jex.append( JobExtra( name=f"person", value="all" ) )
|
|
jex.append( JobExtra( name=f"path_type", value=str(ptype.id) ) )
|
|
job=NewJob( name="run_ai_on_path", num_files=0, wait_for=None, jex=jex, desc="Look for face(s) in storage path(s)" )
|
|
|
|
refimg_schema=RefimgSchema(many=False)
|
|
r_data=refimg_schema.dump(r)
|
|
frl_schema=FaceRefimgLinkSchema(many=False)
|
|
frl_data=refimg_schema.dump(r)
|
|
|
|
return make_response( jsonify( refimg=r_data, frl=frl_data ) )
|
|
|
|
################################################################################
|
|
# /add_force_match_override -> POST
|
|
################################################################################
|
|
@app.route("/add_force_match_override", methods=["POST"])
|
|
@login_required
|
|
def add_force_match_override():
|
|
person_id = request.form['person_id']
|
|
p = Person.query.get(person_id);
|
|
if not p:
|
|
raise Exception( f"could not find person (id={person_id}) to add override too!" )
|
|
|
|
face_id = request.form['face_id']
|
|
f = Face.query.get(face_id);
|
|
if not f:
|
|
raise Exception("could not find face to add override for!")
|
|
|
|
mo = FaceForceMatchOverride( face_id=f.id, person_id=p.id )
|
|
db.session.add( mo )
|
|
db.session.commit()
|
|
|
|
jex=[]
|
|
jex.append( JobExtra( name="which", value="add_force_match_override" ) )
|
|
jex.append( JobExtra( name="face_id", value=str(f.id) ) )
|
|
jex.append( JobExtra( name="person_id", value=str(p.id) ) )
|
|
# dont do status update here, the F/E is in the middle of a dbox, just send metadata through to the B/E
|
|
NewJob( "metadata", num_files=0, wait_for=None, jex=jex, desc="create metadata for adding forced match" )
|
|
|
|
# this will reply to the Ajax / POST, and cause the page to re-draw with new face override to person_tag
|
|
person_schema = PersonSchema(many=False)
|
|
p_data = person_schema.dump(p)
|
|
return make_response( jsonify( person=p_data ) )
|
|
|
|
################################################################################
|
|
# /remove_force_match_override -> POST
|
|
################################################################################
|
|
@app.route("/remove_force_match_override", methods=["POST"])
|
|
@login_required
|
|
def remove_force_match_override():
|
|
face_id = request.form['face_id']
|
|
person_tag = request.form['person_tag']
|
|
file_eid = request.form['file_eid']
|
|
|
|
FaceForceMatchOverride.query.filter( FaceForceMatchOverride.face_id==face_id ).delete()
|
|
db.session.commit()
|
|
|
|
# needed to use person_id in job below (allows consistent processing in job_mgr)
|
|
p=Person.query.filter(Person.tag==person_tag).one()
|
|
|
|
jex=[]
|
|
jex.append( JobExtra( name="which", value="remove_force_match_override" ) )
|
|
jex.append( JobExtra( name="face_id", value=str(face_id) ) )
|
|
jex.append( JobExtra( name="person_id", value=str(p.id) ) )
|
|
# dont do status update here, the F/E is in the middle of a dbox, just send metadata through to the B/E
|
|
NewJob( "metadata", num_files=0, wait_for=None, jex=jex, desc="create metadata for removing forced match" )
|
|
|
|
# this will reply to the Ajax / POST, and cause the page to re-draw with new face override (data is not used)
|
|
return make_response( jsonify( person_tag=p.tag ) )
|
|
|
|
################################################################################
|
|
# /remove_no_match_override -> POST
|
|
################################################################################
|
|
@app.route("/remove_no_match_override", methods=["POST"])
|
|
@login_required
|
|
def remove_no_match_override():
|
|
face_id = request.form['face_id']
|
|
type_id = request.form['type_id']
|
|
|
|
FaceNoMatchOverride.query.filter( FaceNoMatchOverride.face_id==face_id, FaceNoMatchOverride.type_id==type_id ).delete()
|
|
db.session.commit()
|
|
|
|
jex=[]
|
|
jex.append( JobExtra( name="which", value="remove_no_match_override" ) )
|
|
jex.append( JobExtra( name="face_id", value=str(face_id) ) )
|
|
jex.append( JobExtra( name="type_id", value=str(type_id) ) )
|
|
# dont do status update here, the F/E is in the middle of a dbox, just send metadata through to the B/E
|
|
NewJob( "metadata", num_files=0, wait_for=None, jex=jex, desc="create metadata for removing forced non-match" )
|
|
|
|
# this will reply to the Ajax / POST, and cause the page to re-draw with new face override (data is not used)
|
|
return make_response( jsonify( face_id=face_id ) )
|
|
|
|
|
|
class FaceOverrideTypeSchema(ma.SQLAlchemyAutoSchema):
|
|
class Meta:
|
|
model = FaceOverrideType
|
|
load_instance = True
|
|
|
|
################################################################################
|
|
# /add_no_match_override -> POST
|
|
################################################################################
|
|
@app.route("/add_no_match_override", methods=["POST"])
|
|
@login_required
|
|
def add_no_match_override():
|
|
face_id = request.form['face_id']
|
|
f = Face.query.get(face_id)
|
|
if not f:
|
|
raise Exception("could not find face to add override too!")
|
|
|
|
type_id = request.form['type_id']
|
|
t = FaceOverrideType.query.get(type_id);
|
|
if not t:
|
|
raise Exception("could not find override_type to add override for!")
|
|
|
|
nmo = FaceNoMatchOverride( face_id=f.id, type_id=t.id )
|
|
db.session.add( nmo )
|
|
db.session.commit()
|
|
|
|
jex=[]
|
|
jex.append( JobExtra( name="which", value="add_no_match_override" ) )
|
|
jex.append( JobExtra( name="face_id", value=str(f.id) ) )
|
|
jex.append( JobExtra( name="type_id", value=str(t.id) ) )
|
|
# dont do status update here, the F/E is in the middle of a dbox, just send metadata through to the B/E
|
|
NewJob( "metadata", num_files=0, wait_for=None, jex=jex, desc="create metadata for adding forced non-match" )
|
|
|
|
fot_schema = FaceOverrideTypeSchema(many=False)
|
|
t_data=fot_schema.dump(t)
|
|
return make_response( jsonify( type_id=t.id, type=t_data ) )
|