version_2

Tools.py

@@ -0,0 +1,248 @@
+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}"}

database.py

@@ -0,0 +1,43 @@
+import pymysql
+import json
+import pandas as pd
+import re
+
+
+def initialize_database():
+    # Database Connection
+    db_params = {"host": "localhost",
+                "user": "cms-readonly-user",
+                "password": "%Reed!!",
+                "port": 3307,
+                "database": "veda_prod_v1"
+                }
+    db = pymysql.connect(**db_params)
+    return db
+
+def execute_query(query):
+    db = initialize_database()
+    cursor = db.cursor()
+    try:
+        cursor.execute(query)
+        description = cursor.description
+        result = cursor.fetchall()  # Fetch all rows from the result set
+        db.commit()
+        return description, result
+    except Exception as e:
+        print("Error executing query:", e)
+        db.rollback()
+        return None  # Return None if an error occurs
+    finally:
+        db.close()
+
+
+def _get_details_mantra_json(self, query):
+    description, data = execute_query(query)
+    df = pd.DataFrame(data)
+    df.columns = [x[0] for x in description]
+    mantra_json = df['mantra_json'].values[0]
+    cleaned_data = re.sub('<[^<]+?>', '', mantra_json)
+    return json.loads(cleaned_data)
+
+

requirements.txt

@@ -3,7 +3,9 @@ sentence_transformers==2.4.0
 llama_index==0.10.4
 llama-index-vector-stores-pinecone
 llama-index-embeddings-huggingface
+llama_index-embeddings-nomic
 pinecone-client==3.1.0
 cohere==4.50 
 chardet==5.2.0
-streamlit==1.31.1
+streamlit==1.31.1
+aksharamukha==2.1.2

utils.py

@@ -0,0 +1,120 @@
+import os
+import re
+import requests
+import json
+import numpy as np
+import pandas as pd
+from bs4 import BeautifulSoup
+from database import execute_query
+from aksharamukha import transliterate
+from sentence_transformers import util
+from llama_index.embeddings.nomic import NomicEmbedding
+
+nomic_api_key = os.getenv('NOMIC_API_KEY')
+#nomic embed model used for similarity scores
+nomic_embed_model = NomicEmbedding(
+    api_key=nomic_api_key,
+    dimensionality=128,
+    model_name="nomic-embed-text-v1.5",
+)
+
+
+
+def get_list_meaning_word(word):
+    pada_meanings = {'pada': word,
+                     'Monier-Williams Sanskrit-English Dictionary (1899)': [],
+                     'Shabda-Sagara (1900)': [],
+                     'Apte-Practical Sanskrit-English Dictionary (1890)': [],
+                     }
+    url = f"https://ambuda.org/tools/dictionaries/mw,shabdasagara,apte/{word}"
+
+    try:
+        # Fetch HTML content
+        response = requests.get(url)
+        response.raise_for_status()
+
+        # Parse HTML with BeautifulSoup
+        soup = BeautifulSoup(response.text, 'html.parser')
+
+        # Extracting text from different tags
+        divs = soup.find_all('div', class_='my-4', attrs={'x-show': 'show'})
+        
+        try:
+            # Find all list items <li> within the specified <ul> tag
+            div_items_0 = divs[0].find('ul').find_all('li', class_='dict-entry mw-entry')
+            # Print the text content of each list item
+            dive_text_0 = [li_tag.get_text(strip=True) for li_tag in div_items_0]
+            text_0_trans = [transliterate.process(src='Devanagari', tgt='IAST', txt=text) for text in dive_text_0]
+            pada_meanings['Monier-Williams Sanskrit-English Dictionary (1899)'] = text_0_trans
+        except :
+            print("Error: Unable to find Monier-Williams Sanskrit-English Dictionary (1899) data.")
+
+        try:
+            div_items_1 = divs[1].find_all('div')
+            dive_text_1 = [item.get_text(strip=True) for item in div_items_1]
+            text_1_trans = [transliterate.process(src='Devanagari', tgt='IAST', txt=text) for text in dive_text_1]
+            pada_meanings['Shabda-Sagara (1900)'] = text_1_trans
+        except :
+            print("Error: Unable to find Shabda-Sagara (1900) data.")
+
+        try:
+            apte_meanings = []
+            for tag in divs[2].find_all('b'):
+                if tag.text.strip() != '—':
+                    text1 = tag.text.strip()  # English text within <b> tag
+                    sibling = tag.find_next_sibling()  # Text following <b> tag
+                    text2 = tag.next_sibling.strip() + ' ' # English text following <b> tag
+                    while sibling.name != 'div':
+                        if sibling.name is None:  # Handling non-tag text
+                            text2 += " "
+                        elif sibling.name == 'span':  # Handling <b> tag
+                            IAST_text = transliterate.process(src='Devanagari', tgt='IAST', txt=sibling.text.strip()) 
+                            text2 += IAST_text  +  ' ' + sibling.next_sibling.strip()
+                        else:
+                            text2 += sibling.text.strip() +  ' ' +  sibling.next_sibling.strip()
+                        sibling = sibling.find_next_sibling()
+                    apte_meanings.append(text2)
+            pada_meanings['Apte-Practical Sanskrit-English Dictionary (1890)'] = apte_meanings[:-1]
+        except:
+            print("Error: Unable to find Apte-Practical Sanskrit-English Dictionary (1890) data.")
+
+    except requests.exceptions.RequestException as e:
+        print(f"Error: Failed to fetch data from {url}. {e}")
+
+    return pada_meanings
+
+#get similarity scores
+def word_sentence_similarity(meanings, root_stem_word):
+    # Check if the word embeddings are not empty
+    if not meanings or not root_stem_word:
+        return None
+    
+    meaning_embedding = np.array(nomic_embed_model.get_text_embedding(meanings))
+    all_meanings = []
+    word_score_pair = []
+    all_meanings.extend(get_list_meaning_word(root_stem_word)['Monier-Williams Sanskrit-English Dictionary (1899)'])
+    all_meanings.extend(get_list_meaning_word(root_stem_word)['Shabda-Sagara (1900)'])
+    for word_meaning in all_meanings:
+        root_stem_word_meaning_embedding = np.array(nomic_embed_model.get_text_embedding(word_meaning))
+        # Calculate cosine similarity
+        similarity_score = util.pytorch_cos_sim(meaning_embedding, root_stem_word_meaning_embedding).item()
+        word_score_pair.append((word_meaning,similarity_score))
+    # Sort the list in descending order based on similarity scores
+    sorted_word_score_pairs = sorted(word_score_pair, key=lambda x: x[1], reverse=True)
+    return sorted_word_score_pairs
+
+#extract the adhibautic meaning of the mantra from the vedamantra
+def extract_meaning_by_language(data_list, target_language='English'):
+    for data_dict in data_list:
+        if data_dict.get('languageName') == target_language:
+            return data_dict.get('mahatma', {})
+    return None  
+
+#mantra_json_details
+def get_details_mantra_json(query):
+    description, data = execute_query(query)
+    df = pd.DataFrame(data)
+    df.columns = [x[0] for x in description]
+    mantra_json = df['mantra_json'].values[0]
+    cleaned_data = re.sub('<[^<]+?>', '', mantra_json)
+    return json.loads(cleaned_data)