Spaces:
Build error
Build error
| # Author: Firqa Aqila Noor Arasyi | |
| # Date: 2023-12-04 | |
| import os | |
| import io | |
| import json | |
| import pandas as pd | |
| import streamlit as st | |
| from stqdm import stqdm | |
| from ast import literal_eval | |
| from tempfile import NamedTemporaryFile | |
| from json_repair import repair_json | |
| import PyPDF2 | |
| import pdf2image | |
| import pytesseract | |
| from utils import * | |
| from schema import * | |
| from summ import get_summ | |
| from datetime import datetime | |
| import time | |
| import base64 | |
| import string | |
| import random | |
| import numpy as np | |
| from langchain.llms import OpenAI | |
| from langchain.chains import RetrievalQA | |
| from langchain.vectorstores import Chroma | |
| from langchain.chat_models import ChatOpenAI | |
| from langchain.document_loaders import TextLoader | |
| from chromadb.utils import embedding_functions | |
| from unstructured.partition.pdf import partition_pdf | |
| from unstructured.staging.base import elements_to_json | |
| from langchain.text_splitter import CharacterTextSplitter | |
| from langchain.embeddings.openai import OpenAIEmbeddings | |
| from langchain.chains import create_extraction_chain | |
| from Bio import Entrez | |
| nltk.download("punkt") | |
| os.environ['OPENAI_API_KEY'] = os.getenv("OPENAI_API_KEY") | |
| Entrez.email = os.getenv("ENTREZ_EMAIL") | |
| Entrez.api_key = os.getenv("ENTREZ_API_KEY") | |
| fold = -1 | |
| buffer = io.BytesIO() | |
| st.cache_data() | |
| def convert_df(df): | |
| return df.to_csv().encode("utf-8") | |
| # Function to create a download link for an Excel file | |
| # def create_excel_download_link(df, file_name): | |
| # output = io.BytesIO() | |
| # with pd.ExcelWriter(output, engine='xlsxwriter') as writer: | |
| # df.to_excel(writer, sheet_name='Sheet1', index=False) | |
| # excel_data = output.getvalue() | |
| # st.download_button(label="Download Excel File", data=excel_data, key=file_name, file_name=f"{file_name}.xlsx") | |
| class Journal: | |
| def __init__(self, name, bytes): | |
| self.name = name | |
| self.bytes = bytes | |
| def __repr__(self): | |
| return f"Journal(name='{self.name}', bytes='{self.bytes}')" | |
| llm = ChatOpenAI(temperature=0, model="gpt-3.5-turbo-1106") | |
| textex_chain = create_extraction_chain(textex_schema, llm) | |
| tablex_chain = create_extraction_chain(tablex_schema, llm) | |
| st.set_page_config(page_title="NutriGenMe Paper Extractor") | |
| st.title("NutriGenMe - Paper Extraction") | |
| st.markdown("<div style='text-align: left; color: white; font-size: 16px'>In its latest version, the app is equipped to extract essential information from papers, including tables in both horizontal and vertical orientations, images, and text exclusively.</div><br>", unsafe_allow_html=True) | |
| uploaded_files = st.file_uploader("Upload Paper(s) here :", type="pdf", accept_multiple_files=True) | |
| if uploaded_files: | |
| st.warning(""" | |
| Warning! Prior to proceeding, please take a moment to review the following : \n | |
| Certain guidelines apply when utilizing this application, particularly if you intend to extract information from tables, whether they are oriented horizontally or vertically. | |
| - If you intend to perform multiple PDF processes using Horizontal Table Extraction, ensure that all your PDF files adhere to a horizontal table format | |
| - If you plan to undertake multiple PDF processes with Vertical Table Extraction, ensure that all your PDF files conform to a vertical table format | |
| """, icon="⚠️") | |
| col1, col2, col3 = st.columns(3) | |
| if uploaded_files: | |
| journals = [] | |
| strategy = "hi_res" | |
| model_name = "yolox" | |
| on_h, on_v, on_t = None, None, None | |
| parseButtonH, parseButtonV, parseButtonT = None, None, None | |
| # if uploaded_files: | |
| with col1: | |
| if on_v or on_t: | |
| on_h = st.toggle("Horizontal Table Extraction", disabled=True) | |
| else: | |
| on_h = st.toggle("Horizontal Table Extraction") | |
| if on_h: | |
| chunk_size_h = st.selectbox( | |
| 'Tokens amounts per process :', | |
| (15000, 12000, 10000, 8000, 5000), key='table_h' | |
| ) | |
| parseButtonH = st.button("Get Result", key='table_H') | |
| with col2: | |
| if on_h or on_t: | |
| on_v = st.toggle("Vertical Table Extraction", disabled=True) | |
| else: | |
| on_v = st.toggle("Vertical Table Extraction") | |
| if on_v: | |
| chunk_size_v = st.selectbox( | |
| 'Tokens amounts per process :', | |
| (15000, 12000, 10000, 8000, 5000), key='table_v' | |
| ) | |
| parseButtonV = st.button("Get Result", key='table_V') | |
| with col3: | |
| if on_h or on_v: | |
| on_t = st.toggle("Text Extraction ", disabled=True) | |
| else: | |
| on_t = st.toggle("Text Extraction ") | |
| if on_t: | |
| chunk_size_t = st.selectbox( | |
| 'Tokens amounts per process :', | |
| (15000, 12000, 10000, 8000, 5000), key='no_table' | |
| ) | |
| parseButtonT = st.button("Get Result", key="no_Table") | |
| if on_h: | |
| if parseButtonH: | |
| with st.status("Extraction in progress ...", expanded=True) as status: | |
| st.write("Getting Result ...") | |
| csv = pd.DataFrame() | |
| for uploaded_file in stqdm(uploaded_files): | |
| with NamedTemporaryFile(dir='.', suffix=".pdf") as pdf: | |
| pdf.write(uploaded_file.getbuffer()) | |
| # st.write(pdf.name) | |
| L = [] | |
| # Entity Extraction | |
| st.write("☑ Extracting Entities ...") | |
| bytes_data = uploaded_file.read() | |
| journal = Journal(uploaded_file.name, bytes_data) | |
| images = pdf2image.convert_from_bytes(journal.bytes) | |
| extracted_text = "" | |
| for image in images[:-1]: | |
| text = pytesseract.image_to_string(image) | |
| text = clean_text(text) | |
| extracted_text += text + " " | |
| text = replace_quotes(extracted_text) | |
| text_chunk = split_text(text, chunk_size_h) | |
| chunkdf = [] | |
| for i, chunk in enumerate(text_chunk): | |
| inp = chunk | |
| df = pd.DataFrame(literal_eval(str(json.dumps(tablex_chain.run(inp)[0])).replace("\'", '\"')), index=[0]).fillna('') | |
| # df = pd.DataFrame(repair_json(tablex_chain.run(inp)[0])) | |
| chunkdf.append(df) | |
| concat = pd.concat(chunkdf, axis=0).reset_index().drop('index', axis=1).fillna('') | |
| st.write("☑ Entities Extraction Done ..") | |
| time.sleep(0.1) | |
| st.write("☑ Generating Summary ...") | |
| summary = get_summ(pdf.name) | |
| st.write("☑ Generating Summary Done ..") | |
| time.sleep(0.1) | |
| st.write("☑ Table Extraction in progress ...") | |
| # Table Extraction | |
| # L = [] | |
| output_list = [] | |
| elements = partition_pdf(filename=pdf.name, strategy=strategy, infer_table_structure=True, model_name=model_name) | |
| with NamedTemporaryFile(dir=".", suffix=".json") as f: | |
| elements_to_json(elements, filename=f"{f.name.split('/')[-1]}") | |
| json_file_path = os.path.abspath(f.name) # Get the absolute file path | |
| with open(json_file_path, "r", encoding="utf-8") as jsonfile: | |
| data = json.load(jsonfile) | |
| extracted_elements = [] | |
| for entry in data: | |
| if entry["type"] == "Table": | |
| extracted_elements.append(entry["metadata"]["text_as_html"]) | |
| with NamedTemporaryFile(dir='.' , suffix='.txt') as txt_file: | |
| text_file_path = os.path.abspath(txt_file.name) | |
| with open(text_file_path, "w", encoding="utf-8") as txtfile: | |
| for element in extracted_elements: | |
| txtfile.write(element + "\n\n") | |
| loader = TextLoader(text_file_path) | |
| documents = loader.load() | |
| # split it into chunks | |
| text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200, separator="\n") | |
| docs = text_splitter.split_documents(documents) | |
| embeddings = OpenAIEmbeddings() | |
| db = Chroma.from_documents(docs, embeddings) | |
| llm_table = ChatOpenAI(model_name="gpt-3.5-turbo-16k", temperature=0) | |
| qa_chain = RetrievalQA.from_chain_type(llm_table, retriever=db.as_retriever()) | |
| # List of questions | |
| questions = [ | |
| """Mention all genes / locus name with respective rsID / SNP and potential diseases in a curly brackets like this: | |
| Example 1 : {"Genes" : "FTO", "SNPs" : "rs9939609", "Diseases" : "Obesity"} | |
| """, | |
| """Mention all genes / locus name with respective potential diseases in a curly brackets like this: | |
| Example 2 : {"Genes" : "FTO", "SNPs" : "" (if not available), "Diseases" : "Obesity"} | |
| """, | |
| """Mention all rsIDs / SNPs / Variant with respective potential diseases / traits in a curly brackets like this: | |
| Example 3 : {"Genes" : "", "SNPs" : "rs9939609", "Diseases" : "Obesity"} | |
| """ | |
| ] | |
| try: | |
| for query in questions: | |
| response = qa_chain({"query" : query}) | |
| output_list.append(response) | |
| except Exception as e: | |
| pass | |
| db.delete_collection() | |
| st.write(concat) | |
| # 1 | |
| for i in range(len(output_list[0]['result'].split('\n'))): | |
| st.write(output_list[0]['result'].split('\n')) | |
| if output_list[0]['result'].split('\n')[i] != "": | |
| try: | |
| row = literal_eval(repair_json(output_list[0]['result'].split('\n')[i]))[0] | |
| st.write(row) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| # 'Population' : concat['population_race'][0], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| # 'Sample Size' : concat['sample_size'][0] | |
| }} | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| # 'Population' : concat['population_race'][0], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| # 'Sample Size' : concat['sample_size'][0], | |
| 'Genes' : g.strip().upper().replace('Unknown', ''), | |
| 'SNPs' : row['SNPs'].replace('Unknown', ''), | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', '').replace('Unknown', '') | |
| }) | |
| else: | |
| L.append(row) | |
| except KeyError: | |
| row = literal_eval(repair_json(output_list[0]['result'].split('\n')[i])) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| "Publisher Name" : concat['publisher_name'][0], | |
| # 'Population' : concat['population_race'][0], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| # 'Sample Size' : concat['sample_size'][0] | |
| } | |
| } | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| 'Genes' : g.strip().upper().replace('Unknown', ''), | |
| 'SNPs' : row['SNPs'].replace('Unknown', ''), | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', '').replace('Unknown', '') | |
| }) | |
| else: | |
| L.append(row) | |
| except SyntaxError: | |
| row = literal_eval(repair_json(output_list[0]['result'].split('\n')[i])) | |
| row = f"""{row}""" | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| # 'Population' : concat['population_race'][0], | |
| # 'Sample Size' : concat['sample_size'][0] | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except ValueError: | |
| if type(output_list[0]['result'].split('\n')[i]) is dict: | |
| row = repair_json(output_list[0]['result'].split('\n')[i]) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| # 2 | |
| for i in range(len(output_list[1]['result'].split('\n'))): | |
| if output_list[1]['result'].split('\n')[i] != "": | |
| try: | |
| row = literal_eval(repair_json(output_list[1]['result'].split('\n')[i]))[0] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if row['SNPs'] != "Not available": | |
| row.update({ | |
| 'SNPs' : "Not available" | |
| }) | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| 'Genes' : g.strip().upper().replace('Unknown', ''), | |
| "SNPs" : "Not available", | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', '').replace('Unknown', '') | |
| }) | |
| else: | |
| L.append(row) | |
| except KeyError: | |
| row = literal_eval(repair_json(output_list[1]['result'].split('\n')[i])) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if row['SNPs'] != "Not available": | |
| row.update({ | |
| 'SNPs' : "Not available" | |
| }) | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| 'Genes' : g.strip().upper().replace('Unknown', ''), | |
| "SNPs" : "Not available", | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', '').replace('Unknown', '') | |
| }) | |
| else: | |
| L.append(row) | |
| except SyntaxError: | |
| row = f"""{row}""" | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except ValueError: | |
| if type(output_list[1]['result'].split('\n')[i]) is dict: | |
| row = output_list[1]['result'].split('\n')[i] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| # 3 | |
| for i in range(len(output_list[2]['result'].split('\n'))): | |
| if output_list[2]['result'].split('\n')[i] != "": | |
| try: | |
| row = literal_eval(repair_json(output_list[2]['result'].split('\n')[i]))[0] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except KeyError: | |
| row = literal_eval(repair_json(output_list[2]['result'].split('\n')[i])) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except SyntaxError: | |
| row = f"""{row}""" | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except ValueError: | |
| if type(output_list[2]['result'].split('\n')[i]) is dict: | |
| row = output_list[2]['result'].split('\n')[i] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : upper_abbreviation(' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title()) if 'population_race' in concat.columns else concat.assign(population_race='')['population_race'], | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title()) if 'study_methodology' in concat.columns else concat.assign(study_methodology='')['study_methodology'], | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title()) if 'study_level' in concat.columns else concat.assign(study_level='')['study_level'], | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| st.write(output_list[2]['result'].split('\n')) | |
| st.write("☑ Table Extraction Done ...") | |
| status.update(label="Gene and SNPs succesfully collected.") | |
| L = [{key: ''.join(['' if item == 'Unknow' else item for item in value]) for key, value in d.items()} for d in L] | |
| L = [{key: ''.join(['Not Available' if item == '' else item for item in value]) for key, value in d.items()} for d in L] | |
| csv = pd.DataFrame(L) | |
| st.dataframe(csv) | |
| generated_key = ''.join(random.choice(string.ascii_letters + string.digits) for i in range(16)) | |
| # if st.button("Download Excel File", key=generated_key): | |
| # excel_link = create_excel_download_link(csv, uploaded_file.name.replace('.pdf', '')) | |
| # st.markdown(excel_link, unsafe_allow_html=True) | |
| with pd.ExcelWriter(buffer, engine='xlsxwriter') as writer: | |
| # Write each dataframe to a different worksheet | |
| csv.to_excel(writer, sheet_name='Result') | |
| writer.close() | |
| # time_now = datetime.now() | |
| # current_time = time_now.strftime("%H:%M:%S") | |
| csv = convert_df(csv) | |
| st.download_button( | |
| label="Save Result", | |
| data=buffer, | |
| file_name=f'{uploaded_file.name}'.replace('.pdf', '') + '.xlsx', | |
| mime='application/vnd.ms-excel', | |
| key=generated_key | |
| ) | |
| if on_v: | |
| if parseButtonV: | |
| with st.status("Extraction in progress ...", expanded=True) as status: | |
| st.write("Getting Result ...") | |
| csv = pd.DataFrame() | |
| for uploaded_file in stqdm(uploaded_files): | |
| L = [] | |
| with NamedTemporaryFile(dir='.', suffix=".pdf") as pdf: | |
| pdf.write(uploaded_file.getbuffer()) | |
| # Open the PDF file in read-binary mode | |
| with open(pdf.name, 'rb') as pdf_file: | |
| # Create a PDF reader object | |
| pdf_reader = PyPDF2.PdfReader(pdf_file) | |
| # Create a PDF writer object to write the rotated pages to a new PDF | |
| pdf_writer = PyPDF2.PdfWriter() | |
| # Iterate through each page in the original PDF | |
| for page_num in range(len(pdf_reader.pages)): | |
| # Get the page object | |
| page = pdf_reader.pages[page_num] | |
| # Rotate the page 90 degrees clockwise (use -90 for counterclockwise) | |
| page.rotate(90) | |
| # Add the rotated page to the PDF writer | |
| pdf_writer.add_page(page) | |
| with NamedTemporaryFile(dir='.', suffix=".pdf") as rotated_pdf: | |
| pdf_writer.write(rotated_pdf.name) | |
| # Entity Extraction | |
| st.write("☑ Extracting Entities ...") | |
| bytes_data = uploaded_file.read() | |
| journal = Journal(uploaded_file.name, bytes_data) | |
| images = pdf2image.convert_from_bytes(journal.bytes) | |
| extracted_text = "" | |
| for image in images[:-1]: | |
| text = pytesseract.image_to_string(image) | |
| text = clean_text(text) | |
| extracted_text += text + " " | |
| text = replace_quotes(extracted_text) | |
| text_chunk = split_text(text, chunk_size_v) | |
| chunkdf = [] | |
| for i, chunk in enumerate(text_chunk): | |
| inp = chunk | |
| df = pd.DataFrame(literal_eval(str(json.dumps(tablex_chain.run(inp)[0])).replace("\'", '\"')), index=[0]).fillna('') | |
| chunkdf.append(df) | |
| concat = pd.concat(chunkdf, axis=0).reset_index().drop('index', axis=1).fillna('') | |
| st.write("☑ Entities Extraction Done ..") | |
| time.sleep(0.1) | |
| st.write("☑ Generating Summary ...") | |
| summary = get_summ(pdf.name) | |
| st.write("☑ Generating Summary Done ..") | |
| time.sleep(0.1) | |
| st.write("☑ Table Extraction in progress ...") | |
| # Table Extraction | |
| output_list = [] | |
| elements = partition_pdf(filename=rotated_pdf.name, strategy=strategy, infer_table_structure=True, model_name=model_name) | |
| with NamedTemporaryFile(dir=".", suffix=".json") as f: | |
| elements_to_json(elements, filename=f"{f.name.split('/')[-1]}") | |
| json_file_path = os.path.abspath(f.name) # Get the absolute file path | |
| with open(json_file_path, "r", encoding="utf-8") as jsonfile: | |
| data = json.load(jsonfile) | |
| extracted_elements = [] | |
| for entry in data: | |
| if entry["type"] == "Table": | |
| extracted_elements.append(entry["metadata"]["text_as_html"]) | |
| with NamedTemporaryFile(dir='.' , suffix='.txt') as txt_file: | |
| text_file_path = os.path.abspath(txt_file.name) | |
| with open(text_file_path, "w", encoding="utf-8") as txtfile: | |
| for element in extracted_elements: | |
| txtfile.write(element + "\n\n") | |
| loader = TextLoader(text_file_path) | |
| documents = loader.load() | |
| # split it into chunks | |
| text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=200, separator="\n") | |
| docs = text_splitter.split_documents(documents) | |
| embeddings = OpenAIEmbeddings() | |
| db = Chroma.from_documents(docs, embeddings) | |
| llm_table = ChatOpenAI(model_name="gpt-3.5-turbo-16k", temperature=0) | |
| qa_chain = RetrievalQA.from_chain_type(llm_table, retriever=db.as_retriever()) | |
| # List of questions | |
| questions = [ | |
| """Mention all genes / locus name with respective rsID / SNP and potential diseases in a curly brackets like this: | |
| Example 1 : {"Genes" : "FTO", "SNPs" : "rs9939609", "Diseases" : "Obesity"} | |
| """, | |
| """Mention all genes / locus name with respective potential diseases in a curly brackets like this: | |
| Example 2 : {"Genes" : "FTO", "SNPs" : "" (if not available), "Diseases" : "Obesitya"} | |
| """, | |
| """Mention all rsIDs / SNPs / Variant with respective potential diseases / traits in a curly brackets like this: | |
| Example 3 : {"Genes" : "", "SNPs" : "rs9939609", "Diseases" : "Obesity"} | |
| """ | |
| ] | |
| try: | |
| for query in questions: | |
| response = qa_chain({"query" : query}) | |
| output_list.append(response) | |
| except Exception as e: | |
| pass | |
| db.delete_collection() | |
| # 1 | |
| for i in range(len(output_list[0]['result'].split('\n'))): | |
| if output_list[0]['result'].split('\n')[i] != "": | |
| try: | |
| row = literal_eval(output_list[0]['result'].split('\n')[i])[0] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }} | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Genes' : g.strip().upper(), | |
| 'SNPs' : row['SNPs'], | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', ''), | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }) | |
| else: | |
| L.append(row) | |
| except KeyError: | |
| row = literal_eval(output_list[0]['result'].split('\n')[i]) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }} | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Genes' : g.strip().upper(), | |
| 'SNPs' : row['SNPs'], | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', ''), | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }) | |
| else: | |
| L.append(row) | |
| except ValueError: | |
| if type(output_list[0]['result'].split('\n')[i]) is dict: | |
| row = output_list[0]['result'].split('\n')[i] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except SyntaxError: | |
| row = literal_eval("""{}""".format(output_list[2]['result'].split('\n')[i])) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| # 2 | |
| for i in range(len(output_list[1]['result'].split('\n'))): | |
| if output_list[1]['result'].split('\n')[i] != "": | |
| try: | |
| row = literal_eval(output_list[1]['result'].split('\n')[i])[0] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }} | |
| if row['SNPs'] != "Not available": | |
| row.update({ | |
| 'SNPs' : "Not available" | |
| }) | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Genes' : g.strip().upper(), | |
| "SNPs" : "Not available", | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', ''), | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }) | |
| else: | |
| L.append(row) | |
| except KeyError: | |
| row = literal_eval(output_list[1]['result'].split('\n')[i]) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }} | |
| if row['SNPs'] != "Not available": | |
| row.update({ | |
| 'SNPs' : "Not available" | |
| }) | |
| if len(row['Genes'].strip().split(',')) > 1: | |
| for g in row['Genes'].strip().split(','): | |
| L.append({ | |
| 'Genes' : g.strip().upper(), | |
| "SNPs" : "Not available", | |
| "Diseases" : ''.join(list(row['Diseases'].title() if row['Diseases'] not in ['T2D', 'T2DM', 'NAFLD', 'CVD'] else row['Diseases'])).replace('Unknown', ''), | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| }) | |
| else: | |
| L.append(row) | |
| except ValueError: | |
| if type(output_list[1]['result'].split('\n')[i]) is dict: | |
| row = output_list[1]['result'].split('\n')[i] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except SyntaxError: | |
| row = literal_eval("""{}""".format(output_list[2]['result'].split('\n')[i])) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| # 3 | |
| for i in range(len(output_list[2]['result'].split('\n'))): | |
| if output_list[2]['result'].split('\n')[i] != "": | |
| try: | |
| row = literal_eval(output_list[2]['result'].split('\n')[i])[0] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except KeyError: | |
| row = literal_eval(output_list[2]['result'].split('\n')[i]) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except ValueError: | |
| if type(output_list[2]['result'].split('\n')[i]) is dict: | |
| row = output_list[2]['result'].split('\n')[i] | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| except SyntaxError: | |
| row = literal_eval("""{}""".format(output_list[2]['result'].split('\n')[i])) | |
| row = {**row, **{ | |
| 'Title' : concat['title'][0], | |
| 'Authors' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher_name'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['year_of_publication'].values.tolist())) if 'year_of_publication' in concat.columns else concat.assign(year_of_publication='')['year_of_publication'], | |
| 'Population' : ' '.join(concat['population_race'].values.tolist()).replace('Unknown', '').title(), | |
| 'Sample Size' : sample_size_postproc(' '.join(concat['sample_size'].values.tolist()).replace('Unknown', '').title()) if 'sample_size' in concat.columns else concat.assign(sample_size='')['sample_size'], | |
| 'Study Methodology' : ' '.join(concat['study_methodology'].values.tolist()).replace('Unknown', '').title(), | |
| 'Study Level' : ' '.join(concat['study_level'].values.tolist()).replace('Unknown', '').title(), | |
| 'Recommendation' : summary, | |
| } | |
| } | |
| if not row['SNPs'].startswith("rs"): | |
| row.update({ | |
| 'SNPs' : "-" | |
| }) | |
| else: | |
| L.append(row) | |
| st.write("☑ Table Extraction Done") | |
| status.update(label="Gene and SNPs succesfully collected.") | |
| L = [{key: ''.join(['' if item == 'Unknow' else item for item in value]) for key, value in d.items()} for d in L] | |
| L = [{key: ''.join(['Not Available' if item == '' else item for item in value]) for key, value in d.items()} for d in L] | |
| csv = pd.DataFrame(L) | |
| st.dataframe(csv) | |
| with pd.ExcelWriter(buffer, engine='xlsxwriter') as writer: | |
| # Write each dataframe to a different worksheet | |
| csv.to_excel(writer, sheet_name='Result') | |
| writer.close() | |
| time_now = datetime.now() | |
| current_time = time_now.strftime("%H:%M:%S") | |
| csv = convert_df(csv) | |
| st.download_button( | |
| label="Save Result", | |
| data=buffer, | |
| file_name=f'{uploaded_file.name}'.replace('.pdf', '') + '.xlsx', | |
| mime='application/vnd.ms-excel' | |
| ) | |
| if on_t: | |
| if parseButtonT: | |
| with st.status("Extraction in progress ...", expanded=True) as status: | |
| st.write("Getting Result ...") | |
| csv = pd.DataFrame() | |
| for uploaded_file in stqdm(uploaded_files): | |
| L = [] | |
| with NamedTemporaryFile(dir='.', suffix=".pdf") as pdf: | |
| pdf.write(uploaded_file.getbuffer()) | |
| # Entity Extraction | |
| st.write("☑ Extracting Entities ...") | |
| bytes_data = uploaded_file.read() | |
| journal = Journal(uploaded_file.name, bytes_data) | |
| images = pdf2image.convert_from_bytes(journal.bytes) | |
| extracted_text = "" | |
| for image in images[:-1]: | |
| text = pytesseract.image_to_string(image) | |
| text = clean_text(text) | |
| extracted_text += text + " " | |
| text = replace_quotes(extracted_text) | |
| text_chunk = split_text(text, chunk_size_t) | |
| chunkdf = [] | |
| for i, chunk in enumerate(text_chunk): | |
| inp = chunk | |
| df = pd.DataFrame(literal_eval(str(json.dumps(textex_chain.run(inp)[0])).replace("\'", "\"")), index=[0]).fillna('') | |
| chunkdf.append(df) | |
| concat = pd.concat(chunkdf, axis=0).reset_index().drop('index', axis=1).fillna('') | |
| st.write("☑ Entities Extraction Done ..") | |
| time.sleep(0.1) | |
| st.write("☑ Generating Summary ...") | |
| concat['SNPs'] = concat['SNPs'].apply(lambda x: x if x.startswith('rs') else '') | |
| for col in list(concat.columns): | |
| concat[col] = concat[col].apply(lambda x: x if x not in ['N/A', 'not mentioned', 'Not mentioned', 'Unknown'] else '') | |
| summary = get_summ(pdf.name) | |
| time.sleep(0.1) | |
| st.write("☑ Generating Summary Done...") | |
| for i in range(len(concat)): | |
| if (len(concat['genes_locus'][i].split(',')) >= 1) and concat['SNPs'][i] == '': | |
| for g in concat['genes_locus'][i].split(','): | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Author' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['publication_year'].values.tolist())), | |
| 'Genes' : g.upper(), | |
| 'Population' : upper_abbreviation(' '.join(np.unique(concat['population_race'].values.tolist())).title()), | |
| 'Diseases' : upper_abbreviation(' '.join(concat['diseases'].values.tolist()).title()), | |
| 'Sample Size' : sample_size_postproc(upper_abbreviation(' '.join(concat['sample_size'].values.tolist()).title())), | |
| 'SNPs' : concat['SNPs'][i], | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).title()), | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).title()), | |
| 'Recommendation' : summary, | |
| }) | |
| elif (len(concat['SNPs'][i].split(',')) >= 1): | |
| for s in concat['SNPs'][i].split(','): | |
| try: | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Author' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['publication_year'].values.tolist())), | |
| 'Genes' : get_geneName(s.strip()).upper(), | |
| 'Population' : upper_abbreviation(' '.join(np.unique(concat['population_race'].values.tolist())).title()), | |
| 'Diseases' : upper_abbreviation(' '.join(concat['diseases'].values.tolist()).title()), | |
| 'Sample Size' : sample_size_postproc(upper_abbreviation(' '.join(concat['sample_size'].values.tolist()).title())), | |
| 'SNPs' : s, | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).title()), | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).title()), | |
| 'Recommendation' : summary, | |
| }) | |
| except Exception as e: | |
| L.append({ | |
| 'Title' : concat['title'][0], | |
| 'Author' : concat['authors'][0], | |
| 'Publisher Name' : concat['publisher'][0], | |
| 'Publication Year' : get_valid_year(' '.join(concat['publication_year'].values.tolist())), | |
| 'Genes' : '', | |
| 'Population' : upper_abbreviation(' '.join(np.unique(concat['population_race'].values.tolist())).title()), | |
| 'Diseases' : upper_abbreviation(' '.join(concat['diseases'].values.tolist()).title()), | |
| 'Sample Size' : sample_size_postproc(upper_abbreviation(' '.join(concat['sample_size'].values.tolist()).title())), | |
| 'SNPs' : s, | |
| 'Study Methodology' : upper_abbreviation(' '.join(concat['study_methodology'].values.tolist()).title()), | |
| 'Study Level' : upper_abbreviation(' '.join(concat['study_level'].values.tolist()).title()), | |
| 'Recommendation' : summary, | |
| }) | |
| csv = pd.concat([csv, pd.DataFrame(L)], ignore_index=True) | |
| status.update(label="Gene and SNPs succesfully collected.") | |
| st.dataframe(csv) | |
| with pd.ExcelWriter(buffer, engine='xlsxwriter') as writer: | |
| # Write each dataframe to a different worksheet | |
| csv.to_excel(writer, sheet_name='Result') | |
| writer.close() | |
| time_now = datetime.now() | |
| current_time = time_now.strftime("%H:%M:%S") | |
| csv = convert_df(csv) | |
| st.download_button( | |
| label="Save Result", | |
| data=buffer, | |
| file_name=f'{uploaded_file.name}'.replace('.pdf', '') + '.xlsx', | |
| mime='application/vnd.ms-excel' | |
| ) |