PSGReportExtract_Embla.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "execution_count": null,
   "id": "4b0f7c9e",
   "metadata": {},
   "outputs": [],
   "source": [
    "## Natus Embla KeyLogic PSG Report Extractor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "729779f7-5e49-4ad9-b280-05a6a4444b65",
   "metadata": {},
   "outputs": [],
   "source": [
    "!pip install python-docx pymupdf olefile tqdm"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 248,
   "id": "84302552-aca9-4507-b089-a8ac28280f49",
   "metadata": {},
   "outputs": [],
   "source": [
    "folder_path = \"<FOLDER PATH HERE>\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 249,
   "id": "494a995b-53ee-44de-a291-1b9c4e990b58",
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "import re\n",
    "import fitz  # PyMuPDF\n",
    "from docx import Document\n",
    "import pandas as pd\n",
    "import olefile\n",
    "from tqdm import tqdm\n",
    "\n",
    "pd.set_option('display.max_columns', None)\n",
    "\n",
    "def correct_turkish_characters(text):\n",
    "    character_map = {\n",
    "        'Þ': 'Ş',\n",
    "        'þ': 'ş',\n",
    "        'Ý': 'İ',\n",
    "        'ý': 'ı',\n",
    "        'Ð': 'Ğ',\n",
    "        'ð': 'ğ',\n",
    "        'Ç': 'Ç',\n",
    "        'ç': 'ç',\n",
    "        'Ö': 'Ö',\n",
    "        'ö': 'ö',\n",
    "        'Ü': 'Ü',\n",
    "        'ü': 'ü'\n",
    "    }\n",
    "    for incorrect, correct in character_map.items():\n",
    "        text = text.replace(incorrect, correct)\n",
    "    return text\n",
    "\n",
    "# Function to extract text from a DOC file using olefile\n",
    "def doc_to_text(doc_path):\n",
    "    try:\n",
    "        ole = olefile.OleFileIO(doc_path)\n",
    "        if ole.exists('WordDocument'):\n",
    "            data = ole.openstream('WordDocument').read()\n",
    "            text = ''\n",
    "            for char in data:\n",
    "                if 32 <= char <= 126:\n",
    "                    text += chr(char)\n",
    "            return text\n",
    "    except Exception as e:\n",
    "        print(f\"Error processing DOC file {doc_path}: {e}\")\n",
    "        return None\n",
    "\n",
    "# Function to extract text from a DOCX file\n",
    "def docx_to_text_with_tables(file_path):\n",
    "    try:\n",
    "        doc = Document(file_path)\n",
    "        full_text = []\n",
    "        for para in doc.paragraphs:\n",
    "            full_text.append(para.text)\n",
    "        for table in doc.tables:\n",
    "            for row in table.rows:\n",
    "                row_text = ' | '.join(cell.text.strip() for cell in row.cells)\n",
    "                full_text.append(row_text)\n",
    "        return '\\n'.join(full_text)\n",
    "    except Exception as e:\n",
    "        print(f\"Error processing DOCX file {file_path}: {e}\")\n",
    "        return None\n",
    "\n",
    "# Function to preprocess DOCX content\n",
    "def preprocess_docx_content(content):\n",
    "    lines = content.split('\\n')\n",
    "    processed_lines = []\n",
    "    for line in lines:\n",
    "        parts = line.split('|', 2)\n",
    "        if len(parts) == 3:\n",
    "            line = parts[0] + '|' + parts[1] + '\\n' + parts[2]\n",
    "        if line.endswith('|'):\n",
    "            line = line[:-1]\n",
    "        processed_lines.append(line)\n",
    "    return '\\n'.join(processed_lines)\n",
    "\n",
    "# Function to extract information after a keyword\n",
    "def extract_information(preprocessed_text, keyword):\n",
    "    keyword_escaped = re.escape(keyword.strip())\n",
    "    pattern = re.compile(rf'{keyword_escaped}\\s*\\|\\s*([^\\|\\n]+)')\n",
    "    match = pattern.search(preprocessed_text)\n",
    "    if match:\n",
    "        return match.group(1).strip()\n",
    "    else:\n",
    "        pattern_with_spaces = re.compile(rf'\\s*{keyword_escaped}\\s*\\|\\s*([^\\|\\n]+)')\n",
    "        match_with_spaces = pattern_with_spaces.search(preprocessed_text)\n",
    "        if match_with_spaces:\n",
    "            return match_with_spaces.group(1).strip()\n",
    "        return None\n",
    "\n",
    "# Function to extract plain text between keywords\n",
    "def extract_plaintext(preprocessed_text, keyword, end_keywords=None):\n",
    "    keyword_escaped = re.escape(keyword.strip())\n",
    "    if end_keywords:\n",
    "        end_keywords_escaped = '|'.join(re.escape(end_kw.strip()) for end_kw in end_keywords)\n",
    "        pattern = re.compile(rf'{keyword_escaped}\\s*(.*?)\\s*(?:{end_keywords_escaped})', re.DOTALL)\n",
    "    else:\n",
    "        pattern = re.compile(rf'{keyword_escaped}\\s*([^\\|\\n]+)')\n",
    "    match = pattern.search(preprocessed_text)\n",
    "    if match:\n",
    "        return match.group(1).strip()\n",
    "    else:\n",
    "        #print(f\"No match found for keyword '{keyword}'.\")\n",
    "        return None\n",
    "\n",
    "# Function to extract text from a PDF file\n",
    "def pdf_to_text(file_path):\n",
    "    try:\n",
    "        document = fitz.open(file_path)\n",
    "        full_text = \"\"\n",
    "        for page_num in range(len(document)):\n",
    "            page = document.load_page(page_num)\n",
    "            full_text += page.get_text()\n",
    "        return full_text\n",
    "    except Exception as e:\n",
    "        print(f\"Error processing PDF file {file_path}: {e}\")\n",
    "        return None\n",
    "\n",
    "# Function to extract all text after a keyword\n",
    "def extract_text_after_keyword(text, keyword):\n",
    "    pattern = re.compile(rf'{keyword}:\\s*(.*?)\\s*(?=\\\\n|$)', re.DOTALL)\n",
    "    match = pattern.search(text)\n",
    "    if match:\n",
    "        return match.group(1).strip()\n",
    "    else:\n",
    "        #print(f\"No match found for keyword '{keyword}'.\")\n",
    "        return None\n",
    "\n",
    "# Function to get the next line after a keyword\n",
    "def get_next_line_after_keyword(preprocessed_text, keyword):\n",
    "    # Escape special characters in the keyword for regex\n",
    "    keyword_escaped = re.escape(keyword.strip())\n",
    "\n",
    "    # Split the content into lines\n",
    "    lines = preprocessed_text.split('\\n')\n",
    "\n",
    "    # Iterate over lines to find the keyword\n",
    "    for i, line in enumerate(lines):\n",
    "        if re.search(keyword_escaped, line):\n",
    "            # Return the next line if it exists\n",
    "            if i + 1 < len(lines):\n",
    "                return lines[i + 1].strip()\n",
    "            else:\n",
    "                return None\n",
    "\n",
    "    # Return None if the keyword is not found or there is no next line\n",
    "    return None\n",
    "\n",
    "def get_text_after_keyword_until_eol(content, keyword):\n",
    "    # Escape special characters in the keyword for regex\n",
    "    keyword_escaped = re.escape(keyword.strip())\n",
    "    \n",
    "    # Split the content into lines\n",
    "    lines = content.split('\\n')\n",
    "    \n",
    "    # Iterate over lines to find the keyword\n",
    "    for line in lines:\n",
    "        if re.search(keyword_escaped, line):\n",
    "            # Extract the text after the keyword until the end of the line\n",
    "            pattern = re.compile(rf'{keyword_escaped}(.*)')\n",
    "            match = pattern.search(line)\n",
    "            if match:\n",
    "                result = match.group(1).strip()\n",
    "                # Strip leading or trailing '|' characters\n",
    "                result = result.strip('|')\n",
    "                # Split by '|' and return as list of columns\n",
    "                columns = result.split('|')\n",
    "                return [col.strip() for col in columns]\n",
    "    \n",
    "    # Return None if the keyword is not found\n",
    "    return None\n",
    "\n",
    "def extract_summary(text, keyword, line_count):\n",
    "    # Escape the keyword for regex\n",
    "    keyword_escaped = re.escape(keyword)\n",
    "    pattern = re.compile(rf'{keyword_escaped}')\n",
    "    match = pattern.search(text)\n",
    "\n",
    "    if match:\n",
    "        start_line = text[:match.end()].count('\\n')\n",
    "        lines = text.split('\\n')\n",
    "        column_names = lines[start_line + 1:start_line + 1 + line_count]\n",
    "        values = lines[start_line + 1 + line_count:start_line + 1 + 2 * line_count]\n",
    "        \n",
    "        return column_names, values\n",
    "    else:\n",
    "        #print(f\"Keyword '{keyword}' not found in text.\")\n",
    "        return None, None\n",
    "\n",
    "def extract_value_lineskip(text, keyword, line_offset):\n",
    "    keyword_escaped = re.escape(keyword)\n",
    "    #pattern = re.compile(rf'{keyword_escaped}')\n",
    "    #pattern = re.compile(rf'\\b{keyword_escaped}\\b')\n",
    "    pattern = re.compile(rf'^{keyword_escaped}\\s*$', re.MULTILINE)\n",
    "\n",
    "\n",
    "    match = pattern.search(text)\n",
    "\n",
    "    if match:\n",
    "        start_line = text[:match.end()].count('\\n')\n",
    "        lines = text.split('\\n')\n",
    "        target_line_index = start_line + line_offset\n",
    "        if target_line_index < len(lines):\n",
    "            return lines[target_line_index].strip()\n",
    "        else:\n",
    "            print(f\"Line offset {line_offset} goes beyond the number of lines in the text.\")\n",
    "            return None\n",
    "    else:\n",
    "        #print(f\"Keyword '{keyword}' not found in text.\")\n",
    "        return None\n",
    "\n",
    "def get_all_doc_files(folder_path):\n",
    "    docx_files = []\n",
    "    for root, dirs, files in os.walk(folder_path):\n",
    "        for file in files:\n",
    "            if file.endswith(('.docx', '.DOCX', '.doc', '.DOC')):\n",
    "                docx_files.append(os.path.join(root, file))\n",
    "    return docx_files\n",
    "\n",
    "def get_all_pdf_files(folder_path):\n",
    "    docx_files = []\n",
    "    for root, dirs, files in os.walk(folder_path):\n",
    "        for file in files:\n",
    "            if file.endswith(('.pdf', '.PDF')):\n",
    "                docx_files.append(os.path.join(root, file))\n",
    "    return docx_files"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 250,
   "id": "fbb6b67f-eea6-42fc-bb45-f2acbdb8e4d1",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Function to process files\n",
    "def process_files(folder_path):\n",
    "    pdf_files = get_all_pdf_files(folder_path)\n",
    "    doc_files = get_all_doc_files(folder_path)\n",
    "    data = []\n",
    "\n",
    "    for pdf_file in tqdm(pdf_files, desc=\"Processing files\", unit=\"file\"):\n",
    "        row = {\"File\": pdf_file}\n",
    "        pdf_content = pdf_to_text(pdf_file)\n",
    "        pdf_content_corrected = correct_turkish_characters(pdf_content)\n",
    "        base_name = os.path.splitext(os.path.basename(pdf_file))[0]\n",
    "        \n",
    "        # Find the corresponding DOCX/DOC file\n",
    "        doc_file = next((doc for doc in doc_files if os.path.splitext(os.path.basename(doc))[0] == base_name), None)\n",
    "        \n",
    "        if doc_file:\n",
    "            if doc_file.lower().endswith('.docx'):\n",
    "                content = docx_to_text_with_tables(doc_file)\n",
    "            else:\n",
    "                content = doc_to_text(doc_file)\n",
    "\n",
    "            if content:\n",
    "                preprocessed_text = preprocess_docx_content(content)\n",
    "                diagnosis = extract_plaintext(preprocessed_text, \"Tanı:\", [\"Saygılarımızla\", \"Prof\", \"Uz\", \"Doç\"])\n",
    "                row[\"Tanı\"] = diagnosis\n",
    "                sonuc = extract_plaintext(preprocessed_text, \"Sonuç:\", [\"Saygılarımızla\", \"Prof\", \"Uz\", \"Doç\"])\n",
    "                row[\"Sonuç\"] = sonuc\n",
    "        else: row[\"Tanı\"] = \"Rapor bulunamadı.\"\n",
    "        \n",
    "        # Extract information from the PDF file\n",
    "        if pdf_content_corrected:\n",
    "            # Keywords to extract\n",
    "\n",
    "            # CUSTOMIZE THIS LIST FOR YOUR OWN REPORT. \n",
    "            # keywords will return the characters between the keyword and \"|\"\n",
    "            # keywords_nextline will return next line after the keyword.\n",
    "            # keywords_eol will return all characters until end of that line after the keyword.\n",
    "            # keyword_lineskip works like nextline but skips lines.\n",
    "            keywords = []\n",
    "            keywords_nextline = [\"Name:\", \"Date of Birth:\", \"Gender:\", \"Height:\", \"Weight:\", \"Age:\", \"RDI:\"]\n",
    "            keywords_eol = [\"Total Recording Time:\", \"Lights Off Clock Time:\", \"T.C:\", \"TC:\"]\n",
    "            keyword_lineskip = [\n",
    "                (\"Sleep Period:\", 22), (\"Wake After Sleep Onset:\", 22), (\"Total Sleep Time:\", 22), (\"Sleep Onset:\", 22),\n",
    "                (\"Sleep Efficiency:\", 22), (\"Number of Awakenings:\", 22), (\"Sleep Latency to N1:\", 22), (\"Sleep Latency to N2:\", 22),\n",
    "                (\"Sleep Latency to N3 (SWS):\", 22), (\"Stage R Latency from Sleep\", 22),\n",
    "                (\"Apnea + Hypopnea (A+H):\", 18), (\"Obstructive Apnea:\", 22), (\"Central Apnea:\", 22), (\"Mixed Apnea:\", 22),\n",
    "                (\"Hypopnea (All)\", 21), (\"Oxygen Desaturation Events\", 21),\n",
    "                (\"Limb Movement:\", 20), (\"PLMS:\", 20),\n",
    "                (\"N1\", 5), (\"N2\", 6), (\"N3\", 7), (\"R\", 8), (\"Wake\", 9),\n",
    "                (\"Total Arousals\", 169-157),\n",
    "                (\"Average Heart Rate during Sleep:\", 24), (\"Highest Heart Rate during Sleep:\", 24),\n",
    "                (\"Highest Heart Rate during Recording:\", 24), (\"Lowest Heart Rate during Sleep:\", 24),\n",
    "                (\"Lowest Heart Rate during Recording:\", 24)\n",
    "            ]\n",
    "\n",
    "            for keyword in keywords:\n",
    "                value = extract_text_after_keyword(pdf_content_corrected, keyword)\n",
    "                row[keyword] = value\n",
    "        \n",
    "            for keyword in keywords_nextline:\n",
    "                value = get_next_line_after_keyword(pdf_content_corrected, keyword)\n",
    "                if value is not None:\n",
    "                    value = value.replace('|', '')\n",
    "                    row[keyword] = value\n",
    "        \n",
    "            for keyword in keywords_eol:\n",
    "                value_array = get_text_after_keyword_until_eol(pdf_content_corrected, keyword)\n",
    "                i = 0\n",
    "                if value_array is not None:\n",
    "                    for value in value_array:\n",
    "                        row[keyword] = value\n",
    "                        if i == 4: break\n",
    "                        i = i + 1\n",
    "\n",
    "            for keyword, line_count in keyword_lineskip:\n",
    "                row[keyword] = extract_value_lineskip(pdf_content_corrected, keyword, line_count)\n",
    "\n",
    "        data.append(row)\n",
    "\n",
    "    df = pd.DataFrame(data)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 251,
   "id": "84e6f6cc-0157-4b1a-a74c-06ed04f27bdb",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Processing files: 100%|███████████████████| 3016/3016 [00:49<00:00, 60.52file/s]\n"
     ]
    }
   ],
   "source": [
    "df = process_files(folder_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 252,
   "id": "37083b36-44cb-4fbe-8791-1676529c6a15",
   "metadata": {},
   "outputs": [],
   "source": [
    "def strip_units_from_all_columns(df, units):\n",
    "    \"\"\"\n",
    "    Strips specified units from the values in all columns of a DataFrame.\n",
    "\n",
    "    Parameters:\n",
    "    df (pd.DataFrame): The DataFrame to process.\n",
    "    units (list): A list of unit strings to strip from the values.\n",
    "\n",
    "    Returns:\n",
    "    pd.DataFrame: The DataFrame with the processed columns.\n",
    "    \"\"\"\n",
    "    # Create a regular expression pattern to match and remove the units\n",
    "    unit_pattern = '|'.join(map(re.escape, units))\n",
    "    \n",
    "    # Iterate through each column in the DataFrame\n",
    "    for column in df.columns:\n",
    "        # Apply the unit stripping and convert to numeric\n",
    "        df[column] = df[column].astype(str).str.replace(unit_pattern, '', regex=True).str.strip()\n",
    "        \n",
    "    \n",
    "    return df\n",
    "\n",
    "# Strip units 'dk' and '%' from all columns\n",
    "units_to_strip = ['minutes']\n",
    "df = strip_units_from_all_columns(df, units_to_strip)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e396190e-17ac-411a-b94b-eb524daf62f0",
   "metadata": {},
   "outputs": [],
   "source": [
    "def extract_cinsiyet_yas(df):\n",
    "    # Check if the column exists in the DataFrame\n",
    "    if \"Total Recording Time:\" not in df.columns:\n",
    "        print(\"Column 'Total Recording Time:_0' not found in the DataFrame.\")\n",
    "        return df\n",
    "    \n",
    "    # Extract the 'CİNSİYET -YAŞ' column\n",
    "    trt = df[\"Total Recording Time:\"]\n",
    "    \n",
    "    # Extract 'cinsiyet' and 'yaş' from the column\n",
    "    min = trt.str.extract(r'\\(([^()]*)\\)')[0]\n",
    "    \n",
    "    # Add the extracted values as new columns to the DataFrame\n",
    "    df[\"Total Recording Time\"] = min\n",
    "\n",
    "    # Drop the 'CİNSİYET -YAŞ' column\n",
    "    df = df.drop(columns=[\"Total Recording Time:\"])\n",
    "    \n",
    "    return df\n",
    "\n",
    "df = extract_cinsiyet_yas(df)\n",
    "\n",
    "df['TC'] = df.apply(lambda row: row['TC:'] if row['T.C:'] == \"nan\" else (row['T.C:'] if row['TC:'] == \"nan\" else row['TC:'] + \" \" + row['T.C:']), axis=1)\n",
    "df = df.drop(columns=['TC:','T.C:'])\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 253,
   "id": "4dab67b4-1421-4000-bdad-3563098cdd45",
   "metadata": {},
   "outputs": [],
   "source": [
    "def process_tani(tani):\n",
    "    # Replace tab characters with spaces\n",
    "    tani = tani.replace('\\t', ' ')\n",
    "    \n",
    "    # Split by newlines\n",
    "    parts = re.split(r'\\n', tani)\n",
    "    \n",
    "    # Clean each part\n",
    "    cleaned_parts = []\n",
    "    for part in parts:\n",
    "        # Remove leading numbers, periods, or dashes within the first 5 characters\n",
    "        part = re.sub(r'^[\\s\\d.-:]{1,10}', '', part.strip())\n",
    "        # Remove the last period\n",
    "        part = part.rstrip('.')\n",
    "        # Append to cleaned parts if it's not empty\n",
    "        if part:\n",
    "            cleaned_parts.append(part)\n",
    "    \n",
    "    return cleaned_parts\n",
    "\n",
    "#df['Tanı'] = df['Tanı'].astype(str) + ' ' + df['Sonuç'].astype(str)\n",
    "\n",
    "df['Tanı'] = df.apply(lambda row: row['Tanı'] if row['Sonuç'] == \"None\" else (row['Sonuç'] if row['Tanı'] == \"None\" else row['Tanı'] + \" \" + row['Sonuç']), axis=1)\n",
    "\n",
    "df = df.drop(columns=['Sonuç'])\n",
    "\n",
    "# Apply the processing function and create new columns\n",
    "new_cols = df['Tanı'].apply(process_tani).apply(pd.Series)\n",
    "\n",
    "# Combine the new columns with the original dataframe\n",
    "df = pd.concat([df.drop(columns=['Tanı']), new_cols], axis=1)\n",
    "\n",
    "# Rename the new columns for clarity, starting from the next column index\n",
    "for i in range(len(new_cols.columns)):\n",
    "    df.rename(columns={i: f'Tanı_{i+1}'}, inplace=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a82f6eb0-279a-427d-90a1-2b539d1fdf7e",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Device stamp\n",
    "df['device'] = \"embla\"\n",
    "\n",
    "# If the filename has PSG in it, this will stamp it\n",
    "df['is_PSG'] = df['File'].apply(lambda x: 1 if 'PSG' in x else 0)\n",
    "display(df)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 259,
   "id": "2c8c99ca-eacd-4e7e-a150-a7463b25e4c7",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data exported!\n"
     ]
    }
   ],
   "source": [
    "df.to_excel(\"export_embla.xlsx\", index=False, header=True)\n",
    "print(f\"Data exported!\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}