from appPublic.worker import awaitify from appPublic.log import debug, exception from appPublic.uniqueID import getID from sqlor.dbpools import DBPools from ahserver.filestorage import FileStorage import face_recognition from PIL import Image async def create_knownface(imgid, logo_url, name): db = DBPools() dbname = 'mediadb' async with db.sqlorContext(dbname) as sor: return await _create_knownface(sor, imgid, logo_url, name) e = Exception(f'{imgid=}, {logo_url=}, {name=}:create_knownface() error') exception(f'{e=}') raise e async def _create_knownface(sor, imgid, logo_url, name): fs = FileStorage() imgpath = fs.realPath(logo_url) imgobj = await awaitify(face_recognition.load_image_file)(imgpath) infos = await awaitify(face_recognition.face_encodings)(imgobj) locs = await awaitify(face_recognition.face_locations)(imgobj) if len(locs) > 1: e = Exception(f'{imgpath} found more than one face') exception(f'{e=}') raise e if len(locs) == 0: e = Exception(f'{imgpath} found not face') exception(f'{e=}') raise e l = locs[0] faceinfo = json.dumps({ 'feature':infos[0], "face_top":top, "face_bottom":bottom, "face_left":left, "face_right":right }) ns = { "id":getID(), "mediaid":imgid, "faceinfo":faceinfo, "personname":name } await sor.C('photofaces', ns)