add api for character image classification and background loading

Author: Mahir Mahbub

.env

@@ -4,5 +4,6 @@ DB_NAME=postgres
 DB_USERNAME=admin
 DB_PASSWORD=secret
 DB_SSL_MODE=prefer
-FILE_SOURCE_FOLDER=/app/source/
+FILE_SOURCE_FOLDER=/app/data/raw_training_set/
+OCR_IMAGE_SOURCE_FOLDER=/app/data/ocr_image/
 APP_HOST_PORT=7003

app/cruds/character.py

@@ -0,0 +1,24 @@
+from sqlalchemy.orm import Session
+
+from app.cruds.table_repository import TableRepository
+from db import models
+
+
+class CharacterCrud(TableRepository):
+
+    def __init__(self, db: Session):
+        super().__init__(db=db, entity=models.Characters)
+
+    def store(self, item, checker=None):
+        item = item.dict(exclude_unset=True)
+        exist = False
+        if checker:
+            exist = self.db.query(self.entity).filter_by(**checker).first()
+        if not exist:
+            ocr_model_object = self.entity(**item)
+            self.db.add(ocr_model_object)
+            return ocr_model_object
+
+    def get_images(self, limit=5):
+        return self.db.query(self.entity).filter(self.entity.is_labeled == False,
+                                                 self.entity.class_id == None).limit(limit).all()

app/cruds/class_label.py

@@ -0,0 +1,25 @@
+from sqlalchemy import func
+from sqlalchemy.orm import Session
+
+from app.cruds.table_repository import TableRepository
+from db import models
+
+
+class ClassLabelCrud(TableRepository):
+
+    def __init__(self, db: Session):
+        super().__init__(db=db, entity=models.ClassLabel)
+
+    def store(self, item, checker=None):
+        item = item.dict(exclude_unset=True)
+        exist = False
+        if checker:
+            exist = self.db.query(self.entity).filter_by(**checker).first()
+        if not exist:
+            ocr_model_object = self.entity(**item)
+            self.db.add(ocr_model_object)
+            return ocr_model_object
+
+    def get_by_round_robin(self):
+        sub_query = self.db.query(func.min(self.entity.round_robin_marker))
+        return self.db.query(self.entity).filter(self.entity.round_robin_marker == sub_query).first()

app/cruds/ocr_tools.py

@@ -0,0 +1,13 @@
+from sqlalchemy.orm import Session
+
+from app.cruds.table_repository import TableRepository
+from db import models
+
+
+class OcrToolCrud(TableRepository):
+
+    def __init__(self, db: Session):
+        super().__init__(db=db, entity=models.OcrData)
+
+    def get_by_non_extracted(self):
+        return self.db.query(self.entity).filter(self.entity.is_extracted == False).all()

app/cruds/table_repository.py

@@ -0,0 +1,32 @@
+from typing import TypeVar, Generic
+
+from sqlalchemy.orm import Session
+
+from db import models
+
+T = TypeVar('T')
+
+
+class TableRepository:
+    entity: Generic[T] = None
+    db: Session = NotImplementedError
+
+    def __init__(self, db: Session, entity: Generic[T]):
+        self.db: Session = db
+        self.entity: Generic[T] = entity
+
+    def store(self, item):
+        item = item.dict(exclude_unset=True)
+        ocr_model_object = self.entity(**item)
+        self.db.add(ocr_model_object)
+        return ocr_model_object
+
+    def get(self, id_: int):
+        return self.db.query(self.entity).filter(self.entity.id == id_).first()
+
+    def gets(self):
+        return self.db.query(self.entity).all()
+
+    def update(self, id_, item):
+        item_dict = item.dict(exclude_unset=True)
+        return self.db.query(self.entity).filter(self.entity.id == id_).update(item_dict)

app/custom_classes/file_path.py

