call Archives

Tuning your model using the python-based low-code machine-learning library PyCaret

Posted on March 31, 2023March 31, 2023 by SatyakiDe in ai, Azure, call, cloud, code, computing, Crossplatform, Data Science, features, function, json, Keras, machine-learning, Model, numpy, Pandas, Performance, Python, Technology, video

Today, I’ll discuss another important topic before I will share the excellent use case next month, as I still need some time to finish that one. We’ll see how we can leverage the brilliant capability of a low-code machine-learning library named PyCaret.

But before going through the details, why don’t we view the demo & then go through it?

Demo

Architecture:

Let us understand the flow of events –

As one can see, the initial training requests are triggered from the PyCaret-driven training models. And the application can successfully process & identify the best models out of the other combinations.

Python Packages:

Following are the python packages that are necessary to develop this use case –

pip install pandas
pip install pycaret

PyCaret is dependent on a combination of other popular python packages. So, you need to install them successfully to run this package.

CODE:

clsConfigClient.py (Main configuration file)

	################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 15-May-2020 ####
	#### Modified On: 31-Mar-2023 ####
	#### ####
	#### Objective: This script is a config ####
	#### file, contains all the keys for ####
	#### personal AI-driven voice assistant. ####
	#### ####
	################################################

	import os
	import platform as pl

	class clsConfigClient(object):
	Curr_Path = os.path.dirname(os.path.realpath(__file__))

	os_det = pl.system()
	if os_det == "Windows":
	sep = '\\'
	else:
	sep = '/'

	conf = {
	'APP_ID': 1,
	'ARCH_DIR': Curr_Path + sep + 'arch' + sep,
	'PROFILE_PATH': Curr_Path + sep + 'profile' + sep,
	'LOG_PATH': Curr_Path + sep + 'log' + sep,
	'DATA_PATH': Curr_Path + sep + 'data' + sep,
	'MODEL_PATH': Curr_Path + sep + 'model' + sep,
	'TEMP_PATH': Curr_Path + sep + 'temp' + sep,
	'MODEL_DIR': 'model',
	'APP_DESC_1': 'PyCaret Training!',
	'DEBUG_IND': 'N',
	'INIT_PATH': Curr_Path,
	'FILE_NAME': 'Titanic.csv',
	'MODEL_NAME': 'PyCaret-ft-personal-2023-03-31-04-29-53',
	'TITLE': "PyCaret Training!",
	'PATH' : Curr_Path,
	'OUT_DIR': 'data'
	}

view raw

clsConfigClient.py

hosted with ❤ by GitHub

I’m skipping this section as it is self-explanatory.

