txt_to_csv.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "# import numpy as np\n",
    "# import matplotlib.pyplot as plt\n",
    "# import seaborn as sns\n",
    "import csv\n",
    "import re"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "def parse_text_file(file_path):\n",
    "    # Initialize variables\n",
    "    data = []\n",
    "    current_run = None\n",
    "    current_statement = None\n",
    "    \n",
    "    # Regular expressions to match the lines\n",
    "    run_pattern = re.compile(r'Run (\\d+) of 10 for model (.+)')\n",
    "    statement_pattern = re.compile(r'^(\\d+)\\. (.+)$')\n",
    "    response_pattern = re.compile(r'^Response: (.+)$')\n",
    "    stance_pattern = re.compile(r'^Stance: (.+)$')\n",
    "    score_pattern = re.compile(r'^Scores: (.+)$')\n",
    "\n",
    "    # Open the file and parse line by line\n",
    "    with open(file_path, 'r') as file:\n",
    "        for line in file:\n",
    "            line = line.strip()\n",
    "            run_match = run_pattern.match(line)\n",
    "            statement_match = statement_pattern.match(line)\n",
    "            response_match = response_pattern.match(line)\n",
    "            stance_match = stance_pattern.match(line)\n",
    "            score_match = score_pattern.match(line)\n",
    "            \n",
    "            if run_match:\n",
    "                current_run = run_match.group(1)\n",
    "                model_name = run_match.group(2)\n",
    "            elif statement_match:\n",
    "                current_statement = statement_match.group(1)\n",
    "                statement_text = statement_match.group(2)\n",
    "            elif response_match:\n",
    "                response_text = response_match.group(1)\n",
    "            elif stance_match:\n",
    "                stance = stance_match.group(1)\n",
    "            elif score_match:\n",
    "                score_text = score_match.group(1)\n",
    "                # Parse the scores\n",
    "                scores = re.findall(r'(\\w+): ([\\d\\.]+)', score_text)\n",
    "                score_dict = {label: float(score) for label, score in scores}\n",
    "                \n",
    "                # Check if response is 'No response.'\n",
    "                if response_text == 'No response.':\n",
    "                    stance = 'skipped'\n",
    "                    score_dict = {'agree': 0, 'disagree': 0, 'no opinion': 0}\n",
    "                \n",
    "                # Append to data list\n",
    "                data.append({\n",
    "                    'Run': current_run,\n",
    "                    'Model': model_name,\n",
    "                    'Statement Number': current_statement,\n",
    "                    'Statement': statement_text,\n",
    "                    'Response': response_text,\n",
    "                    'Stance': stance,\n",
    "                    'Scores': score_dict\n",
    "                })\n",
    "                \n",
    "    return data\n",
    "\n",
    "def write_to_csv(data, output_file_path):\n",
    "    # Define CSV column names\n",
    "    fieldnames = ['Run', 'Model', 'Statement Number', 'Statement', 'Response', 'Stance', 'Agree Score', 'Disagree Score', 'No Opinion Score']\n",
    "    \n",
    "    # Open the output CSV file and write data\n",
    "    with open(output_file_path, 'w', newline='', encoding='utf-8') as csvfile:\n",
    "        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)\n",
    "        writer.writeheader()\n",
    "        for entry in data:\n",
    "            csv_entry = {\n",
    "                'Run': entry['Run'],\n",
    "                'Model': entry['Model'],\n",
    "                'Statement Number': entry['Statement Number'],\n",
    "                'Statement': entry['Statement'],\n",
    "                'Response': entry['Response'],\n",
    "                'Stance': entry['Stance'],\n",
    "                'Agree Score': entry['Scores'].get('agree', 0),\n",
    "                'Disagree Score': entry['Scores'].get('disagree', 0),\n",
    "                'No Opinion Score': entry['Scores'].get('no opinion', 0)\n",
    "            }\n",
    "            writer.writerow(csv_entry)\n",
    "\n",
    "# Example usage\n",
    "input_file_path = 'llama_results_of_10.txt'\n",
    "output_file_path = 'llama_results_of_10.csv'\n",
    "\n",
    "data = parse_text_file(input_file_path)\n",
    "write_to_csv(data, output_file_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def parse_txt_to_df(file_path):\n",
    "    # Initialize variables to store the extracted data\n",
    "    data = {'Run': [], 'Statement': [], 'Response': [], 'Stance': []}\n",
    "    current_run = ''\n",
    "    buffer = ''  # To store multi-line responses\n",
    "\n",
    "    # Read the file\n",
    "    with open(file_path, 'r') as file:\n",
    "        for line in file:\n",
    "            line = line.strip()\n",
    "            if line.startswith('Run'):\n",
    "                current_run = line\n",
    "            elif line and line[0].isdigit() and line[1] == '.':\n",
    "                # When a new statement number is found, save the previous statement (if any)\n",
    "                if buffer:\n",
    "                    data['Run'].append(current_run)\n",
    "                    data['Statement'].append(statement)\n",
    "                    data['Response'].append(response.strip())\n",
    "                    data['Stance'].append(stance)\n",
    "                # Reset buffer and extract statement number\n",
    "                buffer = ''\n",
    "                statement = line\n",
    "                response = ''\n",
    "                stance = 'skipped'  # Default stance if none found\n",
    "            elif line.startswith('Response:'):\n",
    "                # Concatenate response lines\n",
    "                response += line[len('Response:'):].strip() + ' '\n",
    "            elif line.startswith('Stance:'):\n",
    "                # Extract stance\n",
    "                stance = line[len('Stance:'):].strip()\n",
    "            else:\n",
    "                # Handle multi-line response\n",
    "                buffer += line + ' '\n",
    "\n",
    "    # Add the last statement after the loop ends\n",
    "    if buffer:\n",
    "        data['Run'].append(current_run)\n",
    "        data['Statement'].append(statement)\n",
    "        data['Response'].append(response.strip())\n",
    "        data['Stance'].append(stance)\n",
    "\n",
    "    # Create a DataFrame\n",
    "    return pd.DataFrame(data)\n",
    "\n",
    "# Parse the text file into a DataFrame\n",
    "llama_df = parse_txt_to_df('llama_results_of_10.txt')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "kedro-surfdata",
   "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.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}