@@ -0,0 +1,16 @@
+import os
+
+
+def next_file_name(file_name_pattern, bucket_id, main_file_name):
+    if not os.path.exists(bucket_id + main_file_name):
+        return main_file_name
+    i = 1
+    while os.path.exists(bucket_id + file_name_pattern % i):
+        i = i * 2
+    left, right = (i // 2, i)
+    while left + 1 < right:
+        middle = (left + right) // 2
+        left, right = (middle, right) if os.path.exists(bucket_id + file_name_pattern % middle) else (left, middle)
+
+    return file_name_pattern % right
+

app/custom_classes/job_manager.py

@@ -0,0 +1,94 @@
+import os
+import time
+from typing import List
+
+import imageio
+from fastapi.encoders import jsonable_encoder
+
+from app.cruds.character import CharacterCrud
+from app.cruds.class_label import ClassLabelCrud
+from app.cruds.ocr_tools import OcrToolCrud
+from app.custom_classes.ocr_character_seperator import OcrCharacterSeperator
+from db import models
+from db.database import SessionLocal
+from db.schemas import CharacterCreate, OcrDataUpdate, ClassLabelCreate
+
+
+class BaseJobManager(object):
+    def __init__(self):
+        self.db = SessionLocal()
+
+    @staticmethod
+    def execute():
+        pass
+
+
+class PrintJobManager(BaseJobManager):
+    def __init__(self):
+        super().__init__()
+
+    def print_hello_activity(self, should_run):
+        """Work Flow Start"""
+        print("nabila")
+        time.sleep(4)
+        """Work Flow End"""
+
+    @staticmethod
+    def execute():
+        PrintJobManager().print_hello_activity(should_run=True)
+
+
+class PreOcrCharacterLoad(BaseJobManager):
+    def __init__(self):
+        super().__init__()
+
+    def ocr_character_collection_activity(self, should_run):
+        # preload_flag = self.db.query(models.Properties).filter(models.Properties.name == "CharacterDataPreLoad").first()
+        # print(preload_flag)
+        # if not preload_flag:
+        current_path = os.getcwd()
+        class_data_path = "/app/data/training_set/"
+        list_dir = os.listdir(current_path + class_data_path)
+        for class_name in list_dir:
+            label_item = ClassLabelCreate(class_id=class_name)
+            ClassLabelCrud(db=self.db).store(item=label_item, checker={"class_id": class_name})
+
+            list_of_files = [current_path + class_data_path + os.path.join(class_name, f) for f in
+                                 os.listdir(current_path + class_data_path + class_name + "/")]
+            for file in list_of_files:
+                item = CharacterCreate(character_path=file,
+                                           class_id=class_name,
+                                           is_labeled=True)
+                CharacterCrud(db=self.db).store(item=item, checker={"character_path": file})
+            # self.db.commit()
+        self.db.add(models.Properties(name="CharacterDataPreLoad", value=True))
+        self.db.commit()
+
+    @staticmethod
+    def execute():
+        PreOcrCharacterLoad().ocr_character_collection_activity(should_run=True)
+
+
+class CharacterExtractorManager(BaseJobManager):
+    def __init__(self):
+        super().__init__()
+
+    def character_extract_activity(self, should_run):
+        ocr_processing_object: OcrCharacterSeperator = OcrCharacterSeperator()
+        ocr_image_paths: List[models.OcrData] = OcrToolCrud(db=self.db).get_by_non_extracted()
+        print(ocr_image_paths)
+        # print(os.getcwd())
+        for ocr_image in ocr_image_paths:
+            images_and_save_path = ocr_processing_object.character_extractor(ocr_image.file_path)
+            for save_path, char_img in images_and_save_path:
+                imageio.imwrite(save_path, char_img)
+                item = CharacterCreate(character_path=save_path)
+                character_model_object = CharacterCrud(db=self.db).store(jsonable_encoder(item))
+                self.db.add(character_model_object)
+            item = OcrDataUpdate(is_extracted=True)
+            OcrToolCrud(db=self.db).update(id_=ocr_image.id, item=item)
+            self.db.commit()
+
+    @staticmethod
+    def execute():
+        CharacterExtractorManager().character_extract_activity(should_run=True)

app/custom_classes/job_trigger.py

@@ -0,0 +1,37 @@
+# import datetime
+#
+# from apscheduler.triggers.base import BaseTrigger
+#
+#
+# class IntervalTrigger(BaseTrigger):
+#     def __init__(self, seconds=0, minutes=0, hours=0, days=0, milliseconds=0, weeks=0, microseconds=0):
+#         self.seconds = seconds
+#         self.minutes = minutes
+#         self.hours = hours
+#         self.days = days
+#         self.milliseconds = milliseconds
+#         self.weeks = weeks
+#         self.microseconds = microseconds
+#
+#     def get_next_fire_time(self, previous_fire_time, now):
+#         next_fire_time = previous_fire_time + datetime.timedelta(days=self.days, seconds=self.seconds, hours=self.hours,
+#                                                                  microseconds=self.microseconds,
+#                                                                  milliseconds=self.milliseconds, minutes=self.minutes,
+#                                                                  weeks=self.weeks)
+from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.date import DateTrigger
+from apscheduler.triggers.interval import IntervalTrigger
+
+
+class BuildInJobTrigger(object):
+    def __new__(cls, trigger, cron_enable, **kwargs):
+        register = {}
+        register["Interval"] = IntervalTrigger
+        print(list(kwargs.values()))
+        register["Date"] = DateTrigger
+        if cron_enable:
+            register["Cron"] = CronTrigger.from_crontab(list(kwargs.values())[0])
+            return register["Cron"]
+        else:
+            register["Cron"] = CronTrigger
+        return register[trigger](**kwargs)