khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Resetting utils subtree.

Browse Source
This commit is contained in:
khushal
2025-01-07 18:51:02 +05:30
parent b20d4850c2
commit 9a9be7d4dc
187 changed files with 140 additions and 141306 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils_v2/ai/__init__.py
View File
utils_v2/ai/image_classification/__init__.py
View File
utils_v2/ai/image_classification/async_blur.py
-202
View File
@@ -1,202 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 2nd Sept., 2024
OBJECTIVE:
To provide an easy way to assess images for blurriness.
Tried and implemented using HuggingFace behaviour_models.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# For using the AI model:
import torch
from transformers import pipeline
# To download images from URLs:
import requests
# To read images:
from PIL import Image
# Common:
from shared import variables
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AssessImageBlur:
def __init__(
self,
model,
blur_label = "Blur",
clarity_label = "Normal",
blur_threshold = 0.25,
clarity_threshold = 0.65
):
"""
This class assesses the input image and tells if it is blurry or clear.
:param model: The model to use. Either the name of the HuggingFace repo, or the directory where the model is
stored.
:param blur_threshold: The max allowed blurriness (0 to 1 range).
:param clarity_threshold: The minimum needed clarity (0 to 1 range).
"""
# Note down the config:
self.__model = model
self.__device = "gpu" if torch.cuda.is_available() else "cpu"
self.__blur_label = blur_label
self.__clarity_label = clarity_label
self.__blur_threshold = blur_threshold
self.__clarity_threshold = clarity_threshold
# Initialize the classifier:
self.__classifier = pipeline(
task = "image-classification",
model = self.__model,
device = self.__device
)
@staticmethod
async def read_image(source):
"""
Reads an image in whichever format it is provided and returns it as a PIL object.
:param source: The image as either a path or a URL or a io.BytesIO object.
:return: The image opened as a PIL object.
"""
# If the input image is already a PIL image:
if isinstance(source, Image.Image): return source
# If a buffer is provided:
elif isinstance(source, io.BytesIO):
source.seek(0)
return Image.open(source)
# If a string is provided (local path or URL):
elif isinstance(source, str):
if source.startswith("http://") or source.startswith("https://"):
response = await variables.http_client.get(source)
return Image.open(io.BytesIO(response.content))
else: return Image.open(source)
async def classify(self, image):
"""
To get the prediction of the model from the given input image.
:param image: The image (as a PIL object or file path or a URL).
:return: The dictionary of classes with their respective predictions.
"""
image = await self.read_image(image)
predictions = self.__classifier(image)
classes = {p["label"]: p["score"] for p in predictions}
return classes
async def is_ok(
self,
image,
blur_threshold = None,
clarity_threshold = None
):
"""
Checks if the image given to it can be used, or should be rejected.
:param image: The image (as a PIL object or file path or a URL).
:param blur_threshold: A custom threshold to test against. If not provided, the default will be taken that was
provided when the instance was created.
:param clarity_threshold: A custom threshold to test against. If not provided, the default will be taken that
was provided when the instance was created.
:return: True if the image is okay, else False.
"""
image = await self.read_image(image)
blur_threshold = blur_threshold or self.__blur_threshold
clarity_threshold = clarity_threshold or self.__clarity_threshold
classes = await self.classify(image)
if (
classes[self.__blur_label] <= blur_threshold and
classes[self.__clarity_label] >= clarity_threshold
): return True
return False
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import asyncio
async def main():
image = r"/home/developer/Downloads/low-res-check.png"
assessor = AssessImageBlur(model = r"/home/developer/PycharmProjects/utils/data/ai/behaviour_models/hugging_face/image_classification/BlurOrBokeh")
usable = await assessor.is_ok(image)
classes = await assessor.classify(image)
print("IS OKAY:", usable)
print("CLASSES:", classes)
asyncio.run(main())
utils_v2/ai/image_classification/async_nsfw.py
-188
View File
@@ -1,188 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Monday, 2nd Sept., 2024
OBJECTIVE:
To provide an easy way to assess images for adult content.
Tried and implemented using HuggingFace behaviour_models.
REFERENCES:
N/A
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# For using the AI model:
import torch
from transformers import pipeline
# To download images from URLs:
import requests
# To read images:
from PIL import Image
# Common:
from shared import variables
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class AssessImageNSFW:
def __init__(
self,
model,
nsfw_label = "nsfw",
nsfw_threshold = 0.25,
):
"""
This class assesses the input image and tells if it is blurry or clear.
:param model: The model to use. Either the name of the HuggingFace repo, or the directory where the model is
stored.
:param nsfw_threshold: The max allowed blurriness (0 to 1 range).
"""
# Note down the config:
self.__model = model
self.__device = "gpu" if torch.cuda.is_available() else "cpu"
self.__nsfw_label = nsfw_label
self.__nsfw_threshold = nsfw_threshold
# Initialize the classifier:
self.__classifier = pipeline(
task = "image-classification",
model = self.__model,
device = self.__device
)
@staticmethod
async def read_image(source):
"""
Reads an image in whichever format it is provided and returns it as a PIL object.
:param source: The image as either a path or a URL or a io.BytesIO object.
:return: The image opened as a PIL object.
"""
# If the input image is already a PIL image:
if isinstance(source, Image.Image): return source
# If a buffer is provided:
elif isinstance(source, io.BytesIO):
source.seek(0)
return Image.open(source)
# If a string is provided (local path or URL):
elif isinstance(source, str):
if source.startswith("http://") or source.startswith("https://"):
response = await variables.http_client.get(source)
return Image.open(io.BytesIO(response.content))
else: return Image.open(source)
async def classify(self, image):
"""
To get the prediction of the model from the given input image.
:param image: The image (as a PIL object or file path or a URL).
:return: The dictionary of classes with their respective predictions.
"""
image = await self.read_image(image)
predictions = self.__classifier(image)
classes = {p["label"]: p["score"] for p in predictions}
return classes
async def is_ok(
self,
image,
nsfw_threshold = None
):
"""
Checks if the image given to it can be used, or should be rejected.
:param image: The image (as a PIL object or file path or a URL).
:param nsfw_threshold: A custom threshold to test against. If not provided, the default will be taken that was
provided when the instance was created.
:return: True if the image is okay, else False.
"""
image = await self.read_image(image)
nsfw_threshold = nsfw_threshold or self.__nsfw_threshold
classes = await self.classify(image)
return True if classes[self.__nsfw_label] <= nsfw_threshold else False
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import asyncio
async def main():
assessor = AssessImageNSFW(model = r"/path/to/model")
usable = await assessor.is_ok(r"/path/to/image/img.jpg")
classes = await assessor.classify(r"https://...")
asyncio.run(main())
utils_v2/ai/image_segmentation/__init__.py
View File
utils_v2/ai/image_segmentation/async_dichotomous_segmentation.py
-223
View File
@@ -1,223 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Saturday, 14th Sept., 2024
OBJECTIVE:
To provide an easy way to get masks from dichotomous image segmentation.
This code uses a very specific model from HuggingFace: "ZhengPeng7/BiRefNet-portrait".
You may experiment with other behaviour_models too, but make sure that model is made for "dichotomous" behaviour. This
means that the model should have only two classes like "foreground", and "background". The specified model was
trained for implementing portrait mode style blurring of backgrounds.
The originally tested model has an MIT license as per their GitHub page. The code in this file may or may not
support drop-in replacement for other behaviour_models, please be aware about this.
REFERENCES:
01. https://huggingface.co/ZhengPeng7/BiRefNet-portrait
02. https://github.com/ZhengPeng7/BiRefNet
DOWNLOADS:
01. https://huggingface.co/ZhengPeng7/BiRefNet-portrait/tree/main
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# For system-level activities:
import io
# To work with PIL images:
from PIL import Image
# To make asynchronous HTTP calls:
from shared.variables import http_client
# For using the AI model:
import torch
from torchvision import transforms
from transformers import AutoModelForImageSegmentation
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class DichotomousSegmenter:
def __init__(
self,
model = r"zhengpeng7/BiRefNet-portrait"
):
"""
This class uses Dichotomous Image Segmentation to produce a mask of what is in the foreground (or what is of
interest in a given image).
:param model: The path of the model.
"""
# Check if GPU is available for faster predictions:
self.__cuda_is_available = True if torch.cuda.is_available() else False
# Initialize the model:
self.__model = AutoModelForImageSegmentation.from_pretrained(
pretrained_model_name_or_path = model,
trust_remote_code = True
)
torch.set_float32_matmul_precision(["high", "highest"][0])
self.__model.to("cuda" if self.__cuda_is_available else "cpu")
@staticmethod
async def read_image(source):
"""
Reads an image in whichever format it is provided and returns it as a PIL object.
:param source: The image as either a path or a URL or a io.BytesIO object.
:return: The image opened as a PIL object.
"""
# If the input image is already a PIL image:
if isinstance(source, Image.Image): return source
# If a buffer is provided:
elif isinstance(source, io.BytesIO):
source.seek(0)
return Image.open(source)
# If a string is provided (local path or URL):
elif isinstance(source, str):
if source.startswith("http://") or source.startswith("https://"):
response = await http_client.get(source)
return Image.open(io.BytesIO(response.content))
else: return Image.open(source)
async def get_mask(self, image):
"""
Process the image to figure out the mask. In some cases (like the image of a sunset) you will have no white
pixels in the mask. White pixels represent the areas that the AI considered to be the foreground (area of
interest) and the black parts are the background.
:param image: The image data as either a PIL object, or a path to a file on the local disk, or a URL.
:return: The mask as a PIL object.
"""
# Prepare the image transformer:
image_size = (1024, 1024)
transform_image = transforms.Compose([
transforms.Resize(image_size),
transforms.ToTensor(),
transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
])
# Read and transform the image:
image = await self.read_image(image)
input_image = transform_image(image).unsqueeze(0).to("cuda" if self.__cuda_is_available else "cpu")
# Make the prediction:
with torch.no_grad():
if self.__cuda_is_available: mask = self.__model(input_image)[-1].sigmoid().cuda()
else: mask = self.__model(input_image)[-1].sigmoid().cpu()
# Create a PIL object of the mask data:
mask = mask[0].squeeze()
mask = transforms.ToPILImage()(mask)
mask = mask.resize(image.size)
# Done here:
return mask
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
import asyncio
import numpy as np
async def main():
model = DichotomousSegmenter()
images = [
# r"/home/developer/Downloads/kate.jpg",
# r"/home/developer/Downloads/IMG-20240730-WA0001.jpg",
# r"/home/developer/Downloads/pexels-pixabay-57416.jpg",
# r"/home/developer/Downloads/summer_clothes.jpg",
# r"/home/developer/Downloads/low_res_cat - upscaled.png",
# r"/home/developer/Downloads/blurry_traffic.jpg",
r"/home/developer/Downloads/card_0.jpg",
r"/home/developer/Downloads/card_1.jpg",
r"/home/developer/Downloads/card_2.jpg",
r"/home/developer/Downloads/card_3.jpg",
r"/home/developer/Downloads/sushmita_card.jpg",
r"/home/developer/Downloads/niranjan_card.jpg",
r"/home/developer/Downloads/niranjan_card_2.jpg",
]
for image in images:
print(image.split("/")[-1])
image = await DichotomousSegmenter.read_image(image)
mask = await model.get_mask(image)
image.show()
mask.show()
print("MASK!")
array = np.array(mask)
print("Dimensions of the array:", array.shape)
masked_count = np.sum(array > 125)
print(masked_count, type(masked_count), np.size, type(np.size))
percentage = masked_count / np.prod(array.shape)
print("Area:", round(percentage * 100, 4))
# image.putalpha(mask)
# image.show()
# mask.show()
asyncio.run(main())
utils_v2/ai/object_detection/__init__.py
View File
utils_v2/ai/object_detection/yolo.py
-176
View File
@@ -1,176 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 4th Jul, 2024
OBJECTIVE:
To provide a class to detect objects in images using YOLO behaviour_models.
REFERENCES:
1) Code Examples: https://docs.ultralytics.com/usage/python/
2) Models: https://docs.ultralytics.com/models/yolov8/#supported-tasks-and-modes
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import os
# To use YOLO architecture:
from ultralytics import YOLO
# To work with images:
from PIL import Image
# utils:
from utils_v2.string import json
# Common:
from shared import constants
# For debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** FUNCTIONS ***
# ***** ****
# *****************************************************************************************************************
class YoloDetect:
def __init__(
self,
model_file,
debug = True,
debug_prefix = "YOLO | "
):
# Load the model:
self._model = YOLO(model = model_file)
# Initialize the dbugging tool:
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self._printer.disable()
def disable_debug(self):
self._printer.disable()
def enable_debug(self):
self._printer.enable()
@staticmethod
def _process_result(raw_result):
# Extract the various types of results available from the inference:
speed = {k: (v / 100.0) for k, v in raw_result.speed.items()}
class_mapping = raw_result.names
probabilities = raw_result.probs
boxes = raw_result.boxes
masks = raw_result.masks
# Construct a default response:
response_json = {
"speed": speed,
"classes": class_mapping,
"boxes": None
}
# Object-Detection results:
formatted_boxes = []
for box in boxes:
detected_class = int(box.cls.numpy()[0])
x1, y1, x2, y2 = box.xyxy[0]
formatted_boxes.append(
{
"class": detected_class,
"className": class_mapping[detected_class],
"confidence": float(box.conf.numpy()[0]),
"x1": int(x1.numpy()),
"y1": int(y1.numpy()),
"x2": int(x2.numpy()),
"y2": int(y2.numpy())
}
)
response_json["boxes"] = formatted_boxes
# Done here:
return response_json
def predict(self, image, show = False, verbose = False):
# Process the input image with the given task:
results = self._model(
source = image,
show = show,
verbose = verbose
)
# Based on the task, interpret the results:
results_json = self._process_result(results[0])
# Done here:
return results_json
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
# model_file_path = os.path.join(constants.PROJECT_DIRECTORY, "ai", "yolo", "behaviour_models", "yolov8x.pt")
model_file_path = r"/home/developer/PycharmProjects/utils/data/ai/models/hugging_face/object_detection/YOLOv10-Document-Layout-Analysis/yolov10x_best.pt"
# sample_image_path = r"/home/developer/Downloads/2_cats.jpg"
sample_image_path = r"/home/developer/Downloads/flattened_image.jpg"
my_yolo = YoloDetect(model_file = model_file_path)
results = my_yolo.predict(image = Image.open(sample_image_path))
print("FINAL RESULTS:", json.to_string(results))
utils_v2/ai/tts/__init__.py
View File
utils_v2/ai/tts/easy_tts.py
-175
View File
@@ -1,175 +0,0 @@
"""
AUTHOR:
Khushal P Soonderji
DATE:
Thursday, 26th Sept., 2024
OBJECTIVE:
To provide a fast way to make audio from TTS engines.
REFERENCES:
01. Usage: https://github.com/myshell-ai/MeloTTS/blob/main/docs/install.md#python-api
02. Installation: https://github.com/myshell-ai/MeloTTS/blob/main/docs/install.md#linux-and-macos-install
DOWNLOADS:
N/A
"""
# *****************************************************************************************************************
# ***** ****
# *** IMPORT ***
# ***** ****
# *****************************************************************************************************************
# To make sibling directories accessible for imports:
import sys
sys.path.append(".")
sys.path.append("..")
# System-level activities:
import io
# To use the AI:
from melo.api import TTS
import numpy as np
import soundfile
# *****************************************************************************************************************
# ***** ****
# *** MACROS / ONE-TIME INIT ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** VARIABLES ***
# ***** ****
# *****************************************************************************************************************
# --- Nothing Yet
# *****************************************************************************************************************
# ***** ****
# *** CLASSES ***
# ***** ****
# *****************************************************************************************************************
class EasyTTS:
__sampling_rate = 44_100
def __init__(
self,
language = "EN",
speed = 1.0
):
self.__audio = np.zeros(1)
self.__language = language
self.__speed = speed
self.__model = TTS(language = language, device = "auto")
self.__speakers = self.__model.hps.data.spk2id
def list_speakers(self):
return list(self.__speakers.keys())
def speak(
self,
text,
speaker
):
this_audio = self.__model.tts_to_file(
text,
self.__speakers[speaker],
speed = self.__speed,
quiet = True
)
self.__audio = np.concatenate((self.__audio, this_audio))
def pause(self, seconds):
self.__audio = np.concatenate((
self.__audio,
np.zeros(int(self.__sampling_rate * seconds))
))
def to_wav(self, path = None):
if path is None:
audio_buffer = io.BytesIO()
soundfile.write(audio_buffer, self.__audio, self.__sampling_rate)
return audio_buffer
else: soundfile.write(path, self.__audio, self.__sampling_rate)
# *****************************************************************************************************************
# ***** ****
# *** MAIN PROGRAM ***
# ***** ****
# *****************************************************************************************************************
if __name__ == "__main__":
speaker = "EN-BR"
tts_maker = EasyTTS(language = "EN", speed = 0.9)
tts_maker.speak(
text = """
Imagine delighting your doctors with a personalized calendar,
crafted from their own cherished memories.
""",
speaker = speaker
)
tts_maker.pause(seconds = 0.3)
tts_maker.speak(
text = """
Every day, as they turn the page,
they’ll not only relive those special moments but also remember you,
the one who made it happen.
""",
speaker = speaker
)
tts_maker.pause(seconds = 0.75)
tts_maker.speak(
text = "STEP 1:",
speaker = speaker
)
tts_maker.pause(seconds = 0.3)
tts_maker.speak(
text = "Start by identifying the doctors you’d like to engage with, and add them to our app.",
speaker = speaker
)
tts_maker.pause(seconds = 0.3)
tts_maker.speak(
text = "No rush, you can add their photographs later as well.",
speaker = speaker
)
tts_maker.pause(seconds = 0.3)
tts_maker.speak(
text = "With this, your engagement funnel is created.",
speaker = speaker
)
tts_maker.to_wav(r"/home/developer/Downloads/voiceover.wav")
utils_v2/ai/tts/test.wav
BIN
View File
Binary file not shown.