app_old.py

import pandas as pd
from utils import normalize_text
import streamlit as st

### Data paths
# WIKIPEDIA_PATH = "./kensho_en_wiki_typing_technical.csv"
# WIKIDATA_PATH = "./wikidata_ss_processed.csv"
# REBEL_INFER_PATH = "./rebel_inference_processed_ss.csv"
# ENTITY_LINKING_PATH = "./linking_df_technical_min.csv"

relation_to_id = {
    "uses": 2283,
    "has_use": 366,
    "part_of": 361,
    "has_part": 527,
    "made_from_material": 186
}

st.title("Materials use case search app")



# User Input
input_text = st.text_input(
    label="Enter the name of a material i.e steel, sand, plastic, etc and press Enter",
    value="steel",
    key="ent",
)

st.write("preparing data ...")

# Wikipedia metadata
@st.cache_data(persist="disk")
def get_wiki_df(path="./kensho_en_wiki_typing_technical.csv"):
    wiki_df = pd.read_csv(path)
    
    # filter out technical articles
    exclude_ids = set(wiki_df[(wiki_df.exclude == True) | (wiki_df.technical == False)].page_id.to_list())
    include_skpes = set(wiki_df[wiki_df.page_id.apply(lambda x: x not in exclude_ids)].skpe_id.to_list())

    skpe_to_wikidata = dict(zip(wiki_df.skpe_id.to_list(), wiki_df.item_id.to_list()))
    
    wiki_df = wiki_df.drop(columns=['Unnamed: 0', 'en_probs', 'exclude'])
    wiki_df = wiki_df.rename(columns={'title_x': 'en_title'})
    
    return wiki_df, include_skpes, skpe_to_wikidata

wiki_df, include_skpes, skpe_to_wikidata = get_wiki_df()

# KG data source 1: Wikidata
@st.cache_data(persist="disk")
def get_wikidata_df(path="./wikidata_ss_processed.csv"):
    wikidata_df = pd.read_csv(path)
    
    # filter technical wikidata
    wikidata_df = wikidata_df[wikidata_df.apply(lambda x: x.source_skpe in include_skpes and x.target_skpe in include_skpes, axis=1)]
    
    wikidata_df['source_wikidata'] = wikidata_df.source_skpe.apply(lambda x: skpe_to_wikidata[x])
    wikidata_df['target_wikidata'] = wikidata_df.target_skpe.apply(lambda x: skpe_to_wikidata[x])
    wikidata_df = wikidata_df.drop(columns=['source_skpe', 'target_skpe'])
    
    wikidata_df['source'] = 'wikidata'
    
    return wikidata_df

wikidata_df = get_wikidata_df()

@st.cache_data(persist="disk")
def get_rebel_infer_df(path="./rebel_inference_processed_ss.csv"):
    rebel_infer_df = pd.read_csv(path)
    
    # filter technical
    rebel_infer_df = rebel_infer_df[rebel_infer_df.apply(lambda x: type(x.source_skpe_id) == str and type(x.target_skpe_id) == str, axis=1)]

    rebel_infer_df = rebel_infer_df[rebel_infer_df.apply(lambda x: x.source_skpe_id in skpe_to_wikidata.keys() and x.target_skpe_id in skpe_to_wikidata.keys(), axis=1)]
    rebel_infer_df['source_wikidata'] = rebel_infer_df.source_skpe_id.apply(lambda x: skpe_to_wikidata[x])
    rebel_infer_df['target_wikidata'] = rebel_infer_df.target_skpe_id.apply(lambda x: skpe_to_wikidata[x])
    # rebel_infer_df['title_page_id'] = rebel_infer_df.page_skpe_id.apply(lambda x: skpe_to_wikidata[x])

    rebel_infer_df = rebel_infer_df.drop(columns=['instance_id', 'source_text', 'target_text', 'page_skpe_id', 'source_skpe_id', 'target_skpe_id'])
    rebel_infer_df = rebel_infer_df.rename(columns={'source_skpe_id': 'source_skpe', 'target_skpe_id': 'target_skpe', 'source': 'source_en', 'target': 'target_en'})
    rebel_infer_df = rebel_infer_df[rebel_infer_df.source_wikidata != rebel_infer_df.target_wikidata]

    rebel_infer_df['source'] = 'rebel_wikipedia'
    
    return rebel_infer_df

rebel_infer_df = get_rebel_infer_df()

kg_df = pd.concat([wikidata_df, rebel_infer_df])

@st.cache_data(persist="disk")
def get_entity_linking_df(path="./linking_df_technical_min.csv"):
    linking_df = pd.read_csv(path)
    return linking_df

st.write("matching input text ...")

linking_df = get_entity_linking_df()