clsTrainModel.py (This is the main class that contains the core logic of low-code machine-learning library to evaluate the best model for your solutions.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 31-Mar-2023 ####
	#### Modified On 31-Mar-2023 ####
	#### ####
	#### Objective: This is the main class that ####
	#### contains the core logic of low-code ####
	#### machine-learning library to evaluate the ####
	#### best model for your solutions. ####
	#### ####
	#####################################################

	import clsL as cl
	from clsConfigClient import clsConfigClient as cf
	import datetime

	# Import necessary libraries
	import pandas as p
	from pycaret.classification import *

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	######################################
	### Get your global values ####
	######################################
	debug_ind = 'Y'

	# Initiating Logging Instances
	clog = cl.clsL()
	###############################################
	### End of Global Section ###
	###############################################


	class clsTrainModel:
	def __init__(self):
	self.model_path = cf.conf['MODEL_PATH']
	self.model_name = cf.conf['MODEL_NAME']

	def trainModel(self, FullFileName):
	try:
	df = p.read_csv(FullFileName)
	row_count = int(df.shape[0])
	print('Number of rows: ', str(row_count))

	print(df)

	# Initialize the setup in PyCaret
	clf_setup = setup(
	data=df,
	target="Survived",
	train_size=0.8, # 80% for training, 20% for testing
	categorical_features=["Sex", "Embarked"],
	ordinal_features={"Pclass": ["1", "2", "3"]},
	ignore_features=["Name", "Ticket", "Cabin", "PassengerId"],
	#silent=True, # Set to False for interactive setup
	)

	# Compare various models
	best_model = compare_models()

	# Create a specific model (e.g., Random Forest)
	rf_model = create_model("rf")

	# Hyperparameter tuning
	tuned_rf_model = tune_model(rf_model)

	# Evaluate model performance
	plot_model(tuned_rf_model, plot="confusion_matrix")
	plot_model(tuned_rf_model, plot="auc")

	# Finalize the model (train on the complete dataset)
	final_rf_model = finalize_model(tuned_rf_model)

	# Make predictions on new data
	new_data = df.drop("Survived", axis=1)
	predictions = predict_model(final_rf_model, data=new_data)

	# Writing into the Model
	FullModelName = self.model_path + self.model_name

	print('Model Output @:: ', str(FullModelName))
	print()

	# Save the fine-tuned model
	save_model(final_rf_model, FullModelName)

	return 0

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	return 1

view raw

clsTrainModel.py

hosted with ❤ by GitHub

Let us understand the code in simple terms –

Import necessary libraries and load the Titanic dataset.
Initialize the PyCaret setup, specifying the target variable, train-test split, categorical and ordinal features, and features to ignore.
Compare various models to find the best-performing one.
Create a specific model (Random Forest in this case).
Perform hyper-parameter tuning on the Random Forest model.
Evaluate the model’s performance using a confusion matrix and AUC-ROC curve.
Finalize the model by training it on the complete dataset.
Make predictions on new data.
Save the trained model for future use.

trainPYCARETModel.py (This is the main calling python script that will invoke the training class of PyCaret package.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 31-Mar-2023 ####
	#### Modified On 31-Mar-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### training class of Pycaret package. ####
	#### ####
	#####################################################

	import clsL as cl
	from clsConfigClient import clsConfigClient as cf
	import datetime

	import clsTrainModel as tm

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	######################################
	### Get your global values ####
	######################################
	debug_ind = 'Y'

	# Initiating Logging Instances
	clog = cl.clsL()

	data_path = cf.conf['DATA_PATH']
	data_file_name = cf.conf['FILE_NAME']

	tModel = tm.clsTrainModel()

	######################################
	#### Global Flag ########
	######################################

	def main():
	try:
	var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print(''120)
	print('Start Time: ' + str(var))
	print(''120)

	FullFileName = data_path + data_file_name

	r1 = tModel.trainModel(FullFileName)

	if r1 == 0:
	print('Successfully Trained!')
	else:
	print('Failed to Train!')

	print(''120)
	var1 = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print('End Time: ' + str(var1))

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	if __name__ == "__main__":
	main()

view raw

trainPYCARETModel.py

hosted with ❤ by GitHub

The above code is pretty self-explanatory as well.

testPYCARETModel.py (This is the main calling python script that will invoke the testing script for PyCaret package.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 31-Mar-2023 ####
	#### Modified On 31-Mar-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### testing script for PyCaret package. ####
	#### ####
	#####################################################

	import clsL as cl
	from clsConfigClient import clsConfigClient as cf
	import datetime

	from pycaret.classification import load_model, predict_model

	import pandas as p

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	######################################
	### Get your global values ####
	######################################
	debug_ind = 'Y'

	# Initiating Logging Instances
	clog = cl.clsL()

	model_path = cf.conf['MODEL_PATH']
	model_name = cf.conf['MODEL_NAME']

	######################################
	#### Global Flag ########
	######################################

	def main():
	try:
	var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print(''120)
	print('Start Time: ' + str(var))
	print(''120)

	FullFileName = model_path + model_name

	# Load the saved model
	loaded_model = load_model(FullFileName)

	# Prepare new data for testing (make sure it has the same columns as the original data)
	new_data = p.DataFrame({
	"Pclass": [3, 1],
	"Sex": ["male", "female"],
	"Age": [22, 38],
	"SibSp": [1, 1],
	"Parch": [0, 0],
	"Fare": [7.25, 71.2833],
	"Embarked": ["S", "C"]
	})

	# Make predictions using the loaded model
	predictions = predict_model(loaded_model, data=new_data)

	# Display the predictions
	print(predictions)

	print(''120)
	var1 = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print('End Time: ' + str(var1))

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	if __name__ == "__main__":
	main()

view raw

testPYCARETModel.py

hosted with ❤ by GitHub

In this code, the application uses the stored model & then forecasts based on the optimized PyCaret model tuning.

Conclusion:

The above code demonstrates an end-to-end binary classification pipeline using the PyCaret library for the Titanic dataset. The goal is to predict whether a passenger survived based on the available features. Here are some conclusions you can draw from the code and data:

Ease of use: The code showcases how PyCaret simplifies the machine learning process, from data preprocessing to model training, evaluation, and deployment. With just a few lines of code, you can perform tasks that would require much more effort using lower-level libraries.
Model selection: The compare_models() function provides a quick and easy way to compare various machine learning algorithms and identify the best-performing one based on the chosen evaluation metric (accuracy by default). This selection helps you select a suitable model for the given problem.
Hyper-parameter tuning: The tune_model() function automates the process of hyper-parameter tuning to improve model performance. We tuned a Random Forest model to optimize its predictive power in the example.
Model evaluation: PyCaret provides several built-in visualization tools for assessing model performance. In the example, we used a confusion matrix and AUC-ROC curve to evaluate the performance of the tuned Random Forest model.
Model deployment: The example demonstrates how to make predictions using the trained model and save the model for future use. This deployment showcases how PyCaret can streamline the process of deploying a machine-learning model in a production environment.

It is important to note that the conclusions drawn from the code and data are specific to the Titanic dataset and the chosen features. Adjust the feature engineering, preprocessing, and model selection steps for different datasets or problems accordingly. However, the general workflow and benefits provided by PyCaret would remain the same.

So, finally, we’ve done it.

I know that this post is relatively bigger than my earlier post. But, I think, you can get all the details once you go through it.

You will get the complete codebase in the following GitHub link.

I’ll bring some more exciting topics in the coming days from the Python verse. Please share & subscribe to my post & let me know your feedback.

Till then, Happy Avenging! 🙂

Note: All the data & scenarios posted here are representational data & scenarios & available over the internet & for educational purposes only. Some of the images (except my photo) we’ve used are available over the net. We don’t claim ownership of these images. There is always room for improvement & especially in the prediction quality.

Handling unique data using the python-based FastDataMask Package package

Posted on February 18, 2023February 19, 2023 by SatyakiDe in api, Azure, call, cast, cloud, code, commit, Data Science, FastDataMask, function, numpy, Pandas, Python

Today, I’ll discuss one widespread use case of handling unique & critical data using a new python-based FastDataMask package. But before going through the details, why don’t we view the demo & then go through it?

Demo

Great! Let us understand in detail.

Architecture:

Let us understand the flow of events –

The application first invokes the FastDataMask python package, which accepts individual data in nature & then generates non-recoverable masked data, keeping the data pattern & nature in mind. Hence, anyone can still use the data for their analysis, whereas you can encapsulate the information from unauthorized pairs of eyes. Yet, they can get the essence & close data patterns to decide from any data analysis.

Python Packages:

Following are the python packages that are necessary to develop this brilliant use case –

pip install FastDataMask==0.0.6
pip install imutils==0.5.3
pip install numpy==1.23.2
pip install pandas==1.4.3

CODE:

Let us now understand the code. For this use case, we will only discuss three python scripts. However, we need more than these three. However, we have already discussed them in some of the early posts. Hence, we will skip them here.

clsConfigClient.py (Main configuration file)

	################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 15-May-2020 ####
	#### Modified On: 15-Feb-2023 ####
	#### ####
	#### Objective: This script is a config ####
	#### file, contains all the keys for ####
	#### personal AI-driven voice assistant. ####
	#### ####
	################################################

	import os
	import platform as pl

	class clsConfigClient(object):
	Curr_Path = os.path.dirname(os.path.realpath(__file__))

	os_det = pl.system()
	if os_det == "Windows":
	sep = '\\'
	else:
	sep = '/'

	conf = {
	'APP_ID': 1,
	'ARCH_DIR': Curr_Path + sep + 'arch' + sep,
	'PROFILE_PATH': Curr_Path + sep + 'profile' + sep,
	'LOG_PATH': Curr_Path + sep + 'log' + sep,
	'REPORT_PATH': Curr_Path + sep + 'output' + sep,
	'REPORT_DIR': 'output',
	'SRC_PATH': Curr_Path + sep + 'data' + sep,
	'APP_DESC_1': 'Masking PII Data!',
	'DEBUG_IND': 'N',
	'INIT_PATH': Curr_Path,
	'FILE_NAME': 'PII.csv',
	'TITLE': "Masking PII Data!",
	'PATH' : Curr_Path
	}

view raw

clsConfigClient.py

hosted with ❤ by GitHub

Key entries from the above scripts are as follows –

'FILE_NAME': 'PII.csv',

This excel is a dummy input file, which looks like this –

In the above screenshot, our applications will use critical information like – First Name, Email, Address, Phone, Date Of Birth, SSN & Sal.

playPII.py (Main calling python script)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 12-Feb-2023 ####
	#### Modified On 16-Feb-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### newly created light data masking class. ####
	#### ####
	#####################################################

	import pandas as p
	import clsL as cl
	from clsConfigClient import clsConfigClient as cf
	import datetime

	from FastDataMask import clsCircularList as ccl

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	######################################
	### Get your global values ####
	######################################
	debug_ind = 'Y'
	charList = ccl.clsCircularList()

	CurrPath = cf.conf['SRC_PATH']
	FileName = cf.conf['FILE_NAME']
	######################################
	#### Global Flag ########
	######################################

	######################################
	### Wrapper functions to invoke ###
	### the desired class from newly ###
	### built class. ###
	######################################
	def mask_email(email):
	try:
	maskedEmail = charList.maskEmail(email)
	return maskedEmail
	except:
	return ''

	def mask_phone(phone):
	try:
	maskedPhone = charList.maskPhone(phone)
	return maskedPhone
	except:
	return ''

	def mask_name(flname):
	try:
	maskedFLName = charList.maskFLName(flname)
	return maskedFLName
	except:
	return ''

	def mask_date(dt):
	try:
	maskedDate = charList.maskDate(dt)
	return maskedDate
	except:
	return ''

	def mask_uniqueid(unqid):
	try:
	maskedUnqId = charList.maskSSN(unqid)
	return maskedUnqId
	except:
	return ''

	def mask_sal(sal):
	try:
	maskedSal = charList.maskSal(sal)
	return maskedSal
	except:
	return ''
	######################################
	### End of wrapper functions. ###
	######################################

	def main():
	try:
	var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print(''120)
	print('Start Time: ' + str(var))
	print(''120)

	inputFile = CurrPath + FileName

	print('Input File: ', inputFile)

	df = p.read_csv(inputFile)

	print(''120)
	print('Source Data: ')
	print(df)
	print(''120)

	hdr = list(df.columns.values)
	print('Headers:', hdr)

	df["MaskedFirstName"] = df["FirstName"].apply(mask_name)
	df["MaskedEmail"] = df["Email"].apply(mask_email)
	df["MaskedPhone"] = df["Phone"].apply(mask_phone)
	df["MaskedDOB"] = df["DOB"].apply(mask_date)
	df["MaskedSSN"] = df["SSN"].apply(mask_uniqueid)
	df["MaskedSal"] = df["Sal"].apply(mask_sal)

	# Dropping old columns
	df.drop(['FirstName','Email','Phone','DOB','SSN', 'Sal'], axis=1, inplace=True)

	# Renaming columns
	df.rename(columns={'MaskedFirstName': 'FirstName'}, inplace=True)
	df.rename(columns={'MaskedEmail': 'Email'}, inplace=True)
	df.rename(columns={'MaskedPhone': 'Phone'}, inplace=True)
	df.rename(columns={'MaskedDOB': 'DOB'}, inplace=True)
	df.rename(columns={'MaskedSSN': 'SSN'}, inplace=True)
	df.rename(columns={'MaskedSal': 'Sal'}, inplace=True)

	# Repositioning columns of dataframe
	df = df[hdr]

	print('Masked DF: ')
	print(df)

	print(''120)
	var1 = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print('End Time: ' + str(var1))

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	if __name__ == "__main__":
	main()

view raw

playPII.py

hosted with ❤ by GitHub

Let us understand the key lines in details –

def mask_email(email):
    try:
        maskedEmail = charList.maskEmail(email)
        return maskedEmail
    except:
        return ''

def mask_phone(phone):
    try:
        maskedPhone = charList.maskPhone(phone)
        return maskedPhone
    except:
        return ''

def mask_name(flname):
    try:
        maskedFLName = charList.maskFLName(flname)
        return maskedFLName
    except:
        return ''

def mask_date(dt):
    try:
        maskedDate = charList.maskDate(dt)
        return maskedDate
    except:
        return ''

def mask_uniqueid(unqid):
    try:
        maskedUnqId = charList.maskSSN(unqid)
        return maskedUnqId
    except:
        return ''

def mask_sal(sal):
    try:
        maskedSal = charList.maskSal(sal)
        return maskedSal
    except:
        return ''

These functions take a value as input and attempt to mask it using the corresponding masking method from the charList module. If the masking is successful, the process will return a masked value per input; otherwise, the application will return an empty string.

More specifically, the functions are:

mask_email: masks the email address provided as input
mask_phone: masks the phone number provided as input
mask_name: masks the first and last name supplied as input
mask_date: masks the date provided as input
mask_uniqueid: masks the unique ID (e.g., Social Security Number) provided as input
mask_sal: masks the salary supplied as input

The functions use a try-except block to handle any exceptions that may arise when calling the corresponding masking method from the charList module. If the masking method raises an exception, the function will return an empty string to handle cases where the input value is invalid, or the masking method fails for another reason.

inputFile = CurrPath + FileName
df = p.read_csv(inputFile)
hdr = list(df.columns.values)


df["MaskedFirstName"] = df["FirstName"].apply(mask_name)
df["MaskedEmail"] = df["Email"].apply(mask_email)
df["MaskedPhone"] = df["Phone"].apply(mask_phone)
df["MaskedDOB"] = df["DOB"].apply(mask_date)
df["MaskedSSN"] = df["SSN"].apply(mask_uniqueid)
df["MaskedSal"] = df["Sal"].apply(mask_sal)

# Dropping old columns
df.drop(['FirstName','Email','Phone','DOB','SSN', 'Sal'], axis=1, inplace=True)

# Renaming columns
df.rename(columns={'MaskedFirstName': 'FirstName'}, inplace=True)
df.rename(columns={'MaskedEmail': 'Email'}, inplace=True)
df.rename(columns={'MaskedPhone': 'Phone'}, inplace=True)
df.rename(columns={'MaskedDOB': 'DOB'}, inplace=True)
df.rename(columns={'MaskedSSN': 'SSN'}, inplace=True)
df.rename(columns={'MaskedSal': 'Sal'}, inplace=True)

The first line inputFile = CurrPath + FileName concatenates the current working directory path (CurrPath) with the name of a file (FileName) and assigns the resulting file path to a variable inputFile.
The second line df = p.read_csv(inputFile) – reads the file located at inputFile into a Pandas DataFrame object called df.
The following few lines apply certain functions (mask_name, mask_email, mask_phone, mask_date, mask_uniqueid, and mask_sal) to specific columns in the DataFrame to mask sensitive data. These functions likely perform some data masking or obfuscation on the input data.
The following line df.drop([‘FirstName’,’Email’,’Phone’,’DOB’,’SSN’, ‘Sal’], axis=1, inplace=True) drops the original columns that we are supposed to mask (i.e., ‘FirstName’, ‘Email’, ‘Phone’, ‘DOB’, ‘SSN’, and ‘Sal’) from the DataFrame.
The remaining lines rename the masked columns to their original names (i.e., ‘MaskedFirstName’ is renamed to ‘FirstName’, ‘MaskedEmail’ is renamed to ‘Email’, and so on).

Overall, this code reads in a file, masking specific sensitive columns, then outputting a new file with the masked data.

Now, let’s compare the output against the source data –

As you can see the blue highlighted columns are the masked column & you can compare the data pattern against the source.

So, finally, we’ve done it.

I know that this post is relatively bigger than my earlier post. But, I think, you can get all the details once you go through it.

You will get the complete codebase in the following GitHub link.

I’ll bring some more exciting topics in the coming days from the Python verse. Please share & subscribe to my post & let me know your feedback.

Till then, Happy Avenging! 🙂

Personal Virtual Assistant (SJ) implemented using python-based OpenAI, Rev_AI & PyTtSX3.

Posted on January 31, 2023January 31, 2023 by SatyakiDe in analytic function, api, Azure, BOT, call, cloud, code, computing, Data Science, function, human, json, machine-learning, natural-language, numpy, objects, openai, Pandas, Performance, Python, pyttsx3, Real-time, rev_ai

Today, I will discuss our Virtual personal assistant (SJ) with a combination of AI-driven APIs, which is now operational in Python. We will use the three most potent APIs using OpenAI, Rev-AI & Pyttsx3. Why don’t we see the demo first?

Great! Let us understand we can leverage this by writing a tiny snippet using this new AI model.

Architecture:

Let us understand the flow of events –

The application first invokes the API to capture the audio spoken through the audio device & then translate that into text, which is later parsed & shared as input by the openai for the response of the posted queries. Once, OpenAI shares the response, the python-based engine will take the response & using pyttsx3 to convert them to voice.

Python Packages:

Following are the python packages that are necessary to develop this brilliant use case –

pip install openai==0.25.0
pip install PyAudio==0.2.13
pip install playsound==1.3.0
pip install pandas==1.5.2
pip install rev-ai==2.17.1
pip install six==1.16.0
pip install websocket-client==0.59.0

CODE:

clsConfigClient.py (Main configuration file)

	################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 15-May-2020 ####
	#### Modified On: 31-Dec-2022 ####
	#### ####
	#### Objective: This script is a config ####
	#### file, contains all the keys for ####
	#### personal AI-driven voice assistant. ####
	#### ####
	################################################

	import os
	import platform as pl

	class clsConfigClient(object):
	Curr_Path = os.path.dirname(os.path.realpath(__file__))

	os_det = pl.system()
	if os_det == "Windows":
	sep = '\\'
	else:
	sep = '/'

	conf = {
	'APP_ID': 1,
	'ARCH_DIR': Curr_Path + sep + 'arch' + sep,
	'PROFILE_PATH': Curr_Path + sep + 'profile' + sep,
	'LOG_PATH': Curr_Path + sep + 'log' + sep,
	'REPORT_PATH': Curr_Path + sep + 'output' + sep,
	'REPORT_DIR': 'output',
	'SRC_PATH': Curr_Path + sep + 'data' + sep,
	'CODE_PATH': Curr_Path + sep + 'Code' + sep,
	'APP_DESC_1': 'Personal Voice Assistant (SJ)!',
	'DEBUG_IND': 'N',
	'INIT_PATH': Curr_Path,
	'TITLE': "Personal Voice Assistant (SJ)!",
	'PATH' : Curr_Path,
	'OPENAI_API_KEY': "sk-aapwfMWDuFE5XXXUr2BH",
	'REVAI_API_KEY': "02ks6kFhEKjdhdure8474JJAJJ945958_h8P_DEKDNkK6DwNNNHU17aRtCw",
	'MODEL_NAME': "code-davinci-002",
	"speedSpeech": 170,
	"speedPitch": 0.8,
	"soundRate": 44100,
	"contentType": "audio/x-raw",
	"layout": "interleaved",
	"format": "S16LE",
	"channels": 1
	}

view raw

clsConfigClient.py

hosted with ❤ by GitHub

A few of the essential entries from the above snippet, which one should be looked for, are –

'OPENAI_API_KEY': "sk-aapwfMWDuFE5XXXUr2BH",
'REVAI_API_KEY': "02ks6kFhEKjdhdure8474JJAJJ945958_h8P_DEKDNkK6DwNNNHU17aRtCw",
'MODEL_NAME': "code-davinci-002",
"speedSpeech": 170,
"speedPitch": 0.8,
"soundRate": 44100,
"contentType": "audio/x-raw",
"layout": "interleaved",
"format": "S16LE",
"channels": 1

Note that, all the API-key are not real. You need to generate your own key.

clsText2Voice.py (The python script that will convert text to voice)

	###############################################
	#### Written By: SATYAKI DE ####
	#### Written On: 27-Oct-2019 ####
	#### Modified On 28-Jan-2023 ####
	#### ####
	#### Objective: Main class converting ####
	#### text to voice using third-party API. ####
	###############################################

	import pyttsx3
	from clsConfigClient import clsConfigClient as cf

	class clsText2Voice:
	def __init__(self):
	self.speedSpeech = cf.conf['speedSpeech']
	self.speedPitch = cf.conf['speedPitch']

	def getAudio(self, srcString):
	try:
	speedSpeech = self.speedSpeech
	speedPitch = self.speedPitch

	engine = pyttsx3.init()

	# Set the speed of the speech (in words per minute)
	engine.setProperty('rate', speedSpeech)

	# Set the pitch of the speech (1.0 is default)
	engine.setProperty('pitch', speedPitch)

	# Converting to MP3
	engine.say(srcString)
	engine.runAndWait()

	return 0

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	return 1

view raw

clsText2Voice.py

hosted with ❤ by GitHub

Some of the important snippet will be as follows –

def getAudio(self, srcString):
    try:
        speedSpeech = self.speedSpeech
        speedPitch = self.speedPitch
        
        engine = pyttsx3.init()

        # Set the speed of the speech (in words per minute)
        engine.setProperty('rate', speedSpeech)

        # Set the pitch of the speech (1.0 is default)
        engine.setProperty('pitch', speedPitch)

        # Converting to MP3
        engine.say(srcString)
        engine.runAndWait()

        return 0

The code is a function that generates speech audio from a given string using the Pyttsx3 library in Python. The function sets the speech rate and pitch using the “speedSpeech” and “speedPitch” properties of the calling object, initializes the Pyttsx3 engine, sets the speech rate and pitch on the engine, speaks the given string, and waits for the speech to finish. The function returns 0 after the speech is finished.

clsChatEngine.py (This python script will invoke the ChatGPT OpenAI class to initiate the response of the queries in python.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 26-Dec-2022 ####
	#### Modified On 28-Jan-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### ChatGPT OpenAI class to initiate the ####
	#### response of the queries in python. ####
	#####################################################

	import os
	import openai
	import json
	from clsConfigClient import clsConfigClient as cf

	import sys
	import errno

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	###############################################
	### Global Section ###
	###############################################

	CODE_PATH=str(cf.conf['CODE_PATH'])
	MODEL_NAME=str(cf.conf['MODEL_NAME'])

	###############################################
	### End of Global Section ###
	###############################################

	class clsChatEngine:
	def __init__(self):
	self.OPENAI_API_KEY=str(cf.conf['OPENAI_API_KEY'])

	def findFromSJ(self, text):
	try:
	OPENAI_API_KEY = self.OPENAI_API_KEY

	# ChatGPT API_KEY
	openai.api_key = OPENAI_API_KEY

	print('22'*60)

	try:
	# Getting response from ChatGPT
	response = openai.Completion.create(
	engine=MODEL_NAME,
	prompt=text,
	max_tokens=64,
	top_p=1.0,
	n=3,
	temperature=0,
	frequency_penalty=0.0,
	presence_penalty=0.0,
	stop=["\"\"\""]
	)
	except IOError as e:
	if e.errno == errno.EPIPE:
	pass

	print('44'*60)
	res = response.choices[0].text

	return res

	except IOError as e:
	if e.errno == errno.EPIPE:
	pass

	except Exception as e:
	x = str(e)
	print(x)

	print('66'*60)

	return x

view raw

clsChatEngine.py

hosted with ❤ by GitHub

Key snippets from the above-script are as follows –

def findFromSJ(self, text):
      try:
          OPENAI_API_KEY = self.OPENAI_API_KEY

          # ChatGPT API_KEY
          openai.api_key = OPENAI_API_KEY

          print('22'*60)

          try:
              # Getting response from ChatGPT
              response = openai.Completion.create(
              engine=MODEL_NAME,
              prompt=text,
              max_tokens=64,
              top_p=1.0,
              n=3,
              temperature=0,
              frequency_penalty=0.0,
              presence_penalty=0.0,
              stop=["\"\"\""]
              )
          except IOError as e:
              if e.errno == errno.EPIPE:
                  pass

          print('44'*60)
          res = response.choices[0].text

          return res

      except IOError as e:
          if e.errno == errno.EPIPE:
              pass

      except Exception as e:
          x = str(e)
          print(x)

          print('66'*60)

          return x

The code is a function that uses OpenAI’s ChatGPT model to generate text based on a given prompt text. The function takes the text to be completed as input and uses an API key stored in the OPENAI_API_KEY property of the calling object to request OpenAI’s API. If the request is successful, the function returns the top completion generated by the model, as stored in the text field of the first item in the choices list of the API response.

The function includes error handling for IOError and Exception. If an IOError occurs, the function checks if the error number is errno.EPIPE and, if it is, returns without doing anything. If an Exception occurs, the function converts the error message to a string and prints it, then returns the string.

clsVoice2Text.py (This python script will invoke the Rev-AI class to initiate the transformation of audio into the text.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 26-Dec-2022 ####
	#### Modified On 28-Jan-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### Rev-AI class to initiate the transformation ####
	#### of audio into the text. ####
	#####################################################

	import pyaudio
	from rev_ai.models import MediaConfig
	from rev_ai.streamingclient import RevAiStreamingClient
	from six.moves import queue
	import ssl
	import json
	import pandas as p
	import clsMicrophoneStream as ms
	import clsL as cl
	from clsConfigClient import clsConfigClient as cf
	import datetime

	# Initiating Log class
	l = cl.clsL()

	# Bypassing SSL Authentication
	try:
	_create_unverified_https_context = ssl._create_unverified_context
	except AttributeError:
	# Legacy python that doesn't verify HTTPS certificates by default
	pass
	else:
	# Handle target environment that doesn't support HTTPS verification
	ssl._create_default_https_context = _create_unverified_https_context

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	######################################
	### Insert your access token here ####
	######################################
	debug_ind = 'Y'
	################################################################
	### Sampling rate of your microphone and desired chunk size ####
	################################################################

	class clsVoice2Text:
	def __init__(self):
	self.OPENAI_API_KEY=str(cf.conf['OPENAI_API_KEY'])
	self.rate = cf.conf['soundRate']

	def processVoice(self, var):
	try:
	OPENAI_API_KEY = self.OPENAI_API_KEY
	accessToken = cf.conf['REVAI_API_KEY']
	rate = self.rate
	chunk = int(rate/10)

	################################################################
	### Creates a media config with the settings set for a raw ####
	### microphone input ####
	################################################################

	sampleMC = MediaConfig('audio/x-raw', 'interleaved', 44100, 'S16LE', 1)

	streamclient = RevAiStreamingClient(accessToken, sampleMC)

	#####################################################################
	### Opens microphone input. The input will stop after a keyboard ####
	### interrupt. ####
	#####################################################################

	with ms.clsMicrophoneStream(rate, chunk) as stream:

	#####################################################################
	### Uses try method to enable users to manually close the stream ####
	#####################################################################

	try:
	response_gen = ''
	response = ''
	finalText = ''
	#########################################################################
	### Starts the server connection and thread sending microphone audio ####
	#########################################################################

	response_gen = streamclient.start(stream.generator())

	###################################################
	### Iterates through responses and prints them ####
	###################################################

	for response in response_gen:
	try:
	print('JSON:')
	print(response)

	r = json.loads(response)

	df = p.json_normalize(r["elements"])
	l.logr('1.df_' + var + '.csv', debug_ind, df, 'log')
	column_name = "confidence"

	if column_name in df.columns:
	print('DF:: ')
	print(df)

	finalText = "".join(df["value"])
	print("TEXT:")
	print(finalText)

	df = p.DataFrame()

	raise Exception

	except Exception as e:
	x = str(e)
	break

	streamclient.end()

	return finalText

	except Exception as e:
	x = str(e)
	#######################################
	### Ends the WebSocket connection. ####
	#######################################

	streamclient.end()

	return ''

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	streamclient.end()

	return x

view raw

clsVoice2Text.py

hosted with ❤ by GitHub

Here is the important snippet from the above code –

def processVoice(self, var):
      try:
          OPENAI_API_KEY = self.OPENAI_API_KEY
          accessToken = cf.conf['REVAI_API_KEY']
          rate = self.rate
          chunk = int(rate/10)

          ################################################################
          ### Creates a media config with the settings set for a raw  ####
          ### microphone input                                        ####
          ################################################################

          sampleMC = MediaConfig('audio/x-raw', 'interleaved', 44100, 'S16LE', 1)

          streamclient = RevAiStreamingClient(accessToken, sampleMC)

          #####################################################################
          ### Opens microphone input. The input will stop after a keyboard ####
          ### interrupt.                                                   ####
          #####################################################################

          with ms.clsMicrophoneStream(rate, chunk) as stream:

              #####################################################################
              ### Uses try method to enable users to manually close the stream ####
              #####################################################################

              try:
                  response_gen = ''
                  response = ''
                  finalText = ''
                  
                  ############################################
                  ### Starts the server connection        ####
                  ### and thread sending microphone audio #### 
                  ############################################

                  response_gen = streamclient.start(stream.generator())

                  ###################################################
                  ### Iterates through responses and prints them ####
                  ###################################################

                  for response in response_gen:
                      try:
                          print('JSON:')
                          print(response)

                          r = json.loads(response)

                          df = p.json_normalize(r["elements"])
                          l.logr('1.df_' + var + '.csv', debug_ind, df, 'log')
                          column_name = "confidence"

                          if column_name in df.columns:
                              print('DF:: ')
                              print(df)

                              finalText = "".join(df["value"])
                              print("TEXT:")
                              print(finalText)

                              df = p.DataFrame()

                              raise Exception

                      except Exception as e:
                          x = str(e)
                          break

                  streamclient.end()

                  return finalText

              except Exception as e:
                  x = str(e)
                  #######################################
                  ### Ends the WebSocket connection. ####
                  #######################################

                  streamclient.end()

                  return ''

      except Exception as e:
          x = str(e)
          print('Error: ', x)

          streamclient.end()

          return x

The code is a python function called processVoice() that processes a user’s voice input using the Rev.AI API. The function takes in one argument, “var,” which is not used in the code.

Let us understand the code –
- First, the function sets several variables, including the Rev.AI API access token, the sample rate, and the chunk size for the audio input.
- Then, it creates a media configuration object for raw microphone input.
- A RevAiStreamingClient object is created using the access token and the media configuration.
- The code opens the microphone input using a statement and the microphone stream class.
- Within the statement, the code starts the server connection and a thread that sends microphone audio to the server.
- The code then iterates through the responses from the server, normalizing the JSON response and storing the values in a pandas data-frame.
- If the “confidence” column exists in the data-frame, the code joins all the values to form the final text and raises an exception.
  - If there is an exception, the WebSocket connection is ended, and the final text is returned.
  - If there is any error, the WebSocket connection is also ended, and an empty string or the error message is returned.

clsMicrophoneStream.py (This python script invoke the rev_ai template to capture the chunk voice data & stream it to the service for text translation & return the response to app.)

	#####################################################
	#### Modified By: SATYAKI DE ####
	#### Modified On 28-Jan-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### rev_ai template to capture the chunk voice ####
	#### data & stream it to the service for text ####
	#### translation & return the response to app. ####
	#####################################################

	import pyaudio
	from rev_ai.models import MediaConfig
	from six.moves import queue

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	class clsMicrophoneStream(object):
	#############################################
	### Opens a recording stream as a ####
	### generator yielding the audio chunks. ####
	#############################################

	def __init__(self, rate, chunk):
	self._rate = rate
	self._chunk = chunk
	##################################################
	### Create a thread-safe buffer of audio data ####
	##################################################

	self._buff = queue.Queue()
	self.closed = True

	def __enter__(self):
	self._audio_interface = pyaudio.PyAudio()
	self._audio_stream = self._audio_interface.open(
	format=pyaudio.paInt16,

	#########################################################
	### The API currently only supports 1-channel (mono) ####
	### audio. ####
	#########################################################

	channels=1, rate=self._rate,
	input=True, frames_per_buffer=self._chunk,

	####################################################################
	### Run the audio stream asynchronously to fill the buffer ####
	### object. Run the audio stream asynchronously to fill the ####
	### buffer object. This is necessary so that the input device's ####
	### buffer doesn't overflow while the calling thread makes ####
	### network requests, etc. ####
	####################################################################

	stream_callback=self._fill_buffer,
	)

	self.closed = False

	return self

	def __exit__(self, type, value, traceback):
	self._audio_stream.stop_stream()
	self._audio_stream.close()
	self.closed = True

	###############################################################
	### Signal the generator to terminate so that the client's ####
	### streaming_recognize method will not block the process ####
	### termination. ####
	###############################################################

	self._buff.put(None)
	self._audio_interface.terminate()

	def _fill_buffer(self, in_data, frame_count, time_info, status_flags):

	##############################################################
	### Continuously collect data from the audio stream, into ####
	### the buffer. ####
	##############################################################

	self._buff.put(in_data)
	return None, pyaudio.paContinue

	def generator(self):
	while not self.closed:
	######################################################################
	### Use a blocking get() to ensure there's at least one chunk of ####
	### data, and stop iteration if the chunk is None, indicating the ####
	### end of the audio stream. ####
	######################################################################

	chunk = self._buff.get()
	if chunk is None:
	return
	data = [chunk]

	##########################################################
	### Now consume whatever other data's still buffered. ####
	##########################################################

	while True:
	try:
	chunk = self._buff.get(block=False)
	if chunk is None:
	return
	data.append(chunk)
	except queue.Empty:
	break

	yield b''.join(data)

view raw

clsMicrophoneStream.py

hosted with ❤ by GitHub

The key snippet from the above script are as follows –

def __enter__(self):
    self._audio_interface = pyaudio.PyAudio()
    self._audio_stream = self._audio_interface.open(
        format=pyaudio.paInt16,

        #########################################################
        ### The API currently only supports 1-channel (mono) ####
        ### audio.                                           ####
        #########################################################

        channels=1, rate=self._rate,
        input=True, frames_per_buffer=self._chunk,

        ####################################################################
        ### Run the audio stream asynchronously to fill the buffer      ####
        ### object. Run the audio stream asynchronously to fill the     ####
        ### buffer object. This is necessary so that the input device's ####
        ### buffer doesn't overflow while the calling thread makes      ####
        ### network requests, etc.                                      ####
        ####################################################################

        stream_callback=self._fill_buffer,
    )

    self.closed = False

    return self

This code is a part of a context manager class (clsMicrophoneStream) and implements the __enter__ method of the class. The method sets up a PyAudio object and opens an audio stream using the PyAudio object. The audio stream is configured to have the following properties:

Format: 16-bit integer (paInt16)
Channels: 1 (mono)
Rate: The rate specified in the instance of the ms.clsMicrophoneStream class.
Input: True, meaning the audio stream is an input stream, not an output stream.
Frames per buffer: The chunk specified in the instance of the ms.clsMicrophoneStream class.
Stream callback: The method self._fill_buffer will be called when the buffer needs more data.

The self.closed attribute is set to False to indicate that the stream is open. The method returns the instance of the class (self).

def __exit__(self, type, value, traceback):
    self._audio_stream.stop_stream()
    self._audio_stream.close()
    self.closed = True

    ###############################################################
    ### Signal the generator to terminate so that the client's ####
    ### streaming_recognize method will not block the process  ####
    ### termination.                                           ####
    ###############################################################

    self._buff.put(None)
    self._audio_interface.terminate()

The exit method implements the “exit” behavior of a Python context manager. It is automatically called when the context manager is exited using the statement.

The method stops and closes the audio stream, sets the closed attribute to True, and places None in the buffer. The terminate method of the PyAudio interface is then called to release any resources used by the audio stream.

def _fill_buffer(self, in_data, frame_count, time_info, status_flags):

    ##############################################################
    ### Continuously collect data from the audio stream, into ####
    ### the buffer.                                           ####
    ##############################################################

    self._buff.put(in_data)
    return None, pyaudio.paContinue

The _fill_buffer method is a callback function that runs asynchronously to continuously collect data from the audio stream and add it to the buffer.

The _fill_buffer method takes four arguments:

in_data: the raw audio data collected from the audio stream.
frame_count: the number of frames of audio data that was collected.
time_info: information about the timing of the audio data.
status_flags: flags that indicate the status of the audio stream.

The method adds the collected in_data to the buffer using the put method of the buffer object. It returns a tuple of None and pyaudio.paContinue to indicate that the audio stream should continue.

def generator(self):
    while not self.closed:
        ######################################################################
        ### Use a blocking get() to ensure there's at least one chunk of  ####
        ### data, and stop iteration if the chunk is None, indicating the ####
        ### end of the audio stream.                                      ####
        ######################################################################

        chunk = self._buff.get()
        if chunk is None:
            return
        data = [chunk]

        ##########################################################
        ### Now consume whatever other data's still buffered. ####
        ##########################################################

        while True:
            try:
                chunk = self._buff.get(block=False)
                if chunk is None:
                    return
                data.append(chunk)
            except queue.Empty:
                break

        yield b''.join(data)

The logic of the code “def generator(self):” is as follows:

The function generator is an infinite loop that runs until self.closed is True. Within the loop, it uses a blocking get() method of the buffer object (self._buff) to retrieve a chunk of audio data. If the retrieved chunk is None, it means the end of the audio stream has been reached, and the function returns.

If the retrieved chunk is not None, it appends it to the data list. The function then enters another inner loop that continues to retrieve chunks from the buffer using the non-blocking get() method until there are no more chunks left. Finally, the function yields the concatenated chunks of data as a single-byte string.

SJVoiceAssistant.py (Main calling python script)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 26-Dec-2022 ####
	#### Modified On 31-Jan-2023 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### multiple classes to initiate the ####
	#### AI-enabled personal assistant, which would ####
	#### display & answer the queries through voice. ####
	#####################################################

	import pyaudio
	from six.moves import queue
	import ssl
	import json
	import pandas as p
	import clsMicrophoneStream as ms
	import clsL as cl
	from clsConfigClient import clsConfigClient as cf
	import datetime
	import clsChatEngine as ce
	import clsText2Voice as tv
	import clsVoice2Text as vt
	#from signal import signal, SIGPIPE, SIG_DFL
	#signal(SIGPIPE,SIG_DFL)

	###################################################
	##### Adding the Instantiating Global classes #####
	###################################################
	x2 = ce.clsChatEngine()
	x3 = tv.clsText2Voice()
	x4 = vt.clsVoice2Text()
	# Initiating Log class
	l = cl.clsL()
	###################################################
	##### End of Global Classes #######
	###################################################
	# Bypassing SSL Authentication
	try:
	_create_unverified_https_context = ssl._create_unverified_context
	except AttributeError:
	# Legacy python that doesn't verify HTTPS certificates by default
	pass
	else:
	# Handle target environment that doesn't support HTTPS verification
	ssl._create_default_https_context = _create_unverified_https_context

	# Disbling Warning
	def warn(args, *kwargs):
	pass

	import warnings
	warnings.warn = warn

	######################################
	### Insert your access token here ####
	######################################
	debug_ind = 'Y'
	######################################
	#### Global Flag ########
	######################################

	def main():
	try:
	spFlag = True

	var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print(''120)
	print('Start Time: ' + str(var))
	print(''120)

	exitComment = 'THANKS.'

	while True:
	try:
	finalText = ''

	if spFlag == True:
	finalText = x4.processVoice(var)
	else:
	pass

	val = finalText.upper().strip()

	print('Main Return: ', val)
	print('Exit Call: ', exitComment)
	print('Length of Main Return: ', len(val))
	print('Length of Exit Call: ', len(exitComment))

	if val == exitComment:
	break
	elif finalText == '':
	spFlag = True
	else:
	print('spFlag::',spFlag)
	print('Inside: ', finalText)
	resVal = x2.findFromSJ(finalText)

	print('ChatGPT Response:: ')
	print(resVal)

	resAud = x3.getAudio(resVal)
	spFlag = False
	except Exception as e:
	pass

	var1 = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	print(''120)
	print('End Time: ' + str(var1))
	print('SJ Voice Assistant exited successfully!')
	print(''120)

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	if __name__ == "__main__":
	main()

view raw

SJVoiceAssistant.py

hosted with ❤ by GitHub

And, the key snippet from the above script –

def main():
    try:
        spFlag = True

        var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        print('*'*120)
        print('Start Time: ' + str(var))
        print('*'*120)

        exitComment = 'THANKS.'

        while True:
            try:
                finalText = ''

                if spFlag == True:
                    finalText = x4.processVoice(var)
                else:
                    pass

                val = finalText.upper().strip()

                print('Main Return: ', val)
                print('Exit Call: ', exitComment)
                print('Length of Main Return: ', len(val))
                print('Length of Exit Call: ', len(exitComment))

                if val == exitComment:
                    break
                elif finalText == '':
                    spFlag = True
                else:
                    print('spFlag::',spFlag)
                    print('Inside: ', finalText)
                    resVal = x2.findFromSJ(finalText)

                    print('ChatGPT Response:: ')
                    print(resVal)

                    resAud = x3.getAudio(resVal)
                    spFlag = False
            except Exception as e:
                pass

        var1 = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
        print('*'*120)
        print('End Time: ' + str(var1))
        print('SJ Voice Assistant exited successfully!')
        print('*'*120)

    except Exception as e:
        x = str(e)
        print('Error: ', x)

The code is a Python script that implements a voice-based chatbot (likely named “SJ Voice Assistant”). The code performs the following operations:

Initialize the string “exitComment” to “THANKS.” and set the “spFlag” to True.
Start an infinite loop until a specific condition breaks the loop.
In the loop, try to process the input voice with a function called “processVoice()” from an object “x4”. Store the result in “finalText.”
Convert “finalText” to upper case, remove leading/trailing whitespaces, and store it in “val.” Print “Main Return” and “Exit Call” with their length.
If “val” equals “exitComment,” break the loop. Suppose “finalText” is an empty string; set “spFlag” to True. Otherwise, perform further processing: a. Call the function “findFromSJ()” from an object “x2” with the input “finalText.” Store the result in “resVal.” b. Call the function “getAudio()” from an object “x3” with the input “resVal.” Store the result in “resAud.” Set “spFlag” to False.
If an exception occurs, catch it and pass (do nothing).
Finally the application will exit by displaying the following text – “SJ Voice Assistant exited successfully!”
If an exception occurs outside the loop, catch it and print the error message.

So, finally, we’ve done it.

I know that this post is relatively bigger than my earlier post. But, I think, you can get all the details once you go through it.

You will get the complete codebase in the following GitHub link.

I’ll bring some more exciting topics in the coming days from the Python verse. Please share & subscribe to my post & let me know your feedback.

Till then, Happy Avenging! 🙂

Documenting undocumented python scripts using Python-OpenAI

Posted on December 27, 2022December 27, 2022 by SatyakiDe in ai, api, Azure, call, cloud, code, Data Science, IoT, json, openai, Pandas, Python, video

Today, I will discuss another very impressive & innovative new AI, which is now operational in Python. We’ll document a dummy python code with no comment captured through OpenAI’s ChatGPT model. But before we start, don’t we see the demo first?

Demo

Great! Let us understand we can leverage this by writing a tiny snippet using this new AI model.

Architecture:

Let us understand the flow of events –

The above diagram represents the newly released OpenAI ChatGPT, where one needs to supply the code, which was missed to capture the logic earlier due to whatever may be the reasons. We need to provide these scripts (maybe in parts) as source code to be analyzed. Then it will use this new model & translate that into English-like language & capture the logic/comments for that specific snippet.

Python Packages:

Following are the python packages that are necessary to develop this brilliant use case –

pip install pandas
pip install openai

To know more, please click the below – “Continue Reading” link –

Continue reading →

Python performance improvement with 3.11 Version

Posted on October 30, 2022October 30, 2022 by SatyakiDe in call, cloud, code, computing, Crossplatform, function, Python

Today, we’ll share another performance improvement incorporating the latest Python 3.11 version. You can consider this significant advancement over the past versions. Last time, I posted for 3.7 in one of my earlier posts. But, we should diligently update everyone regarding the performance upgrade as it is slowly catching up with some of the finest programming languages.

But, before that, I want to share the latest stats of the machine where I tried these tests (As there is a change of system compared to last time).

Let us explore the base code –

##############################################
#### Written By: SATYAKI DE               ####
#### Written On: 06-May-2021              ####
#### Modified On: 30-Oct-2022             ####
####                                      ####
#### Objective: Main calling scripts for  ####
#### normal execution.                    ####
##############################################

from timeit import default_timer as timer

def vecCompute(sizeNum):
    try:
        total = 0
        for i in range(1, sizeNum):
            for j in range(1, sizeNum):
                total += i + j
        return total
    except Excception as e:
        x = str(e)
        print('Error: ', x)

        return 0


def main():

    start = timer()

    totalM = 0
    totalM = vecCompute(100000)

    print('The result is : ' + str(totalM))
    duration = timer() - start
    print('It took ' + str(duration) + ' seconds to compute')

if __name__ == '__main__':
    main()

And here is the outcome comparison between the 3.10 & 3.11 –

The above screenshot shows an improvement of 23% on an average compared to the previous version.

These performance stats are highly crucial. The result shows how Python is slowly emerging as the universal language for various kinds of work and is now targetting one of the vital threads, i.e., improvement of performance.

So, finally, we have done it.

I’ll bring some more exciting topic in the coming days from the Python verse.

Till then, Happy Avenging! 🙂

Note: All the data & scenario posted here are representational data & scenarios & available over the internet & for educational purpose only.

Scanned data extraction from a prefilled form using OpenCV & Python

Posted on September 18, 2022September 19, 2022 by SatyakiDe in api, Azure, call, cloud, code, Computer-Vision, Data Science, function, json, numpy, Open-CV, Pandas, pytesseract, Python, regular expression, select, Technology

This week we will discuss another important topic that many of us had in our mind. Today, we’ll try extracting the texts from scanned, formatted forms. This use case is instrumental when we need to process information prefilled by someone or some process.

To make things easier, I’ve packaged my entire solution & published that as a PyPi package after a long time. But, even before I start, why don’t we see the demo & then discuss it in detail?

Architecture:

Let us understand the architecture flow –

From the above diagram, one can understand the overall flow of this process. We’ll be using our second PyPi package, which will scan the source scanned copy of a formatted page & then tries to extract the relevant information.

Python Packages:

Following are the key python packages that we need apart from these dependent created packages & they are as follows –

cmake==3.22.1
dlib==19.19.0
imutils==0.5.3
jsonschema==4.4.0
numpy==1.23.2
oauthlib==3.1.1
opencv-contrib-python==4.6.0.66
opencv-contrib-python-headless==4.4.0.46
opencv-python==4.6.0.66
opencv-python-headless==4.5.5.62
pandas==1.4.3
python-dateutil==2.8.2
pytesseract==0.3.10
requests==2.27.1
requests-oauthlib==1.3.0

And the newly created package –

ReadingFilledForm==0.0.7

To know more about this, please visit the following PyPi link.

CODE:

clsConfigClient.py (This is the configuration class of the python script that will extract the text from the preformatted scanned copy.)

	################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 15-May-2020 ####
	#### Modified On: 18-Sep-2022 ####
	#### ####
	#### Objective: This script is a config ####
	#### file, contains all the keys for ####
	#### text extraction via image scanning. ####
	#### ####
	################################################

	import os
	import platform as pl

	my_dict = {}

	class clsConfigClient(object):
	Curr_Path = os.path.dirname(os.path.realpath(__file__))

	os_det = pl.system()
	if os_det == "Windows":
	sep = '\\'
	else:
	sep = '/'

	conf = {
	'APP_ID': 1,
	'ARCH_DIR': Curr_Path + sep + 'arch' + sep,
	'PROFILE_PATH': Curr_Path + sep + 'profile' + sep,
	'LOG_PATH': Curr_Path + sep + 'log' + sep,
	'REPORT_PATH': Curr_Path + sep + 'report',
	'SRC_PATH': Curr_Path + sep + 'data' + sep,
	'FINAL_PATH': Curr_Path + sep + 'Target' + sep,
	'IMAGE_PATH': Curr_Path + sep + 'Scans' + sep,
	'TEMPLATE_PATH': Curr_Path + sep + 'Template' + sep,
	'APP_DESC_1': 'Text Extraction from Video!',
	'DEBUG_IND': 'N',
	'INIT_PATH': Curr_Path,
	'SUBDIR': 'data',
	'WIDTH': 320,
	'HEIGHT': 320,
	'PADDING': 0.1,
	'SEP': sep,
	'MIN_CONFIDENCE':0.5,
	'GPU':–1,
	'FILE_NAME':'FilledUp.jpeg',
	'TEMPLATE_FILE_NAME':'Template.jpeg',
	'TITLE': "Text Reading!",
	'ORIG_TITLE': "Camera Source!",
	'LANG':"en",
	'OEM_VAL': 1,
	'PSM_VAL': 7,
	'DRAW_TAG': (0, 0, 255),
	'LAYER_DET':[
	"feature_fusion/Conv_7/Sigmoid",
	"feature_fusion/concat_3"],
	"CACHE_LIM": 1,
	'ASCII_RANGE': 128,
	'SUBTRACT_PARAM': (123.68, 116.78, 103.94),
	'MY_DICT': {
	"atrib_1": {"id": "FileNo", "bbox": (425, 60, 92, 34), "filter_keywords": tuple(["FILE", "DEPT"])},
	"atrib_2": {"id": "DeptNo", "bbox": (545, 60, 87, 40), "filter_keywords": tuple(["DEPT", "CLOCK"])},
	"atrib_3": {"id": "ClockNo", "bbox": (673, 60, 75, 36), "filter_keywords": tuple(["CLOCK","VCHR.","NO."])},
	"atrib_4": {"id": "VCHRNo", "bbox": (785, 60, 136, 40), "filter_keywords": tuple(["VCHR.","NO."])},
	"atrib_5": {"id": "DigitNo", "bbox": (949, 60, 50, 38), "filter_keywords": tuple(["VCHR.","NO.", "056"])},
	"atrib_6": {"id": "CompanyName", "bbox": (326, 140, 621, 187), "filter_keywords": tuple(["COMPANY","FILE"])},
	"atrib_7": {"id": "StartDate", "bbox": (1264, 143, 539, 44), "filter_keywords": tuple(["Period", "Beginning:"])},
	"atrib_8": {"id": "EndDate", "bbox": (1264, 193, 539, 44), "filter_keywords": tuple(["Period", "Ending:"])},
	"atrib_9": {"id": "PayDate", "bbox": (1264, 233, 539, 44), "filter_keywords": tuple(["Pay", "Date:"])},
	}
	}

view raw

clsConfigClient.py

hosted with ❤ by GitHub

The only important part of these configurations are the following –

'MY_DICT': {
            "atrib_1": {"id": "FileNo", "bbox": (425, 60, 92, 34), "filter_keywords": tuple(["FILE", "DEPT"])},
            "atrib_2": {"id": "DeptNo", "bbox": (545, 60, 87, 40), "filter_keywords": tuple(["DEPT", "CLOCK"])},
            "atrib_3": {"id": "ClockNo", "bbox": (673, 60, 75, 36), "filter_keywords": tuple(["CLOCK","VCHR.","NO."])},
            "atrib_4": {"id": "VCHRNo", "bbox": (785, 60, 136, 40), "filter_keywords": tuple(["VCHR.","NO."])},
            "atrib_5": {"id": "DigitNo", "bbox": (949, 60, 50, 38), "filter_keywords": tuple(["VCHR.","NO.", "056"])},
            "atrib_6": {"id": "CompanyName", "bbox": (326, 140, 621, 187), "filter_keywords": tuple(["COMPANY","FILE"])},
            "atrib_7": {"id": "StartDate", "bbox": (1264, 143, 539, 44), "filter_keywords": tuple(["Period", "Beginning:"])},
            "atrib_8": {"id": "EndDate", "bbox": (1264, 193, 539, 44), "filter_keywords": tuple(["Period", "Ending:"])},
            "atrib_9": {"id": "PayDate", "bbox": (1264, 233, 539, 44), "filter_keywords": tuple(["Pay", "Date:"])},
      }

Let us understand this part, as it is very critical for this entire package.

We need to define the areas in terms of pixel position, which we need to extract. Hence, we follow the following pattern –

"atrib_": {"id": , "bbox": (x-Coordinates, y-Coordinates, Width, Height), "filter_keywords": tuple(["Mention the overlapping printed text that you don't want to capture. Make sure you are following the exact Case to proper detection."])}

You can easily get the individual intended text position by using any Photo editor.

Still not clear how to select?

Let’s watch the next video –

**How to fetch the extracted location pixel metadata – Demo**

The above demo should explain what we are trying to achieve. Also, you need to understand that if your two values are extremely close, then we’re taking both the non-desired labels & put them under the filter keywords to ensure extracting the correct values.

For example, on the top left side, where the values are very close, we’re putting both closed labels as filter keywords. One such example is as follows –

"filter_keywords": tuple(["FILE", "DEPT"])

The same logic applies to the other labels as well.

readingFormLib.py (This is the main calling python script that will extract the text from the preformatted scanned copy.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 22-Jul-2022 ####
	#### Modified On 18-Sep-2022 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### clsReadForm class to initiate ####
	#### the reading capability in real-time ####
	#### & display text from a formatted forms. ####
	#####################################################

	# We keep the setup code in a different class as shown below.
	from ReadingFilledForm import clsReadForm as rf

	from clsConfigClient import clsConfigClient as cf

	import datetime
	import logging

	###############################################
	### Global Section ###
	###############################################
	# Instantiating all the main class
	scannedImagePath = str(cf.conf['IMAGE_PATH']) + str(cf.conf['FILE_NAME'])
	templatePath = str(cf.conf['TEMPLATE_PATH']) + str(cf.conf['TEMPLATE_FILE_NAME'])

	x1 = rf.clsReadForm(scannedImagePath, templatePath)

	###############################################
	### End of Global Section ###
	###############################################

	def main():
	try:
	# Other useful variables
	debugInd = 'Y'
	var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	var1 = datetime.datetime.now()

	print('Start Time: ', str(var))
	# End of useful variables

	# Initiating Log Class
	general_log_path = str(cf.conf['LOG_PATH'])

	# Enabling Logging Info
	logging.basicConfig(filename=general_log_path + 'readingForm.log', level=logging.INFO)

	print('Started extracting text from formatted forms!')

	# Getting the dictionary
	my_dict = cf.conf['MY_DICT']

	# Execute all the pass
	r1 = x1.startProcess(debugInd, var, my_dict)

	if (r1 == 0):
	print('Successfully extracted text from the formatted forms!')
	else:
	print('Failed to extract the text from the formatted forms!')

	var2 = datetime.datetime.now()

	c = var2 – var1
	minutes = c.total_seconds() / 60
	print('Total difference in minutes: ', str(minutes))

	print('End Time: ', str(var1))

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	if __name__ == "__main__":
	main()

view raw

readingFormLib.py

hosted with ❤ by GitHub

Key snippets from the above script –

# We keep the setup code in a different class as shown below.
from ReadingFilledForm import clsReadForm as rf

from clsConfigClient import clsConfigClient as cf

The above lines import the newly created PyPi package into the memory.

###############################################
###           Global Section                ###
###############################################
# Instantiating all the main class
scannedImagePath = str(cf.conf['IMAGE_PATH']) + str(cf.conf['FILE_NAME'])
templatePath = str(cf.conf['TEMPLATE_PATH']) + str(cf.conf['TEMPLATE_FILE_NAME'])

x1 = rf.clsReadForm(scannedImagePath, templatePath)

###############################################
###    End of Global Section                ###
###############################################

Now, the application is fetching both the template copy & the intended scanned copy & load them into the memory.

# Getting the dictionary
my_dict = cf.conf['MY_DICT']

After this, the application will try to extract the focus area dictionary, indicating the areas of particular interest.

# Execute all the pass
r1 = x1.startProcess(debugInd, var, my_dict)

Finally, pass it inside the new package to get the correct outcome.

FOLDER STRUCTURE:

Here is the folder structure that contains all the files & directories in MAC O/S –

Similar structures are present in the Windows environment as well.

You will get the complete calling codebase in the following GitHub link.

I’ll bring some more exciting topics in the coming days from the Python verse. Please share & subscribe to my post & let me know your feedback.

Till then, Happy Avenging! 🙂

Note: All the data & scenarios posted here are representational data & scenarios & available over the internet & for educational purposes only. There is always room for improvement & especially in the prediction quality.

Realtime reading from a Streaming using Computer Vision

Posted on July 26, 2022 by SatyakiDe in api, Azure, call, cloud, code, Computer-Vision, computing, Crossplatform, Data Science, exposure, extends, function, gui, IoT, json, Keras, machine-learning, matplotlib, mobile, Model, numpy, objects, Open-CV, Pandas, pytesseract, Python, Real-time, snippet, video

This week we’re going to extend one of our earlier posts & trying to read an entire text from streaming using computer vision. If you want to view the previous post, please click the following link.

But, before we proceed, why don’t we view the demo first?

Demo

Architecture:

Let us understand the architecture flow –

The above diagram shows that the application, which uses the Open-CV, analyzes individual frames from the source & extracts the complete text within the video & displays it on top of the target screen besides prints the same in the console.

Python Packages:

pip install imutils==0.5.4
pip install matplotlib==3.5.2
pip install numpy==1.21.6
pip install opencv-contrib-python==4.6.0.66
pip install opencv-contrib-python-headless==4.6.0.66
pip install opencv-python==4.6.0.66
pip install opencv-python-headless==4.6.0.66
pip install pandas==1.3.5
pip install Pillow==9.1.1
pip install pytesseract==0.3.9
pip install python-dateutil==2.8.2

CODE:

clsReadingTextFromStream.py (This is the main class of python script that will extract the text from the WebCAM streaming in real-time.)

	##################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 22-Jul-2022 ####
	#### Modified On 25-Jul-2022 ####
	#### ####
	#### Objective: This is the main class of ####
	#### python script that will invoke the ####
	#### extraction of texts from a WebCAM. ####
	#### ####
	##################################################

	# Importing necessary packages
	from clsConfig import clsConfig as cf

	from imutils.object_detection import non_max_suppression
	import numpy as np
	import pytesseract
	import imutils
	import time
	import cv2
	import time

	###############################################
	### Global Section ###
	###############################################

	# Two output layer names for the text detector model

	lNames = cf.conf['LAYER_DET']

	# Tesseract OCR text param values

	strVal = "-l " + str(cf.conf['LANG']) + " –oem " + str(cf.conf['OEM_VAL']) + " –psm " + str(cf.conf['PSM_VAL']) + ""
	config = (strVal)

	###############################################
	### End of Global Section ###
	###############################################

	class clsReadingTextFromStream:
	def __init__(self):
	self.sep = str(cf.conf['SEP'])
	self.Curr_Path = str(cf.conf['INIT_PATH'])
	self.CacheL = int(cf.conf['CACHE_LIM'])
	self.modelPath = str(cf.conf['MODEL_PATH']) + str(cf.conf['MODEL_FILE_NAME'])
	self.minConf = float(cf.conf['MIN_CONFIDENCE'])
	self.wt = int(cf.conf['WIDTH'])
	self.ht = int(cf.conf['HEIGHT'])
	self.pad = float(cf.conf['PADDING'])
	self.title = str(cf.conf['TITLE'])
	self.Otitle = str(cf.conf['ORIG_TITLE'])
	self.drawTag = cf.conf['DRAW_TAG']
	self.aRange = int(cf.conf['ASCII_RANGE'])
	self.sParam = cf.conf['SUBTRACT_PARAM']

	def findBoundBox(self, boxes, res, rW, rH, orig, origW, origH, pad):
	try:
	# Loop over the bounding boxes
	for (spX, spY, epX, epY) in boxes:
	# Scale the bounding box coordinates based on the respective
	# ratios
	spX = int(spX * rW)
	spY = int(spY * rH)
	epX = int(epX * rW)
	epY = int(epY * rH)

	# To obtain a better OCR of the text we can potentially
	# apply a bit of padding surrounding the bounding box.
	# And, computing the deltas in both the x and y directions
	dX = int((epX – spX) * pad)
	dY = int((epY – spY) * pad)

	# Apply padding to each side of the bounding box, respectively
	spX = max(0, spX – dX)
	spY = max(0, spY – dY)
	epX = min(origW, epX + (dX * 2))
	epY = min(origH, epY + (dY * 2))

	# Extract the actual padded ROI
	roi = orig[spY:epY, spX:epX]

	# Choose the proper OCR Config
	text = pytesseract.image_to_string(roi, config=config)

	# Add the bounding box coordinates and OCR'd text to the list
	# of results
	res.append(((spX, spY, epX, epY), text))

	# Sort the results bounding box coordinates from top to bottom
	res = sorted(res, key=lambda r:r[0][1])

	return res
	except Exception as e:
	x = str(e)
	print(x)

	return res

	def predictText(self, imgScore, imgGeo):
	try:
	minConf = self.minConf

	# Initializing the bounding box rectangles & confidence score by
	# extracting the rows & columns from the imgScore volume.
	(numRows, numCols) = imgScore.shape[2:4]
	rects = []
	confScore = []

	for y in range(0, numRows):
	# Extract the imgScore probabilities to derive potential
	# bounding box coordinates that surround text
	imgScoreData = imgScore[0, 0, y]
	xVal0 = imgGeo[0, 0, y]
	xVal1 = imgGeo[0, 1, y]
	xVal2 = imgGeo[0, 2, y]
	xVal3 = imgGeo[0, 3, y]
	anglesData = imgGeo[0, 4, y]

	for x in range(0, numCols):
	# If our score does not have sufficient probability,
	# ignore it
	if imgScoreData[x] < minConf:
	continue

	# Compute the offset factor as our resulting feature
	# maps will be 4x smaller than the input frame
	(offX, offY) = (x * 4.0, y * 4.0)

	# Extract the rotation angle for the prediction and
	# then compute the sin and cosine
	angle = anglesData[x]
	cos = np.cos(angle)
	sin = np.sin(angle)

	# Derive the width and height of the bounding box from
	# imgGeo
	h = xVal0[x] + xVal2[x]
	w = xVal1[x] + xVal3[x]

	# Compute both the starting and ending (x, y)-coordinates
	# for the text prediction bounding box
	epX = int(offX + (cos * xVal1[x]) + (sin * xVal2[x]))
	epY = int(offY – (sin * xVal1[x]) + (cos * xVal2[x]))
	spX = int(epX – w)
	spY = int(epY – h)

	# Adding bounding box coordinates and probability score
	# to the respective lists
	rects.append((spX, spY, epX, epY))
	confScore.append(imgScoreData[x])

	# return a tuple of the bounding boxes and associated confScore
	return (rects, confScore)

	except Exception as e:
	x = str(e)
	print(x)

	rects = []
	confScore = []

	return (rects, confScore)

	def processStream(self, debugInd, var):
	try:
	sep = self.sep
	Curr_Path = self.Curr_Path
	CacheL = self.CacheL
	modelPath = self.modelPath
	minConf = self.minConf
	wt = self.wt
	ht = self.ht
	pad = self.pad
	title = self.title
	Otitle = self.Otitle
	drawTag = self.drawTag
	aRange = self.aRange
	sParam = self.sParam

	val = 0

	# Initialize the video stream and allow the camera sensor to warm up
	print("[INFO] Starting video stream…")
	cap = cv2.VideoCapture(0)

	# Loading the pre-trained text detector
	print("[INFO] Loading Text Detector…")
	net = cv2.dnn.readNet(modelPath)

	# Loop over the frames from the video stream
	while True:
	try:
	# Grab the frame from our video stream and resize it
	success, frame = cap.read()

	orig = frame.copy()
	(origH, origW) = frame.shape[:2]

	# Setting new width and height and then determine the ratio in change
	# for both the width and height
	(newW, newH) = (wt, ht)
	rW = origW / float(newW)
	rH = origH / float(newH)

	# Resize the frame and grab the new frame dimensions
	frame = cv2.resize(frame, (newW, newH))
	(H, W) = frame.shape[:2]

	# Construct a blob from the frame and then perform a forward pass of
	# the model to obtain the two output layer sets
	blob = cv2.dnn.blobFromImage(frame, 1.0, (W, H), sParam, swapRB=True, crop=False)
	net.setInput(blob)
	(confScore, imgGeo) = net.forward(lNames)

	# Decode the predictions, then apply non-maxima suppression to
	# suppress weak, overlapping bounding boxes
	(rects, confidences) = self.predictText(confScore, imgGeo)
	boxes = non_max_suppression(np.array(rects), probs=confidences)

	# Initialize the list of results
	res = []

	# Getting BoundingBox boundaries
	res = self.findBoundBox(boxes, res, rW, rH, orig, origW, origH, pad)

	for ((spX, spY, epX, epY), text) in res:
	# Display the text OCR by using Tesseract APIs
	print("Reading Text::")
	print("=" *60)
	print(text)
	print("=" *60)

	# Removing the non-ASCII text so it can draw the text on the frame
	# using OpenCV, then draw the text and a bounding box surrounding
	# the text region of the input frame
	text = "".join([c if ord(c) < aRange else "" for c in text]).strip()
	output = orig.copy()

	cv2.rectangle(output, (spX, spY), (epX, epY), drawTag, 2)
	cv2.putText(output, text, (spX, spY – 20), cv2.FONT_HERSHEY_SIMPLEX, 1.2, drawTag, 3)

	# Show the output frame
	cv2.imshow(title, output)
	#cv2.imshow(Otitle, frame)

	# If the `q` key was pressed, break from the loop
	if cv2.waitKey(1) == ord('q'):
	break

	val = 0

	except Exception as e:
	x = str(e)
	print(x)

	val = 1

	# Performing cleanup at the end
	cap.release()
	cv2.destroyAllWindows()

	return val
	except Exception as e:
	x = str(e)
	print('Error:', x)

	return 1

view raw

clsReadingTextFromStream.py

hosted with ❤ by GitHub

Please find the key snippet from the above script –

# Two output layer names for the text detector model

lNames = cf.conf['LAYER_DET']

# Tesseract OCR text param values

strVal = "-l " + str(cf.conf['LANG']) + " --oem " + str(cf.conf['OEM_VAL']) + " --psm " + str(cf.conf['PSM_VAL']) + ""
config = (strVal)

The first line contains the two output layers’ names for the text detector model. Among them, the first one indicates the outcome possibilities & the second one use to derive the bounding box coordinates of the predicted text.

The second line contains various options for the tesseract APIs. You need to understand the opportunities in detail to make them work. These are the essential options for our use case –

Language – The intended language, for example, English, Spanish, Hindi, Bengali, etc.
OEM flag – In this case, the application will use 4 to indicate LSTM neural net model for OCR.
OEM Value – In this case, the selected value is 7, indicating that the application treats the ROI as a single line of text.

For more details, please refer to the config file.

print("[INFO] Loading Text Detector...")
net = cv2.dnn.readNet(modelPath)

The above lines bring the already created model & load it to memory for evaluation.

# Setting new width and height and then determine the ratio in change
# for both the width and height
(newW, newH) = (wt, ht)
rW = origW / float(newW)
rH = origH / float(newH)

# Resize the frame and grab the new frame dimensions
frame = cv2.resize(frame, (newW, newH))
(H, W) = frame.shape[:2]

# Construct a blob from the frame and then perform a forward pass of
# the model to obtain the two output layer sets
blob = cv2.dnn.blobFromImage(frame, 1.0, (W, H), sParam, swapRB=True, crop=False)
net.setInput(blob)
(confScore, imgGeo) = net.forward(lNames)

# Decode the predictions, then apply non-maxima suppression to
# suppress weak, overlapping bounding boxes
(rects, confidences) = self.predictText(confScore, imgGeo)
boxes = non_max_suppression(np.array(rects), probs=confidences)

The above lines are more of preparing individual frames to get the bounding box by resizing the height & width followed by a forward pass of the model to obtain two output layer sets. And then apply the non-maxima suppression to remove the weak, overlapping bounding box by interpreting the prediction. In short, this will identify the potential text region & put the bounding box surrounding it.

# Initialize the list of results
res = []

# Getting BoundingBox boundaries
res = self.findBoundBox(boxes, res, rW, rH, orig, origW, origH, pad)

The above function will create the bounding box surrounding the predicted text regions. Also, we will capture the expected text inside the result variable.

for (spX, spY, epX, epY) in boxes:
  # Scale the bounding box coordinates based on the respective
  # ratios
  spX = int(spX * rW)
  spY = int(spY * rH)
  epX = int(epX * rW)
  epY = int(epY * rH)

  # To obtain a better OCR of the text we can potentially
  # apply a bit of padding surrounding the bounding box.
  # And, computing the deltas in both the x and y directions
  dX = int((epX - spX) * pad)
  dY = int((epY - spY) * pad)

  # Apply padding to each side of the bounding box, respectively
  spX = max(0, spX - dX)
  spY = max(0, spY - dY)
  epX = min(origW, epX + (dX * 2))
  epY = min(origH, epY + (dY * 2))

  # Extract the actual padded ROI
  roi = orig[spY:epY, spX:epX]

Now, the application will scale the bounding boxes based on the previously computed ratio for actual text recognition. In this process, the application also padded the bounding boxes & then extracted the padded region of interest.

# Choose the proper OCR Config
text = pytesseract.image_to_string(roi, config=config)

# Add the bounding box coordinates and OCR'd text to the list
# of results
res.append(((spX, spY, epX, epY), text))

Using OCR options, the application extracts the text within the video frame & adds that to the res list.

# Sort the results bounding box coordinates from top to bottom
res = sorted(res, key=lambda r:r[0][1])

It then sends a sorted output to the primary calling functions.

for ((spX, spY, epX, epY), text) in res:
  # Display the text OCR by using Tesseract APIs
  print("Reading Text::")
  print("=" *60)
  print(text)
  print("=" *60)

  # Removing the non-ASCII text so it can draw the text on the frame
  # using OpenCV, then draw the text and a bounding box surrounding
  # the text region of the input frame
  text = "".join([c if ord(c) < aRange else "" for c in text]).strip()
  output = orig.copy()

  cv2.rectangle(output, (spX, spY), (epX, epY), drawTag, 2)
  cv2.putText(output, text, (spX, spY - 20), cv2.FONT_HERSHEY_SIMPLEX, 1.2, drawTag, 3)

  # Show the output frame
  cv2.imshow(title, output)

Finally, it fetches the potential text region along with the text & then prints on top of the source video. Also, it removed some non-printable characters during this time to avoid any cryptic texts.

readingVideo.py (Main calling script.)

	#####################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 22-Jul-2022 ####
	#### Modified On 25-Jul-2022 ####
	#### ####
	#### Objective: This is the main calling ####
	#### python script that will invoke the ####
	#### clsReadingTextFromStream class to initiate ####
	#### the reading capability in real-time ####
	#### & display text via Web-CAM. ####
	#####################################################

	# We keep the setup code in a different class as shown below.
	import clsReadingTextFromStream as rtfs

	from clsConfig import clsConfig as cf

	import datetime
	import logging

	###############################################
	### Global Section ###
	###############################################
	# Instantiating all the main class

	x1 = rtfs.clsReadingTextFromStream()

	###############################################
	### End of Global Section ###
	###############################################

	def main():
	try:
	# Other useful variables
	debugInd = 'Y'
	var = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
	var1 = datetime.datetime.now()

	print('Start Time: ', str(var))
	# End of useful variables

	# Initiating Log Class
	general_log_path = str(cf.conf['LOG_PATH'])

	# Enabling Logging Info
	logging.basicConfig(filename=general_log_path + 'readingTextFromVideo.log', level=logging.INFO)

	print('Started reading text from videos!')

	# Execute all the pass
	r1 = x1.processStream(debugInd, var)

	if (r1 == 0):
	print('Successfully read text from the Live Stream!')
	else:
	print('Failed to read text from the Live Stream!')

	var2 = datetime.datetime.now()

	c = var2 – var1
	minutes = c.total_seconds() / 60
	print('Total difference in minutes: ', str(minutes))

	print('End Time: ', str(var1))

	except Exception as e:
	x = str(e)
	print('Error: ', x)

	if __name__ == "__main__":
	main()

view raw

readingVideo.py

hosted with ❤ by GitHub

Please find the key snippet –

# Instantiating all the main class

x1 = rtfs.clsReadingTextFromStream()

# Execute all the pass
r1 = x1.processStream(debugInd, var)

if (r1 == 0):
    print('Successfully read text from the Live Stream!')
else:
    print('Failed to read text from the Live Stream!')

The above lines instantiate the main calling class & then invoke the function to get the desired extracted text from the live streaming video if that is successful.

FOLDER STRUCTURE:

Here is the folder structure that contains all the files & directories in MAC O/S –

You will get the complete codebase in the following Github link.

Unfortunately, I cannot upload the model due to it’s size. I will share on the need basis.

I’ll bring some more exciting topic in the coming days from the Python verse. Please share & subscribe my post & let me know your feedback.

Till then, Happy Avenging! 🙂

Note: All the data & scenario posted here are representational data & scenarios & available over the internet & for educational purpose only. Some of the images (except my photo) that we’ve used are available over the net. We don’t claim the ownership of these images. There is an always room for improvement & especially the prediction quality.

Real-time augmented reality (AR) using Python-based Computer Vision

Posted on June 25, 2022 by SatyakiDe in analytic function, api, audio, Azure, break, call, cloud, code, combining, Computer-Vision, computing, Crossplatform, Data Science, design, exposure, features, function, gui, human, IoT, machine-learning, matplotlib, mobile, Model, numpy, objects, Open-CV, Pandas, pygame, Python, Real-time, return, snippet, Technology, video

Hi Team,

Today, I’m going to discuss another Computer Vision installment. I’ll discuss how to implement Augmented Reality using Open-CV Computer Vision with full audio. We will be using part of a Bengali OTT Series called “Feludar Goendagiri” entirely for educational purposes & also as a tribute to the great legendary director, late Satyajit Roy. To know more about him, please click the following link.

Why don’t we see the demo first before jumping into the technical details?

Demo

Architecture:

Let us understand the architecture –

The above diagram shows that the application, which uses the Open-CV, analyzes individual frames from the source & blends that with the video trailer. Finally, it creates another video by correctly mixing the source audio.

Python Packages:

Following are the python packages that are necessary to develop this brilliant use case –

pip install opencv-python
pip install pygame

CODE:

clsAugmentedReality.py (This is the main class of python script that will embed the source video with the WebCAM streams in real-time.)

	##################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 20-Jun-2022 ####
	#### Modified On 25-Jun-2022 ####
	#### ####
	#### Objective: This is the main class of ####
	#### python script that will embed the source ####
	#### video with the WebCAM streams in ####
	#### real-time. ####
	##################################################

	# Importing necessary packages
	import numpy as np
	import cv2

	from clsConfig import clsConfig as cf

	# Initialize our cached reference points
	CACHED_REF_PTS = None

	class clsAugmentedReality:
	def __init__(self):
	self.TOP_LEFT_X = int(cf.conf['TOP_LEFT_X'])
	self.TOP_LEFT_Y = int(cf.conf['TOP_LEFT_Y'])
	self.TOP_RIGHT_X = int(cf.conf['TOP_RIGHT_X'])
	self.TOP_RIGHT_Y = int(cf.conf['TOP_RIGHT_Y'])
	self.BOTTOM_RIGHT_X = int(cf.conf['BOTTOM_RIGHT_X'])
	self.BOTTOM_RIGHT_Y = int(cf.conf['BOTTOM_RIGHT_Y'])
	self.BOTTOM_LEFT_X = int(cf.conf['BOTTOM_LEFT_X'])
	self.BOTTOM_LEFT_Y = int(cf.conf['BOTTOM_LEFT_Y'])

	def getWarpImages(self, frame, source, cornerIDs, arucoDict, arucoParams, zoomFlag, useCache=False):
	try:
	# Assigning values
	TOP_LEFT_X = self.TOP_LEFT_X
	TOP_LEFT_Y = self.TOP_LEFT_Y
	TOP_RIGHT_X = self.TOP_RIGHT_X
	TOP_RIGHT_Y = self.TOP_RIGHT_Y
	BOTTOM_RIGHT_X = self.BOTTOM_RIGHT_X
	BOTTOM_RIGHT_Y = self.BOTTOM_RIGHT_Y
	BOTTOM_LEFT_X = self.BOTTOM_LEFT_X
	BOTTOM_LEFT_Y = self.BOTTOM_LEFT_Y

	# Grab a reference to our cached reference points
	global CACHED_REF_PTS

	if source is None:
	raise

	# Grab the width and height of the frame and source image,
	# respectively
	# Extracting Frame from Camera
	# Exracting Source from Video
	(imgH, imgW) = frame.shape[:2]
	(srcH, srcW) = source.shape[:2]

	# Detect Aruco markers in the input frame
	(corners, ids, rejected) = cv2.aruco.detectMarkers(frame, arucoDict, parameters=arucoParams)

	print('Ids: ', str(ids))
	print('Rejected: ', str(rejected))

	# if we did not find our four ArUco markers, initialize an
	# empty IDs list, otherwise flatten the ID list
	print('Detecting Corners: ', str(len(corners)))
	ids = np.array([]) if len(corners) != 4 else ids.flatten()

	# Initialize our list of reference points
	refPts = []
	refPtTL1 = []

	# Loop over the IDs of the ArUco markers in Top-Left, Top-Right,
	# Bottom-Right, and Bottom-Left order
	for i in cornerIDs:
	# Grab the index of the corner with the current ID
	j = np.squeeze(np.where(ids == i))

	# If we receive an empty list instead of an integer index,
	# then we could not find the marker with the current ID
	if j.size == 0:
	continue

	# Otherwise, append the corner (x, y)-coordinates to our list
	# of reference points
	corner = np.squeeze(corners[j])
	refPts.append(corner)

	# Check to see if we failed to find the four ArUco markers
	if len(refPts) != 4:
	# If we are allowed to use cached reference points, fall
	# back on them
	if useCache and CACHED_REF_PTS is not None:
	refPts = CACHED_REF_PTS

	# Otherwise, we cannot use the cache and/or there are no
	# previous cached reference points, so return early
	else:
	return None

	# If we are allowed to use cached reference points, then update
	# the cache with the current set
	if useCache:
	CACHED_REF_PTS = refPts

	# Unpack our Aruco reference points and use the reference points
	# to define the Destination transform matrix, making sure the
	# points are specified in Top-Left, Top-Right, Bottom-Right, and
	# Bottom-Left order
	(refPtTL, refPtTR, refPtBR, refPtBL) = refPts
	dstMat = [refPtTL[0], refPtTR[1], refPtBR[2], refPtBL[3]]
	dstMat = np.array(dstMat)

	# For zoom option recalculating all the 4 points
	refPtTL1_L_X = refPtTL[0][0]–TOP_LEFT_X
	refPtTL1_L_Y = refPtTL[0][1]–TOP_LEFT_Y

	refPtTL1.append((refPtTL1_L_X,refPtTL1_L_Y))

	refPtTL1_R_X = refPtTL[1][0]+TOP_RIGHT_X
	refPtTL1_R_Y = refPtTL[1][1]+TOP_RIGHT_Y

	refPtTL1.append((refPtTL1_R_X,refPtTL1_R_Y))

	refPtTD1_L_X = refPtTL[2][0]+BOTTOM_RIGHT_X
	refPtTD1_L_Y = refPtTL[2][1]+BOTTOM_RIGHT_Y

	refPtTL1.append((refPtTD1_L_X,refPtTD1_L_Y))

	refPtTD1_R_X = refPtTL[3][0]–BOTTOM_LEFT_X
	refPtTD1_R_Y = refPtTL[3][1]+BOTTOM_LEFT_Y

	refPtTL1.append((refPtTD1_R_X,refPtTD1_R_Y))

	dstMatMod = [refPtTL1[0], refPtTL1[1], refPtTL1[2], refPtTL1[3]]
	dstMatMod = np.array(dstMatMod)

	# Define the transform matrix for the source image in Top-Left,
	# Top-Right, Bottom-Right, and Bottom-Left order
	srcMat = np.array([[0, 0], [srcW, 0], [srcW, srcH], [0, srcH]])

	# Compute the homography matrix and then warp the source image to
	# the destination based on the homography depending upon the
	# zoom flag
	if zoomFlag == 1:
	(H, _) = cv2.findHomography(srcMat, dstMat)
	else:
	(H, _) = cv2.findHomography(srcMat, dstMatMod)

	warped = cv2.warpPerspective(source, H, (imgW, imgH))

	# Construct a mask for the source image now that the perspective
	# warp has taken place (we'll need this mask to copy the source
	# image into the destination)
	mask = np.zeros((imgH, imgW), dtype="uint8")
	if zoomFlag == 1:
	cv2.fillConvexPoly(mask, dstMat.astype("int32"), (255, 255, 255), cv2.LINE_AA)
	else:
	cv2.fillConvexPoly(mask, dstMatMod.astype("int32"), (255, 255, 255), cv2.LINE_AA)

	# This optional step will give the source image a black
	# border surrounding it when applied to the source image, you
	# can apply a dilation operation
	rect = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 3))
	mask = cv2.dilate(mask, rect, iterations=2)

	# Create a three channel version of the mask by stacking it
	# depth-wise, such that we can copy the warped source image
	# into the input image
	maskScaled = mask.copy() / 255.0
	maskScaled = np.dstack([maskScaled] * 3)

	# Copy the warped source image into the input image by
	# (1) Multiplying the warped image and masked together,
	# (2) Then multiplying the original input image with the
	# mask (giving more weight to the input where there
	# are not masked pixels), and
	# (3) Adding the resulting multiplications together
	warpedMultiplied = cv2.multiply(warped.astype("float"), maskScaled)
	imageMultiplied = cv2.multiply(frame.astype(float), 1.0 – maskScaled)
	output = cv2.add(warpedMultiplied, imageMultiplied)
	output = output.astype("uint8")

	# Return the output frame to the calling function
	return output

	except Exception as e:

	# Delibarately raising the issue
	# That way the control goes to main calling methods
	# exception section
	raise

view raw

clsAugmentedReality.py

hosted with ❤ by GitHub

Please find the key snippet from the above script –

(imgH, imgW) = frame.shape[:2]
(srcH, srcW) = source.shape[:2]

# Detect Aruco markers in the input frame
(corners, ids, rejected) = cv2.aruco.detectMarkers(frame, arucoDict, parameters=arucoParams)

Identifying the Aruco markers are key here. The above lines help the program detect all four corners.

However, let us discuss more on the Aruco markers & strategies that I’ve used for several different surfaces.

As you can see, the right-hand side Aruco marker is tiny compared to the left one. Hence, that one will be ideal for a curve surface like Coffee Mug, Bottle rather than a flat surface.

Also, we’ve demonstrated the zoom capability with the smaller Aruco marker that will Augment almost double the original surface area.

Let us understand why we need that; as you know, any spherical surface like a bottle is round-shaped. Hence, detecting relatively more significant Aruco markers in four corners will be difficult for any camera to identify.

Hence, we need a process where close four corners can be extrapolated mathematically to relatively larger projected areas easily detectable by any WebCAM.

Let’s observe the following figure –

As you can see that the original position of the four corners is represented using the following points, i.e., (x1, y1), (x2, y2), (x3, y3) & (x4, y4).

And these positions are very close to each other. Hence, it will be easier for the camera to detect all the points (like a plain surface) without many retries.

And later, you can add specific values of x & y to them to get the derived four corners as shown in the above figures through the following points, i.e. (x1.1, y1.1), (x2.1, y2.1), (x3.1, y3.1) & (x4.1, y4.1).

# Loop over the IDs of the ArUco markers in Top-Left, Top-Right,
# Bottom-Right, and Bottom-Left order
for i in cornerIDs:
  # Grab the index of the corner with the current ID
  j = np.squeeze(np.where(ids == i))

  # If we receive an empty list instead of an integer index,
  # then we could not find the marker with the current ID
  if j.size == 0:
    continue

  # Otherwise, append the corner (x, y)-coordinates to our list
  # of reference points
  corner = np.squeeze(corners[j])
  refPts.append(corner)

# Check to see if we failed to find the four ArUco markers
if len(refPts) != 4:
  # If we are allowed to use cached reference points, fall
  # back on them
  if useCache and CACHED_REF_PTS is not None:
    refPts = CACHED_REF_PTS

  # Otherwise, we cannot use the cache and/or there are no
  # previous cached reference points, so return early
  else:
    return None

# If we are allowed to use cached reference points, then update
# the cache with the current set
if useCache:
  CACHED_REF_PTS = refPts

# Unpack our Aruco reference points and use the reference points
# to define the Destination transform matrix, making sure the
# points are specified in Top-Left, Top-Right, Bottom-Right, and
# Bottom-Left order
(refPtTL, refPtTR, refPtBR, refPtBL) = refPts
dstMat = [refPtTL[0], refPtTR[1], refPtBR[2], refPtBL[3]]
dstMat = np.array(dstMat)

In the above snippet, the application will scan through all the points & try to detect Aruco markers & then create a list of reference points, which will later be used to define the destination transformation matrix.

# For zoom option recalculating all the 4 points
refPtTL1_L_X = refPtTL[0][0]-TOP_LEFT_X
refPtTL1_L_Y = refPtTL[0][1]-TOP_LEFT_Y

refPtTL1.append((refPtTL1_L_X,refPtTL1_L_Y))

refPtTL1_R_X = refPtTL[1][0]+TOP_RIGHT_X
refPtTL1_R_Y = refPtTL[1][1]+TOP_RIGHT_Y

refPtTL1.append((refPtTL1_R_X,refPtTL1_R_Y))

refPtTD1_L_X = refPtTL[2][0]+BOTTOM_RIGHT_X
refPtTD1_L_Y = refPtTL[2][1]+BOTTOM_RIGHT_Y

refPtTL1.append((refPtTD1_L_X,refPtTD1_L_Y))

refPtTD1_R_X = refPtTL[3][0]-BOTTOM_LEFT_X
refPtTD1_R_Y = refPtTL[3][1]+BOTTOM_LEFT_Y

refPtTL1.append((refPtTD1_R_X,refPtTD1_R_Y))

dstMatMod = [refPtTL1[0], refPtTL1[1], refPtTL1[2], refPtTL1[3]]
dstMatMod = np.array(dstMatMod)

The above snippets calculate the revised points for the zoom-out capabilities as discussed in one of the earlier figures.

# Define the transform matrix for the *source* image in Top-Left,
# Top-Right, Bottom-Right, and Bottom-Left order
srcMat = np.array([[0, 0], [srcW, 0], [srcW, srcH], [0, srcH]])

The above snippet will create a transformation matrix for the video trailer.

# Compute the homography matrix and then warp the source image to
# the destination based on the homography depending upon the
# zoom flag
if zoomFlag == 1:
  (H, _) = cv2.findHomography(srcMat, dstMat)
else:
  (H, _) = cv2.findHomography(srcMat, dstMatMod)

warped = cv2.warpPerspective(source, H, (imgW, imgH))

# Construct a mask for the source image now that the perspective
# warp has taken place (we'll need this mask to copy the source
# image into the destination)
mask = np.zeros((imgH, imgW), dtype="uint8")
if zoomFlag == 1:
  cv2.fillConvexPoly(mask, dstMat.astype("int32"), (255, 255, 255), cv2.LINE_AA)
else:
  cv2.fillConvexPoly(mask, dstMatMod.astype("int32"), (255, 255, 255), cv2.LINE_AA)

# This optional step will give the source image a black
# border surrounding it when applied to the source image, you
# can apply a dilation operation
rect = cv2.getStructuringElement(cv2.MORPH_RECT, (3, 3))
mask = cv2.dilate(mask, rect, iterations=2)

# Create a three channel version of the mask by stacking it
# depth-wise, such that we can copy the warped source image
# into the input image
maskScaled = mask.copy() / 255.0
maskScaled = np.dstack([maskScaled] * 3)

# Copy the warped source image into the input image by
# (1) Multiplying the warped image and masked together,
# (2) Then multiplying the original input image with the
#     mask (giving more weight to the input where there
#     are not masked pixels), and
# (3) Adding the resulting multiplications together
warpedMultiplied = cv2.multiply(warped.astype("float"), maskScaled)
imageMultiplied = cv2.multiply(frame.astype(float), 1.0 - maskScaled)
output = cv2.add(warpedMultiplied, imageMultiplied)
output = output.astype("uint8")

Finally, depending upon the zoom flag, the application will create a warped image surrounded by an optionally black border.

clsEmbedVideoWithStream.py (This is the main class of python script that will invoke the clsAugmentedReality class to initiate augment reality after splitting the audio & video & then project them via the Web-CAM with a seamless broadcast.)

	##################################################
	#### Written By: SATYAKI DE ####
	#### Written On: 22-Jun-2022 ####
	#### Modified On 25-Jun-2022 ####
	#### ####
	#### Objective: This is the main class of ####
	#### python script that will invoke the ####
	#### clsAugmentedReality class to initiate ####
	#### augment reality after splitting the ####
	#### audio & video & then project them via ####
	#### the Web-CAM with a seamless broadcast. ####
	##################################################

	# Importing necessary packages
	import clsAugmentedReality as ar
	from clsConfig import clsConfig as cf

	from imutils.video import VideoStream
	from collections import deque

	import imutils
	import time
	import cv2
	import subprocess
	import os
	import pygame
	import time
	import threading
	import sys

	###############################################
	### Global Section ###
	###############################################
	# Instantiating the dependant class

	x1 = ar.clsAugmentedReality()

	###############################################
	### End of Global Section ###
	###############################################

	class BreakLoop(Exception):
	pass

	class clsEmbedVideoWithStream:
	def __init__(self):
	self.sep = str(cf.conf['SEP'])
	self.Curr_Path = str(cf.conf['INIT_PATH'])
	self.FileName = str(cf.conf['FILE_NAME'])
	self.CacheL = int(cf.conf['CACHE_LIM'])
	self.FileName_1 = str(cf.conf['FILE_NAME_1'])
	self.audioLen = int(cf.conf['audioLen'])
	self.audioFreq = float(cf.conf['audioFreq'])
	self.videoFrame = float(cf.conf['videoFrame'])
	self.stopFlag=cf.conf['stopFlag']
	self.zFlag=int(cf.conf['zoomFlag'])
	self.title = str(cf.conf['TITLE'])

	def playAudio(self, audioFile, audioLen, freq, stopFlag=False):
	try:
	pygame.mixer.init()
	pygame.init()
	pygame.mixer.music.load(audioFile)

	pygame.mixer.music.set_volume(10)

	val = int(audioLen)
	i = 0

	while i < val:
	pygame.mixer.music.play(loops=0, start=float(i))
	time.sleep(freq)

	i = i + 1

	if (i >= val):
	raise BreakLoop

	if (stopFlag==True):
	raise BreakLoop

	return 0
	except BreakLoop as s:
	return 0
	except Exception as e:
	x = str(e)
	print(x)

	return 1

	def extractAudio(self, video_file, output_ext="mp3"):
	try:
	"""Converts video to audio directly using `ffmpeg` command
	with the help of subprocess module"""
	filename, ext = os.path.splitext(video_file)
	subprocess.call(["ffmpeg", "-y", "-i", video_file, f"{filename}.{output_ext}"],
	stdout=subprocess.DEVNULL,
	stderr=subprocess.STDOUT)

	return 0
	except Exception as e:
	x = str(e)
	print('Error: ', x)

	return 1

	def processStream(self, debugInd, var):
	try:
	sep = self.sep
	Curr_Path = self.Curr_Path
	FileName = self.FileName
	CacheL = self.CacheL
	FileName_1 = self.FileName_1
	audioLen = self.audioLen
	audioFreq = self.audioFreq
	videoFrame = self.videoFrame
	stopFlag = self.stopFlag
	zFlag = self.zFlag
	title = self.title

	print('audioFreq:')
	print(str(audioFreq))

	print('videoFrame:')
	print(str(videoFrame))

	# Construct the source for Video & Temporary Audio
	videoFile = Curr_Path + sep + 'Video' + sep + FileName
	audioFile = Curr_Path + sep + 'Video' + sep + FileName_1

	# Load the Aruco dictionary and grab the Aruco parameters
	print("[INFO] initializing marker detector…")
	arucoDict = cv2.aruco.Dictionary_get(cv2.aruco.DICT_ARUCO_ORIGINAL)
	arucoParams = cv2.aruco.DetectorParameters_create()

	# Initialize the video file stream
	print("[INFO] accessing video stream…")
	vf = cv2.VideoCapture(videoFile)

	x = self.extractAudio(videoFile)

	if x == 0:
	print('Successfully Audio extracted from the source file!')
	else:
	print('Failed to extract the source audio!')

	# Initialize a queue to maintain the next frame from the video stream
	Q = deque(maxlen=128)

	# We need to have a frame in our queue to start our augmented reality
	# pipeline, so read the next frame from our video file source and add
	# it to our queue
	(grabbed, source) = vf.read()
	Q.appendleft(source)

	# Initialize the video stream and allow the camera sensor to warm up
	print("[INFO] starting video stream…")
	vs = VideoStream(src=0).start()

	time.sleep(2.0)
	flg = 0

	t = threading.Thread(target=self.playAudio, args=(audioFile, audioLen, audioFreq, stopFlag,))
	t.daemon = True

	try:
	# Loop over the frames from the video stream
	while len(Q) > 0:
	try:
	# Grab the frame from our video stream and resize it
	frame = vs.read()
	frame = imutils.resize(frame, width=1020)

	# Attempt to find the ArUCo markers in the frame, and provided
	# they are found, take the current source image and warp it onto
	# input frame using our augmented reality technique
	warped = x1.getWarpImages(
	frame, source,
	cornerIDs=(923, 1001, 241, 1007),
	arucoDict=arucoDict,
	arucoParams=arucoParams,
	zoomFlag=zFlag,
	useCache=CacheL > 0)

	# If the warped frame is not None, then we know (1) we found the
	# four ArUCo markers and (2) the perspective warp was successfully
	# applied
	if warped is not None:
	# Set the frame to the output augment reality frame and then
	# grab the next video file frame from our queue
	frame = warped
	source = Q.popleft()

	if flg == 0:

	t.start()
	flg = flg + 1

	# For speed/efficiency, we can use a queue to keep the next video
	# frame queue ready for us — the trick is to ensure the queue is
	# always (or nearly full)
	if len(Q) != Q.maxlen:
	# Read the next frame from the video file stream
	(grabbed, nextFrame) = vf.read()

	# If the frame was read (meaning we are not at the end of the
	# video file stream), add the frame to our queue
	if grabbed:
	Q.append(nextFrame)

	# Show the output frame
	cv2.imshow(title, frame)
	time.sleep(videoFrame)

	# If the `q` key was pressed, break from the loop
	if cv2.waitKey(2) & 0xFF == ord('q'):
	stopFlag = True
	break

	except BreakLoop:
	raise BreakLoop
	except Exception as e:
	pass

Category: call

Tuning your model using the python-based low-code machine-learning library PyCaret

Like this:

Handling unique data using the python-based FastDataMask Package package

Like this:

Python performance improvement with 3.11 Version

Like this:

Scanned data extraction from a prefilled form using OpenCV & Python

Like this:

Share this:

Like this:

Share this:

Like this:

Share this:

Like this:

Share this:

Like this:

Share this:

Like this:

Share this:

Like this:

Share this:

Like this: