Spaces:
Running
Running
import json | |
import chardet | |
import pandas as pd | |
import streamlit as st | |
import pymysql | |
import ast | |
import re | |
from utils import word_sentence_similarity, extract_meaning_by_language, get_list_meaning_word, get_details_mantra_json | |
from llama_index.core.tools.tool_spec.base import BaseToolSpec | |
from database import execute_query | |
import pandas as pd | |
import json | |
import ast | |
import logging | |
# Constants | |
SCRIPTURE_DESCRIPTIONS_CSV_PATH = r"Data/scripture_descriptions.csv" | |
VEDAMANTRA_CSV_PATH = r"Data/veda_content_modified_v3.csv" | |
PADA_CSV_PATH = r"Data/term_data_processed_v2.csv" | |
class ScriptureDescriptionToolSpec(BaseToolSpec): | |
spec_functions = ["get_description"] | |
def __init__(self): | |
super().__init__() | |
with open(SCRIPTURE_DESCRIPTIONS_CSV_PATH, 'rb') as f: | |
result = chardet.detect(f.read()) | |
encoding = result['encoding'] | |
self.df = pd.read_csv(SCRIPTURE_DESCRIPTIONS_CSV_PATH, encoding=encoding) | |
def _query_description(self, conditions): | |
try: | |
result = self.df[conditions] | |
if not result.empty: | |
return result.iloc[0].to_dict() | |
else: | |
raise IndexError("Scripture description not found.") | |
except IndexError as e: | |
raise ValueError(f"Failed to get scripture description: {e}") | |
def get_description(self, level_0, level_1=None, level_2=None, level_3=None): | |
try: | |
conditions = (self.df['scripture_name'].str.lower() == level_0.lower()) | |
if level_3 is not None: | |
conditions &= (self.df['level_1'] == str(level_1)) & (self.df['level_2'] == str(level_2)) & (self.df['level_3'] == str(level_3)) | |
elif level_2 is not None: | |
conditions &= (self.df['level_1'] == str(level_1)) & (self.df['level_2'] == str(level_2)) | |
elif level_1 is not None: | |
conditions &= (self.df['level_1'] == str(level_1)) | |
return self._query_description(conditions) | |
except ValueError as e: | |
return {"error": str(e)} | |
class MantraToolSpec(BaseToolSpec): | |
''' | |
To obtain the vedamantra details such as vedamantra, padapata, devata, chandah, rishi etc of vedamantras (or mantras or hyms) from all vedas (RigVeda, AtharvaVeda, SamaVeda, KrishnaYajurVeda, and ShuklaYajurVeda) using the function | |
`get_vedamantra_details`. The mantra summary like anvaya, mantraVishaya, bhavartha/meaning (adhibautic, ahyatmic, adhidaivic), purpose, usage, tippani of vedamantra accessible using the function 'get_vedamantra_summary' | |
Sample Query: | |
1. What is the vedamantra of the mantra from Rigveda, first mandala, first shukta, and first mantra? | |
2. What is the devata of the vedamantra from Rigveda, first mandala, first shukta, and first mantra? | |
3. What is the meaning of the vedamantra from Rigveda, first mandala, first shukta, and first mantra written by Tulsi Ram? | |
4. What is the (adhibautic) meaning of the vedamantra from RigVeda, first mandala, first shukta, and first mantra? | |
5. What is the mantraVishaya of the vedamantra from RigVeda, first mandala, first shukta, and first mantra? | |
''' | |
spec_functions = ["get_vedamantra_details", "get_vedamantra_summary"] | |
def __init__(self): | |
super().__init__() | |
self.df_vedamantra = pd.read_csv(VEDAMANTRA_CSV_PATH, encoding='utf-8') | |
def _get_mantra_details(self, query): | |
try: | |
details = get_details_mantra_json(query) | |
return details['mantraHeader']['language'][1] | |
except Exception as e: | |
raise ValueError(f"Failed to get mantra details: {e}") | |
def _query_db(self, conditions): | |
try: | |
result = self.df_vedamantra[conditions]['mantra_number'].values | |
if len(result) == 0: | |
raise ValueError("Mantra not found.") | |
return result[0] | |
except Exception as e: | |
raise ValueError("Failed to query database.") | |
def _get_query_conditions(self, scripture_name, **kwargs): | |
conditions = (self.df_vedamantra['scripture_name'].str.lower() == scripture_name.lower()) | |
for key, value in kwargs.items(): | |
conditions &= (self.df_vedamantra[key] == value) | |
return conditions | |
def _get_mantra_id(self, scripture_name, **kwargs): | |
conditions = self._get_query_conditions(scripture_name, **kwargs) | |
return self._query_db(conditions) | |
def get_vedamantra_details(self, mantraid=None, scripture_name=None, **kwargs): | |
try: | |
if mantraid: | |
query = f"SELECT mantra_json FROM veda_content WHERE mantra_number = '{mantraid}'" | |
else: | |
mantra_id = self._get_mantra_id(scripture_name, **kwargs) | |
query = f"SELECT mantra_json FROM veda_content WHERE mantra_number = '{mantra_id}'" | |
return self._get_mantra_details(query) | |
except Exception as e: | |
return {"error": str(e)} | |
def get_vedamantra_summary(self, mantraid=None, scripture_name=None, **kwargs): | |
try: | |
if mantraid: | |
query = f"SELECT mantra_json FROM veda_content WHERE mantra_number = '{mantraid}'" | |
else: | |
mantra_id = self._get_mantra_id(scripture_name, **kwargs) | |
query = f"SELECT mantra_json FROM veda_content WHERE mantra_number = '{mantra_id}'" | |
json_dict = get_details_mantra_json(query) | |
mantra_summary = json_dict['mantraSummary']['language'] | |
summary_dict = {"Roman-IAST summary of vedamantra": json_dict['mantraSummary']['language'][1]} | |
for item in mantra_summary: | |
if item['languageName'] == 'English': | |
mahatma = item['mahatma']['mahatmaName'] | |
summary_dict[f"English summary of vedamantra by {mahatma}"] = item | |
return summary_dict | |
except Exception as e: | |
return {"error": str(e)} | |
class PadaToolSpec(BaseToolSpec): | |
''' | |
Purpose: To obtains a complete or meaningful meaning of a word or pada based on context information. | |
1. The function 'get_meaning_pada' used to get all the possible meanings of the pada based on the given information. | |
2. The function 'get_adibauatic_adidaivic_adyatmic_meaning_of_pada' used to get the adibhautic, adidaivic and sdyatmic meaning of a word based on context information.\ | |
Use the context to generate a meaningful meaning of the pada in the vedamantra. | |
Sample query: | |
1. What is the meaning of the word apratidhṛṣṭa-śavasam? | |
2. What is the adibauatic meaning of the word apratidhṛṣṭa-śavasam? | |
3. Whats the adidaivic meaning of the word apratidhṛṣṭa-śavasam? | |
4. What is the adyatmic meaning of the word apratidhṛṣṭa-śavasam? | |
''' | |
spec_functions = ["get_pada_meaning","get_adibauatic_adidaivic_adhyatmic_meaning_of_pada"] | |
def __init__(self): | |
super().__init__() | |
self.df_terms = pd.read_csv(PADA_CSV_PATH, dtype={'AnuvakNumber': 'Int64', 'PrapatakNumber': 'Int64', 'KandahNumber': 'Int64', 'ShuktaNumber': 'Int64', 'ArchikahNumber': 'Int64', 'AdhyayaNumber': 'Int64', 'MandalaNumber': 'Int64', 'ParyayaNumber': 'Int64'}, encoding='utf-8') | |
self.df_vedic_content = pd.read_csv(VEDAMANTRA_CSV_PATH,encoding = 'utf-8') | |
def _get_pada_details_by_scripture(self, pada, scripture_name=None, **kwargs): | |
try: | |
condition = (self.df_terms['Pada'] == pada) | |
if scripture_name: | |
condition &= (self.df_terms['scripture_name'].str.lower() == scripture_name.lower()) | |
for key, value in kwargs.items(): | |
if value is not None: | |
condition &= (self.df_terms[key] == value) | |
filtered_df = self.df_terms[condition] | |
return filtered_df if not filtered_df.empty else None | |
except Exception as e: | |
logging.error(f"Error in _get_pada_details_by_scripture: {e}") | |
return None | |
def _get_vedamantra_meaning(self, mantraID, MahatmaName=None): | |
try: | |
query = f"SELECT mantra_json FROM veda_content WHERE mantra_number = '{mantraID}'" | |
jsonDict = get_details_mantra_json(query) | |
mantraSummary = jsonDict['mantraSummary']['language'] | |
if MahatmaName is not None: | |
filtered_summary = [data_dict for data_dict in mantraSummary if data_dict.get('mahatma', {}).get('mahatmaName') == MahatmaName] | |
if filtered_summary: | |
mantraSummary = filtered_summary | |
best_meaning = None | |
best_count = 0 | |
for data_dict in mantraSummary: | |
if data_dict.get('languageName') == "English": | |
meanings = data_dict['mahatma']['bhavartha'] | |
count = sum(bool(meanings.get(cat, None)) for cat in ['adibhautic', 'adidaivic', 'adhyatmic']) | |
if count >= best_count: | |
best_meaning = {cat: meanings.get(cat, None) for cat in ['adibhautic', 'adidaivic', 'adhyatmic']} | |
best_count = count | |
return best_meaning if best_meaning else {"error": "Required meaning associated with vedamantra is not available."} | |
except Exception as e: | |
logging.error(f"Error in _get_vedamantra_meaning: {e}") | |
return {"error": f"An error occurred: {e}"} | |
def _get_pada_morphology(self, term_details, meanings): | |
try: | |
morphology_list = ast.literal_eval(term_details['Morphology']) | |
term_morph_list = [] | |
for morphs in morphology_list: | |
term_info = {} | |
for field in ['stem', 'root']: | |
morph_word = morphs.get(field) | |
if morph_word: | |
meaning = word_sentence_similarity(meanings, morph_word) | |
term_info[f'{field}_word'] = morph_word | |
term_info[f'{field}_meaning'] = meaning[0][0] if meaning else None | |
term_info[f'{field}_score'] = meaning[0][1] if meaning else None | |
term_info['grammar'] = morphs['grammar'] | |
term_morph_list.append(term_info) | |
return term_morph_list | |
except Exception as e: | |
logging.error(f"Error in _get_pada_morphology: {e}") | |
return [] | |
def get_pada_meaning(self, pada): | |
try: | |
pada_details = self.df_terms[self.df_terms['Pada'] == pada] | |
meanings_list = [] | |
for morphs in ast.literal_eval(pada_details['Morphology'].values[0]): | |
for field in ['stem', 'root']: | |
word = morphs.get(field) | |
if word: | |
meanings_list.append(get_list_meaning_word(word)) | |
return meanings_list | |
except Exception as e: | |
logging.error(f"Error in get_pada_meaning: {e}") | |
return {"error": f"Required meaning associated with pada is not available. {e}"} | |
def get_adibauatic_adidaivic_adhyatmic_meaning_of_pada(self, pada, Pada_position=None, mantraid=None, scripture_name=None, **kwargs): | |
try: | |
if mantraid: | |
details = self.df_terms[(self.df_terms['mantra_id'] == mantraid) & (self.df_terms['Pada'] == pada)] | |
else: | |
if scripture_name is not None: | |
details = self._get_pada_details_by_scripture(pada, scripture_name, **kwargs) | |
if Pada_position: | |
details = details[details['Pada_position'] == Pada_position] | |
if details.empty: | |
return {"error": f"No details found for pada '{pada}'"} | |
pada_details = details.iloc[0] | |
#print(pada_details) | |
mantraID = pada_details['mantra_id'] | |
meanings = self._get_vedamantra_meaning(mantraID,MahatmaName=kwargs.get('MahatmaName')) | |
if 'error' in meanings: | |
return meanings | |
ab_term_morph_list = self._get_pada_morphology(pada_details, meanings['adibhautic']) | |
ad_term_morph_list = self._get_pada_morphology(pada_details, meanings['adidaivic']) | |
at_term_morph_list = self._get_pada_morphology(pada_details, meanings['adhyatmic']) | |
return { | |
f'adibhautic_info_{pada}': ab_term_morph_list, | |
'vedamantra_adibhautic_meaning': meanings['adibhautic'], | |
f'adidavic_info_{pada}': ad_term_morph_list, | |
'vedamantra_adidavic_meaning': meanings['adidaivic'], | |
f'adhyatmic_info_{pada}': at_term_morph_list, | |
'vedamantra_adhyatmic_meaning': meanings['adhyatmic'] | |
} | |
except Exception as e: | |
logging.error(f"Error in get_adibauatic_adidaivic_adhyatmic_meaning_of_pada: {e}") | |
return {"error": f"Failed to get meaning of the word {pada}. {e}"} | |