# normalise and match
text_norm = normalize_text(input_text)
match_df = linking_df[linking_df.text == text_norm]

match_df = match_df[match_df.skpe_id.apply(lambda x: x in skpe_to_wikidata.keys())]
match_df['wikidata_id'] = match_df.skpe_id.apply(lambda x: skpe_to_wikidata[x]) 

# top match skpe
if len(match_df) > 0:

    top_wikidata = match_df.wikidata_id.mode()[0]
    all_wikidata = set(match_df.wikidata_id.to_list())
    wikidata_to_count = dict(match_df.wikidata_id.value_counts())

    # Match list
    wiki_match_df = wiki_df[wiki_df.item_id.apply(lambda x: x in all_wikidata)].copy()
    wiki_match_df['link_score'] = wiki_match_df['item_id'].apply(lambda x: wikidata_to_count[x] / sum(wikidata_to_count.values()))
    wiki_match_df = wiki_match_df.sort_values(by='link_score', ascending=False)

    # show similar results
    st.write(f"Found following matches for the term {input_text}")
    wiki_match_df.sort_values(by='views', ascending=False)[:5]

    # proceeding with top match
    st.write("Performing use case extraction for the following top match ...")
    wiki_df[wiki_df.item_id.apply(lambda x: x == top_wikidata)]
    
    # Stuff that are made out of input
    made_of_df = kg_df[(kg_df.relation == 'made_from_material') & (kg_df.target_wikidata == top_wikidata)].copy()
    # made_of_list = made_of_df.source_wikidata.to_list()

    if len(made_of_df) > 0:

        st.write(f"Discovered following entities made out of {input_text}")
        made_of_df[['source_ja', 'source_en', 'relation', 'target_ja', 'target_en', 'source', 'page_title']]

        st.write("Extracting knowledge graph paths ...")
        
        all_paths = []
        
        # iterate over first rows
        for first_edge in made_of_df.itertuples():
            
            first_item = first_edge.source_wikidata
        
            # applications of stuff made out of first item
            use_df = kg_df[((kg_df.relation == 'has_use') & (kg_df.source_wikidata == first_item)) | ((kg_df.relation == 'uses') & (kg_df.target_wikidata == first_item))]
        
            # add all 2 len paths
            for second_edge in use_df.itertuples():
                all_paths.append([first_edge, second_edge])
                
            # expand to part of
            
            # applications of stuff made out of steel # 1
            part_df = kg_df[((kg_df.relation == 'has_part') & (kg_df.target_wikidata == first_item)) | (kg_df.relation == 'part_of') & (kg_df.source_wikidata == first_item)]
        
            # iterate over all parts of product
            for second_edge in part_df.itertuples():
        
                # select second item
                second_item = second_edge.source_wikidata if second_edge.relation == 'has_part' else second_edge.target_wikidata
        
                # get uses of second item
                use_df = kg_df[((kg_df.relation == 'has_use') & (kg_df.source_wikidata == second_item)) | ((kg_df.relation == 'uses') & (kg_df.target_wikidata == second_item))]
        
                # add all 3 len paths
                for third_edge in use_df.itertuples():
                    all_paths.append([first_edge, second_edge, third_edge])
        
        if len(all_paths) > 0:

            st.write(f"Found {len(all_paths)} knowledge graph paths relevant to use cases of {input_text}")
            st.write("------")

            # print all paths
            for i, path in enumerate(all_paths):
                material = path[0].target_en
                material_wikidata = path[0].target_wikidata
                material_url = f"https://www.wikidata.org/wiki/Q{material_wikidata}"

                use_case = path[-1].source_en if path[-1].relation == 'uses' else path[-1].target_en
                use_case_wikidata = path[-1].source_wikidata if path[-1].relation == 'uses' else path[-1].target_wikidata
                use_case_url = f"https://www.wikidata.org/wiki/Q{use_case_wikidata}"

                st.write(f"**Reasoning Path {i+1}:**")

                for edge in path:

                    source_url = f"https://www.wikidata.org/wiki/Q{edge.source_wikidata}"
                    target_url = f"https://www.wikidata.org/wiki/Q{edge.target_wikidata}"

                    relation_url = f"https://www.wikidata.org/wiki/Property:P{relation_to_id[edge.relation]}"

                    st.markdown(f"[{edge.source_en}]({source_url}) --[{edge.relation}]({relation_url})--> [{edge.target_en}]({target_url}) (source: {edge.source})")
                st.write("**Conclusion:**")
                st.write(f"[{material}]({material_url}) is useful for [{use_case}]({use_case_url})")
                st.write("------")
        else:
            st.write("Found no knowledge graph paths relevant to use cases")
    else:
        st.write("Found no entities that are made from {input_text}")

else:
    st.write("no matches")