util.py

import json
import os
from datetime import datetime

import cv2
import numpy.typing as npt


def error(error_str) -> None:
    now = datetime.now()
    date_now, time_now = (
        now.strftime("%Y-%m-%d-%H-%M"),
        now.strftime("%Y-%m-%d-%H-%M-%S"),
    )
    with open("cache/error-logs/errors.txt", "a") as file:
        file.write(f"\n{error_str} @ {date_now} @ {time_now}")


class Run_Utils:
    def __init__(self):
        (
            self.CURRENT_DIR,
            self.CACHE_DIR,
            self.PROCESS_CACHE,
            self.ASSET_FOLDER,
            self.CACHE_FOLDERS,
        ) = [
            os.listdir(os.getcwd()),
            "cache",
            "process_logs",
            "plane_assets",
            ["error-logs", "planes_cache"],
        ]

    def usage_limit(self, image: npt.NDArray) -> tuple:
        img = cv2.imread(image)
        h, w = img.shape[:2]
        t_pixel = int(h, w)
        return t_pixel

    def check_error_file_exists(self) -> None:
        if "errors.txt" not in os.listdir("cache/error-logs"):
            with open("cache/error-logs/errors.txt", "w") as file:
                file.close()

    def check_planes_assets_folder_exists(self) -> None:
        if self.ASSET_FOLDER not in self.CURRENT_DIR:
            os.mkdir(self.ASSET_FOLDER)

    def create_cache_directory(self) -> None:
        if self.CACHE_DIR not in self.CURRENT_DIR:
            os.mkdir(self.CACHE_DIR)

    def create_cache_folders(self) -> None:
        for folder in self.CACHE_FOLDERS:
            if folder not in os.listdir(self.CACHE_DIR):
                os.mkdir(f"{self.CACHE_DIR}/{folder}")

    def locate_completion_file(self) -> None:
        filename = "cm.ini"
        if filename in os.listdir(self.CACHE_DIR):
            try:
                os.remove(f"{self.CACHE_DIR}/{filename}")
            except Exception as e:
                print(e)
        else:
            pass

    def process_cache_exists(self) -> None:
        if "process_logs" in os.listdir(self.CACHE_DIR):
            OUTPUT_PROCESS_LOGS = f"{self.CACHE_DIR}/{self.PROCESS_CACHE}"
            for assets in os.listdir(OUTPUT_PROCESS_LOGS):
                if assets.endswith(".json"):
                    os.remove(f"{OUTPUT_PROCESS_LOGS}/{assets}")
        else:
            os.mkdir(f"{self.CACHE_DIR}/{self.PROCESS_CACHE}")

    def init_checks(self):
        self.check_planes_assets_folder_exists()
        self.create_cache_directory()
        self.create_cache_folders()
        self.locate_completion_file()
        self.process_cache_exists()