Delete utils

utils/app_common.py

@@ -1,205 +0,0 @@
-import re
-from bson import ObjectId
-from dotenv import load_dotenv
-import os
-from interviewer_utils.interviewer import Interviewer
-from utils.study import Study
-from utils.study_repository import StudyRepository
-from typing import Union, List
-from utils.database_helper import DatabaseIO
-import gradio as gr
-
-load_dotenv()
-INTERVIEW_USERS_COLLECTION = os.environ['MONGO_COLLECTION_USERS']
-RESEARCHER_USERS_COLLECTION = os.environ['MONGO_COLLECTION_USERS']
-STUDIES_COLLECTION = os.environ['MONGO_COLLECTION_STUDIES']
-
-
-def is_valid_email(email):
-    # Simple regex pattern for email validation
-    pattern = r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$"
-    return bool(re.match(pattern, email))
-
-
-def set_active_study(this_study_repository,
-                     study_id):
-    """
-    Using a repository object creates a study object for the given id
-    returns the study object and a status message
-    """
-
-    study_id = study_id.split(':')[0]
-    this_study = this_study_repository.get_studies(study_id=study_id)
-
-    if not this_study:
-        return {}, "study does not exist"
-
-    if type(this_study) == list and len(this_study):
-        this_study = this_study[0]
-
-    # initialize a new study with this study in memory!
-    try:
-        # study validation is here! if db schema is older, will be notified during dev
-        active_study_obj = Study(**this_study)
-    except Exception as e:
-        return None, f"Study not found due to {e}"
-
-    return active_study_obj, ""
-
-
-def configure_interviewer(this_interviewer: Interviewer,
-                          this_study_obj: Study,
-                          history: List[List[str]]):
-    """
-    takes an existing interviewer object and initializes it with details from the provided study object
-
-    :param history:
-    :param this_interviewer:
-    :param this_study_obj:
-    :return: interviewer object with the right initializations
-    """
-    this_status = "Successfully configured interviewer object"
-
-    if this_study_obj is None or type(this_study_obj) != Study:
-        this_status = "The research study you are participating in does not" \
-                      " exist or is not ready, please contact the researcher"
-
-    # if this study has
-
-    return this_interviewer, this_status
-
-
-def lookup_userid_by_email(email):
-    """
-    Looks up email address and finds the right user and returns the user_id
-    :param email:
-    :return:
-    """
-    if email is None or not is_valid_email(email):
-        return None, "Try with a valid email please!"
-
-    with DatabaseIO(collection_name=INTERVIEW_USERS_COLLECTION) as db_io:
-        user = db_io.collection.find_one({"email": email}, {"_id": 1})
-        if user:
-            return str(user["_id"]), "Found correct user"
-        else:
-            return None, "Error finding user"
-
-
-def process_transcript_to_chatbot_history(this_transcript):
-    messages_in_chatbot_format = []
-    for i in range(0, len(this_transcript), 2):
-        human_turn = this_transcript[i]['text']
-        ai_turn = this_transcript[i + 1]['text']
-        messages_in_chatbot_format.append([human_turn, ai_turn])
-    return messages_in_chatbot_format
-
-
-def set_user_and_study(
-        this_study_repository,
-        this_study_code,
-        this_user_email,
-
-):
-    this_user_id, user_exists_message = lookup_userid_by_email(this_user_email)
-
-    interaction_disabled = gr.update( interactive=False)
-    interaction_enabled = gr.update( interactive=True)
-
-    if not this_user_id or not len(this_user_id):
-        this_return_message = user_exists_message
-        return None, "", this_return_message, *([interaction_disabled] * 5)
-
-    this_active_study, study_exists_message = set_active_study(this_study_repository, study_id=str(this_study_code))
-    if not this_active_study:
-        this_return_message = study_exists_message
-        return {}, this_user_id, this_return_message, *([interaction_disabled] * 5)
-    this_return_message = user_exists_message + " " + study_exists_message
-
-    return this_active_study, this_user_id, this_return_message, *([interaction_enabled] * 5)
-
-
-def setup_conversation_context(this_active_study, this_study_repository, this_user_id):
-    """
-    Populates the interview object
-    :param this_study_repository:
-    :param this_active_study:
-    :param this_user_id:
-    :return:
-    """
-
-    if not this_active_study or not type(this_active_study) == Study:
-        return [], None, "Study code or user id is invalid"
-    this_transcript = load_previous_transcript(this_active_study,
-                                                         this_user_id,
-                                                         this_study_repository)
-    this_user_id_str = str(this_user_id)
-    this_interviewer = Interviewer(
-        research_topic=this_active_study.study_topic,
-        research_objectives=this_active_study.study_objective,
-        research_questions=this_active_study.research_questions,
-        interview_guidelines=this_active_study.interview_guidelines,
-        existing_conversation_history=this_transcript
-    )
-
-    if not this_transcript or not len(this_transcript):
-        # no previous conversation
-        ai_response = this_interviewer.chat_handler.converse('Hello')
-        this_constructed_transcript = [{'turn': 'Human', 'text': 'Hello'}, {'turn': 'AI', 'text': ai_response}]
-        # transcript did not exist, so must create the user info in database
-
-        this_view_chat_history = process_transcript_to_chatbot_history(this_constructed_transcript)
-        #
-        if 'human_interviews' not in this_active_study.interviews or not this_active_study.interviews['human_interviews']:
-            this_active_study.interviews['human_interviews'] = {}
-        this_previous_interview = this_active_study.interviews['human_interviews'].get(this_user_id)
-        # If not found, initialize an empty one
-        if not this_previous_interview:
-            this_previous_interview = {
-                "AI_interviewer": "Sophia (AI Researcher)",
-                "Human_Respondent": {
-                    "participant_id": this_user_id_str
-                },
-                "transcript": []
-            }
-            this_active_study.interviews['human_interviews'][this_user_id_str] = this_previous_interview
-        this_active_study.interviews['human_interviews'][str(this_user_id)]['transcript'] = this_constructed_transcript
-        this_study_repository.update_study(study_id=this_active_study._id, updated_data=this_active_study.to_dict())
-        this_message = "First time talking to you ... "
-    else:
-        this_view_chat_history = process_transcript_to_chatbot_history(this_transcript)
-        this_message = "Welcome back, lets start where we left off!"
-
-    return this_view_chat_history, this_interviewer, this_message
-
-
-def load_previous_transcript(this_study: Study,
-                             this_user_id: Union[ObjectId | str],
-                             this_study_repository: StudyRepository):
-    """
-    loads existing interview
-    :param this_study:
-    :param this_user_id:
-    :param this_study_repository
-    :return:
-    """
-    this_user_id_str = str(this_user_id)
-
-    # Ensure 'human_interviews' exists and is initialized properly
-
-    try:
-        # Try to get the interview of the participant
-        human_interviews = this_study.interviews['human_interviews']
-        this_previous_interview = human_interviews.get(this_user_id_str, None)
-        return this_previous_interview['transcript']
-
-    except (AttributeError, KeyError, Exception) as ae:
-        return []
-
-
-if __name__ == "__main__":
-    # unit tests
-    study_repository = StudyRepository(collection_name_studies=STUDIES_COLLECTION)
-    sample_study_id = "64dc833a12e518bb5d4ea2b4"
-    sample_email = "paudyalprajwal@qualz.net"
-    interviewer = Interviewer(existing_conversation_history=['a', 'b'])

utils/database_helper.py

@@ -1,92 +0,0 @@
-from datetime import datetime
-
-import pymongo
-from pymongo.errors import PyMongoError
-import os
-import dotenv
-from pymongo.server_api import ServerApi
-
-class DatabaseIO:
-    def __init__(self, db_name=None, collection_name=None):
-        dotenv.load_dotenv()
-
-        mongo_username = os.environ['MONGO_USERNAME']
-        mongo_password = os.environ['MONGO_PASSWORD']
-        client_url = os.environ['MONGO_CLIENT_URL_DEV']
-        uri  = f"mongodb+srv://{mongo_username}:{mongo_password}@{client_url}/?retryWrites=true&w=majority"
-
-        if not db_name:
-            db_name = os.environ['MONGO_DATABASE_NAME']
-        if not collection_name:
-            collection_name = os.environ['MONGO_COLLECTION']
-
-        self.client = pymongo.MongoClient(uri, server_api=ServerApi('1'))
-        self.db = self.client[db_name]
-        self.collection = self.db[collection_name]
-
-    def insert_document(self, article, collection=None, unique_on='_id', upsert=False):
-
-        if not collection:
-            collection = self.collection
-
-        article['date_modified'] = datetime.now().utcnow()
-        existing_document = collection.find_one({unique_on: article[unique_on]})
-        if existing_document:
-            # there is something like with the reddit id already
-
-            if upsert:
-                collection.update_one({unique_on: article[unique_on]}, {"$set": article})
-        else:
-            article['date_created'] = datetime.now().utcnow()
-            collection.insert_one(article)
-
-    def __enter__(self):
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        try:
-            self.client.close()
-        except PyMongoError as e:
-            print(f"An error occurred while closing the database connection: {e}")
-            raise
-
-    def __del__(self):
-        try:
-            self.client.close()
-        except Exception as e:
-            print(e)
-
-    def read_documents(self, query=None, sort_by=None, sort_order=None):
-        if query is None:
-            query = {}
-        try:
-            if sort_by:
-                if not sort_order or sort_order not in [1, -1]:
-                    sort_order= 1
-                for article in self.collection.find(query).sort(sort_by, sort_order):
-                    yield article
-            else:
-                for article in self.collection.find(query):
-                    yield article
-        except Exception as e:
-            print(e)
-
-    def count_documents(self, query=None):
-        if query is None:
-            query = {}
-        try:
-            return self.collection.count_documents(query)
-        except Exception as e:
-            print(e)
-
-    def update_documents(self, query, update, upsert=True):
-        try:
-            self.collection.update_one(query, update, upsert=upsert)
-        except Exception as e:
-            print(e)
-
-    def delete_document(self, query):
-        try:
-            self.collection.delete_one(query)
-        except Exception as e:
-            print(e)

utils/eleven_voice_cache.py

@@ -1,65 +0,0 @@
-# Save default voices to avoid querying the API for unathorized users
-
-import os
-import re
-from typing import Iterator, List, Optional, Union
-
-from elevenlabs.api import TTS, Model, Voice, VoiceClone, Voices, VoiceSettings
-
-
-VOICES_CACHE = [
-    Voice(
-        voice_id="21m00Tcm4TlvDq8ikWAM",
-        name="Rachel",
-        category="premade",
-        settings=VoiceSettings(stability=0.75, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="AZnzlk1XvdvUeBnXmlld",
-        name="Domi",
-        category="premade",
-        settings=VoiceSettings(stability=0.1, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="EXAVITQu4vr4xnSDxMaL",
-        name="Bella",
-        category="premade",
-        settings=VoiceSettings(stability=0.245, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="ErXwobaYiN019PkySvjV",
-        name="Antoni",
-        category="premade",
-        settings=VoiceSettings(stability=0.195, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="MF3mGyEYCl7XYWbV9V6O",
-        name="Elli",
-        category="premade",
-        settings=VoiceSettings(stability=0.755, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="TxGEqnHWrfWFTfGW9XjX",
-        name="Josh",
-        category="premade",
-        settings=VoiceSettings(stability=0.15, similarity_boost=0.51),
-    ),
-    Voice(
-        voice_id="VR6AewLTigWG4xSOukaG",
-        name="Arnold",
-        category="premade",
-        settings=VoiceSettings(stability=0.15, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="pNInz6obpgDQGcFmaJgB",
-        name="Adam",
-        category="premade",
-        settings=VoiceSettings(stability=0.2, similarity_boost=0.75),
-    ),
-    Voice(
-        voice_id="yoZ06aMxZJJ28mfd3POQ",
-        name="Sam",
-        category="premade",
-        settings=VoiceSettings(stability=0.25, similarity_boost=0.75),
-    ),
-]

utils/llm_helper.py

@@ -1,243 +0,0 @@
-from langchain.schema import SystemMessage, HumanMessage
-from langchain.chat_models import ChatOpenAI
-import ast
-import warnings
-from pathlib import Path
-import os, json
-from ra_utils.synthetic_panel_datamodel import parse_synthetic_panel_output
-from langchain.memory import ConversationBufferMemory
-from langchain import LLMChain
-import dotenv
-
-
-class LLMChatHandler:
-    def __init__(self,
-                 llm_model_name: str = None,
-                 llm_temperature: float = None,
-                 expertise_message: str = ""):
-
-        dotenv.load_dotenv()
-        if not llm_model_name:
-            llm_model_name = os.environ['OPENAI_MODEL']
-        if not llm_temperature:
-            llm_temperature=os.environ['OPENAI_MODEL_TEMPERATURE']
-
-        self.llm = ChatOpenAI(temperature=llm_temperature, model_name=llm_model_name)
-
-        project_dir = Path(os.environ['PROJECT_DIR'])
-        assert project_dir.exists()
-        with open(project_dir / 'study_prompts.json') as sp:
-            self.definition_objective_maps = json.load(sp)
-
-        if not len(expertise_message):
-            expertise_message = f"Assistant is a world renowned expert in qualitative research." \
-                                f" Assistant is well known in your field worldwide and has published several books " \
-                                f"and white-papers. Assistant is also an expert qualitative research analyst " \
-                                f"who is thorough and methodical."
-        self.expertise_message = expertise_message
-
-    def predict(self, messages):
-        return self.llm.predict_messages(messages)
-
-    def guess_research_questions(self,
-                                 research_name: str,
-                                 research_topic: str,
-                                 research_objective: str,
-                                 research_framework: str,
-                                 research_questions_draft: str):
-        guess_research_questions_message = f"""
-         Based on the study named {research_name} focusing on the topic {research_topic}, and objectives for the 
-         research as {research_objective} and a theoretical framework as {research_framework}, please help 
-         me craft some research questions. Research questions are overarching broad questions that help me understand 
-         about the research objective. You shouldn't suggest more than 3 questions.
-         Also take note of the specific instructions or draft enclosed within
-         triple backticks. 
-         ```INSTRUCTIONS: {research_questions_draft}```
-         """ + """   Your output should be formatted as a dictionary with two keys such as:
-                          {
-                     "research_questions": A single string with all the research questions you recommend 
-                     separated by a semi-colon. 
-                     It is critical to include only questions here and not any explanation or discussion
-                     "reasoning": A brief explanation on the reasoning for choosing the questions.
-                      Here you can include any additional things you 
-                     want to add including commentary of any kind
-                     } 
-
-                     If there are any concerns or difficulties in crafting the questions, 
-                     please mention them under "reasoning" """
-        messages = [SystemMessage(content=self.expertise_message),
-                    HumanMessage(content=guess_research_questions_message)]
-        research_questions_output = self.predict(messages)
-
-        try:
-
-            research_questions_dict = ast.literal_eval(research_questions_output.content)
-            if not isinstance(research_questions_dict, dict):
-                raise ValueError("Parsing output as dictionary did not work")
-
-            return research_questions_dict.get("research_questions", ""), \
-                research_questions_dict.get("reasoning", "Please try again. Are the study details descriptive enough?")
-        except (SyntaxError, ValueError) as e:
-            print(f"Error parsing output: {e}")
-            return "", "Please try again. Are the study details descriptive enough?"
-
-    def create_synthetic_panel(self,
-                               research_topic: str,
-                               research_objective: str,
-                               research_framework: str,
-                               instructions: str = "",
-                               num_panelists: int = 5):
-
-        panel_creation_prompt_template = self.definition_objective_maps.get('sample_selection').get('prompt_template')
-
-        if num_panelists > 15:
-            warnings.warn("Only returning 15 panelists for now")
-            num_panelists = 15
-
-        m_custom_message = f"The profiles should be created for a qualitative research study with these details. " \
-                           f"Study Topic: {research_topic}" \
-                           f"Study Objective: {research_objective}" \
-                           f"Study theoretical framework: {research_framework}" \
-                           f"Please create {num_panelists} of profiles and consider these " \
-                           f"instructions while creating the included within triple backticks: " \
-                           f"```INSTRUCTIONS FOR PROFILE CREATION: {instructions}```."
-
-        # existing_synthetic_panelist_message = ""
-        # existing_panelists = self.synthetic_panelists
-        #
-        # if existing_panelists and len(existing_panelists) > 0:
-        #     existing_synthetic_panelist_message = " and Here is a list of existing panelists " \
-        #                                           "that should be modified according to the criteria I will state"
-
-        # panel_creation_message = panel_creation_prompt_template + existing_synthetic_panelist_message + m_custom_message
-
-        panel_creation_message = panel_creation_prompt_template + m_custom_message
-
-        messages = [SystemMessage(content=self.expertise_message),
-                    HumanMessage(content=panel_creation_message)]
-        synthetic_panel_output = self.predict(messages).content
-
-        user_panel, rationale, discussion = parse_synthetic_panel_output(synthetic_panel_output)
-
-        return user_panel, rationale, discussion
-
-    def guess_interview_guidelines(self,
-                                   research_topic: str,
-                                   research_objective: str,
-                                   research_framework: str,
-                                   research_questions: str,
-                                   interview_guidelines_draft_or_instructions: str, ):
-        # Chat logic here
-        guess_study_objective_message = f"""
-         For a qualitative study focusing on the topic {research_topic}, please help 
-         come up with some interview guidelines and script. 
-         The objective of the research is to {research_objective}
-         The theoretical framework for the study is {research_framework}
-         The main research questions we have identified are: {research_questions}. 
-         I will also include some draft question or additional instructions below within 
-         triple backticks. Pay attention to these and use them accordingly. 
-         Your output should be a dictionary with two keys: 
-
-         1. "interview_script": Provide a script to guide the interview process. 
-         The purpose is for for a researcher to use this to guide an interview session. The aim of the script is to
-         ensure appropriate questions as asked to fulfill the study objective and research questions. The interview
-         script should be a plain string and NOT a dictionary or any other data structure. If any sub-sections 
-         are needed, use call caps for the subsection name. 
-
-         2. "reasoning": This should be an overview of the rationale behind the questions, and be no longer than two 
-         sentences. 
-
-         If there are any concerns or difficulties in crafting the interview script, please mention them under "reasoning" 
-
-         ```INSTRUCTIONS: {interview_guidelines_draft_or_instructions}```"""
-
-        messages = [SystemMessage(content=self.expertise_message),
-                    HumanMessage(content=guess_study_objective_message)]
-        study_objective_output = self.predict(messages)
-        try:
-
-            study_objective_dict = ast.literal_eval(study_objective_output.content)
-            if not isinstance(study_objective_dict, dict):
-                raise ValueError("Parsing output as dictionary did not work")
-
-            return study_objective_dict.get("interview_script", ""), \
-                study_objective_dict.get("reasoning", "Please try again. Are the study details descriptive enough?")
-        except (SyntaxError, ValueError) as e:
-            print(f"Error parsing output: {e}")
-            return "", "Please try again. Are the study details descriptive enough?"
-
-    def guess_study_objective(self, research_name: str,
-                              research_topic: str,
-                              study_objective_draft_or_instructions: str):
-        # Chat logic here
-        guess_study_objective_message = f"""
-         Based on the study named {research_name} focusing on the topic {research_topic}, please help 
-         me improve my study objective. Take note of the specific instructions and draft enclosed within
-         triple backticks. Your output should be a dictionary with two keys: 
-
-         1. "study_objective": Provide a concise research objective statement that is limited to one sentence. 
-
-         2. "reasoning": This should detail the rationale behind the chosen objective, and be no longer than two 
-         sentences. 
-
-         If there are any concerns or difficulties in crafting the objective, please mention them under "reasoning" 
-
-         ```INSTRUCTIONS: {study_objective_draft_or_instructions}```"""
-
-        messages = [SystemMessage(content=self.expertise_message),
-                    HumanMessage(content=guess_study_objective_message)]
-        study_objective_output = self.predict(messages)
-        try:
-
-            study_objective_dict = ast.literal_eval(study_objective_output.content)
-            if not isinstance(study_objective_dict, dict):
-                raise ValueError("Parsing output as dictionary did not work")
-
-            return study_objective_dict.get("study_objective", ""), \
-                study_objective_dict.get("reasoning", "Please try again. Are the study details descriptive enough?")
-        except (SyntaxError, ValueError) as e:
-            print(f"Error parsing output: {e}")
-            return "", "Please try again. Are the study details descriptive enough?"
-
-    def guess_theoretical_framework(self, research_name: str,
-                                    research_topic: str,
-                                    research_objective: str):
-        # Chat logic here
-        guess_theoretical_framework_message = f"""
-                     Based on the given research details:
-                     - Research Name: {research_name}
-                     - Research Topic: {research_topic}
-                     - Study Objective: {research_objective}
-
-                     Please recommend a suitable theoretical framework for the study. The framework should be
-                      one among the following: Ethnographic, Narrative, 
-                      Grounded Theory, Case Study, Phenomenological, and Exploratory. """ + """You output should 
-                      be a python dictionary with two keys and nothing else. Here is the format: 
-                      {
-                     "theoretical_framework": Recommended framework from the list above. 
-                     It is critical to include only the framework and not any explanation
-                     "reasoning": A brief explanation on why you think the chosen 
-                     framework is the most suitable for the study. Here you can include any additional things you 
-                     want to add including commentary of any kind
-                     } 
-
-                     If there are any concerns or difficulties in crafting the objective, 
-                     please mention them under "reasoning" 
-                     """
-
-        messages = [SystemMessage(content=self.expertise_message),
-                    HumanMessage(content=guess_theoretical_framework_message)]
-        study_guess_theoretical_framework_output = self.predict(messages)
-        try:
-
-            study_guess_theoretical_framework_dict = ast.literal_eval(study_guess_theoretical_framework_output.content)
-            if not isinstance(study_guess_theoretical_framework_dict, dict):
-                raise ValueError("Parsing output as dictionary did not work")
-
-            return study_guess_theoretical_framework_dict.get("theoretical_framework", ""), \
-                study_guess_theoretical_framework_dict.get("reasoning",
-                                                           "Please try again. "
-                                                           "Are the study details descriptive enough?")
-        except (SyntaxError, ValueError) as e:
-            print(f"Error parsing output: {e}")
-            return "", "Please try again. Are the study details descriptive enough?"

utils/study.py

@@ -1,148 +0,0 @@
-from typing import List, Dict, Optional, Union
-from pathlib import Path
-import os
-import json
-import warnings
-from langchain.chat_models import ChatOpenAI
-from bson import ObjectId
-from utils.llm_helper import LLMChatHandler
-
-
-class InterviewSyntheticPanel:
-    def __init__(self, api_config: dict):
-        # Initialize API
-        pass
-
-    def create_panel(self, study_details: dict):
-        # Call the API with study details and create synthetic panel
-        pass
-
-
-class Study:
-    def __init__(self,
-                 _id: str,
-                 study_descriptive_name: str,
-                 study_objective: str,
-                 study_framework: str,
-                 study_topic: str,
-                 research_questions: str,
-                 interview_guidelines: Optional[str] = None,
-                 interviews: Optional[Dict['str', List[Dict]]] = None,
-                 overall_codes: Optional[Dict[str, List[str]]] = None,
-                 synthetic_panel_pending_interviews: Optional [List[Dict]] = None,
-                 **kwargs,
-                 ):
-        self.study_descriptive_name = study_descriptive_name
-        self.study_topic = study_topic
-        self._id = _id
-        self.study_objective = study_objective
-        self.study_framework = study_framework
-        self.interview_guidelines = interview_guidelines if interview_guidelines else ""
-        self.research_questions = research_questions
-        self.interviews = interviews if interviews else {}
-        self.overall_codes = overall_codes if overall_codes else {}
-
-        self.additional_fields = kwargs
-        self.chat_handler = LLMChatHandler()
-
-        self.synthetic_panel_pending_interviews = synthetic_panel_pending_interviews if\
-            synthetic_panel_pending_interviews else []
-        # Check if synthetic_interviews exists and is not empty
-        # if 'synthetic_interviews' in self.interviews and len(self.interviews.get('synthetic_interviews', "")):
-        #     # Extract AI_Respondent from each item in synthetic_interviews and append to self.synthetic_panelists
-        #     self.synthetic_panelists = [item['AI_Respondent'] for item in interviews.get('synthetic_interviews') if
-        #                                 'AI_Respondent' in item]
-
-        project_dir = Path(os.environ['PROJECT_DIR'])
-        assert project_dir.exists()
-        with open(project_dir / 'study_prompts.json') as sp:
-            self.definition_objective_maps = json.load(sp)
-
-    def add_interview(self, interview: Dict):
-        """
-        Add an interview to the study.
-
-        :param interview: The interview data to add.
-        """
-        self.interviews.append(interview)
-
-
-    def save_synthetic_panelists(self):
-        """
-        attemp
-        :return:
-        """
-    def remove_interview(self, interview_index: int):
-        """
-        Remove an interview from the study based on its index.
-
-        :param interview_index: Index of the interview to remove.
-        """
-        if 0 <= interview_index < len(self.interviews):
-            del self.interviews[interview_index]
-
-    def update_overall_codes(self, code: str, quotes: List[str]):
-        """
-        Update or add a code and its associated quotes to the overall_codes.
-
-        :param code: The code to update or add.
-        :param quotes: The associated quotes.
-        """
-        if code in self.overall_codes:
-            self.overall_codes[code].extend(quotes)
-        else:
-            self.overall_codes[code] = quotes
-
-    def to_dict(self) -> dict:
-        """
-        Convert the object to a dictionary representation.
-
-        Returns:
-            dict: A dictionary representation of the object.
-        """
-        return {
-            '_id': self._id,
-            'study_descriptive_name': self.study_descriptive_name,
-            'study_topic': self.study_topic,
-            'study_objective': self.study_objective,
-            'study_framework': self.study_framework,
-            'interview_guidelines': self.interview_guidelines,
-            'research_questions': self.research_questions,
-            'interviews': self.interviews,
-            'overall_codes': self.overall_codes,
-            'synthetic_panel_pending_interviews': self.synthetic_panel_pending_interviews,
-            **self.additional_fields
-        }
-
-
-    @staticmethod
-    def from_dict(data: Dict):
-        """
-        Create a Study object from a dictionary.
-
-        :param data: The dictionary containing study data.
-        :return: A Study object.
-        """
-        return Study(
-            study_id=data["id"],
-            objective=data["objective"],
-            study_framework=data["study_framework"],
-            study_descriptive_name=data["study_descriptive_name"],
-            study_topic=data["study_topic"],
-            interviews=data.get("interviews", []),
-            overall_codes=data.get("overall_codes", {})
-        )
-
-
-# Usage
-# study_data = {
-#     "id": "12345",
-#     "objective": "Test Objective",
-#     "study_framework": "Framework1",
-#     "study_descriptive_name": "Study1",
-#     "study_topic": "Topic1",
-#     "interviews": [],
-#     "overall_codes": {}
-# }
-# study = Study.from_dict(study_data)
-# study_dict = study.to_dict()

utils/study_repository.py

@@ -1,200 +0,0 @@
-import dotenv
-
-import os
-
-from typing import Optional, List, Tuple
-from typing import Dict, Any
-from utils.study import Study
-from utils.database_helper import DatabaseIO
-from bson import ObjectId
-from typing import Union
-
-from utils.llm_helper import LLMChatHandler
-
-
-class StudyRepository:
-    def __init__(self,
-                 db_name=None,
-                 collection_name_studies=None,
-                 collection_name_users=None
-                 ):
-        """
-        Initialize the StudyRepository with a database connection or client.
-
-        :param db: A mock database connection or client.
-        """
-
-        dotenv.load_dotenv()
-
-        if not db_name:
-            db_name = os.environ['MONGO_DATABASE_NAME']
-        if not collection_name_studies:
-            collection_name_studies = os.environ['MONGO_COLLECTION_STUDIES']
-        if not collection_name_users:
-            collection_name_users = os.environ['MONGO_COLLECTION_USERS']
-
-
-
-        self.db_name = db_name
-        self.collection_name_studies = collection_name_studies
-        self.collection_name_users = collection_name_users
-        self.chat_handler = LLMChatHandler()
-        # self.db = self.client[db_name]
-        # self.collection = self.db.get_collection(self.collection_name)  # Assuming 'study_v03' is the collection name
-
-    def create_study(self, study_data: Dict[str, Any]) -> Dict[str, Any]:
-        # Validate the study data using the Study's constructor
-        study_data_with_default_id = study_data.copy()
-        study_data_with_default_id["_id"] = None
-
-        try:
-            study_instance = Study(**study_data_with_default_id)
-        except TypeError as e:
-            # Here, we're assuming the TypeError arises from a mismatch in the fields.
-            # This might not cover all validation cases, but it's a start.
-            raise ValueError(f"Validation error: {str(e)}")
-        except Exception as e:
-            raise Exception(f"Other exception: {str(e)}")
-
-        # If validation passes, insert the study into the database
-
-        with DatabaseIO(self.db_name, self.collection_name_studies) as db_io:
-            study_id = db_io.collection.insert_one(study_data).inserted_id
-            return study_id
-
-    def append_to_transcript(self, study_id, user_id, human_text, ai_response):
-        """
-        Append the recent conversation to the transcript in the database.
-
-        :param db_url: MongoDB database URL.
-        :param db_name: Name of the MongoDB database.
-        :param collection_name: Name of the collection storing the transcripts.
-        :param participant_id: ID of the participant for whom the transcript is being updated.
-        :param human_text: The text/message from the human.
-        :param ai_response: The response from the AI.
-        """
-
-        # Initialize the MongoDB client and select the database and collection
-        user_id = str(user_id)
-        # Define the conversation entries to be appended
-        new_entries = [
-            {
-                "turn": "Human",
-                "text": human_text,
-            },
-            {
-                "turn": "AI",
-                "text": ai_response,
-            }
-        ]
-
-        # Append new entries to the transcript
-        with DatabaseIO(collection_name=self.collection_name_studies) as db_io:
-            query = {"_id": ObjectId(study_id)}
-
-            # Append new entries to the transcript using a dynamic field update
-            update_field = f"interviews.human_interviews.{user_id}.transcript"
-            update_action = {"$push": {update_field: {"$each": new_entries}}}
-
-            # Execute the update
-            db_io.collection.update_one(query, update_action)
-
-
-    def get_studies(self,
-                    study_id: Optional[str] = None,
-                    user_id: Optional[str] = None,
-                    to_filter_by_user_id: bool = True) -> Union[
-        List[Dict], Dict, Tuple[List[Dict], str], Tuple[int, str]]:
-        """Retrieve studies from the database.
-
-        :param study_id: (Optional) If provided, retrieve only the study with this ID.
-        :param user_id: (Optional) If provided, filter results to studies the user is authorized to view.
-        :param to_filter_by_user_id: (Optional) If False, ignores the user_id and
-        returns all studies (todo remove in prod)
-        :return: The studies or a specific study data, or a tuple with a message if an error occurs.
-        """
-
-        if study_id:
-            study_id = ObjectId(study_id)
-            with DatabaseIO(db_name=self.db_name, collection_name=self.collection_name_studies) as db_io:
-                study = db_io.collection.find_one({"_id": study_id})
-                if not study:
-                    return {}
-                return study
-
-        # If user_id is provided and filter is desired, return studies the user is authorized to view.
-        if user_id and to_filter_by_user_id:
-            user_id = ObjectId(user_id)
-            with DatabaseIO(db_name=self.db_name, collection_name=self.collection_name_users) as db_io:
-                this_user = db_io.collection.find_one({'_id': user_id})
-            if not this_user:
-                return [], "User not found, please try authentication again."
-
-            user_study_ids = this_user.get('study_id', [])
-            user_study_ids = [ObjectId(uid) for uid in user_study_ids]
-            with DatabaseIO(db_name=self.db_name, collection_name=self.collection_name_studies) as db_io:
-                matching_studies = list(db_io.collection.find({'_id': {'$in': user_study_ids}}))
-                return matching_studies
-
-        # If neither is provided, return all studies.
-        with DatabaseIO(db_name=self.db_name, collection_name=self.collection_name_studies) as db_io:
-            return list(db_io.collection.find({}))
-
-    def update_study(self, study_id: str, updated_data: Dict):
-        """
-        Update an existing study by its ID.
-
-        :param study_id: The ID of the study to update.
-        :param updated_data: The new data for the study.
-        """
-
-        study_id = ObjectId(study_id)
-
-        # validate all data is of the right type
-        updated_data['_id'] = study_id
-        try:
-            study_instance = Study(**updated_data)
-        except TypeError as e:
-            # Here, we're assuming the TypeError arises from a mismatch in the fields.
-            # This might not cover all validation cases, but it's a start.
-            raise ValueError(f"Validation error: {str(e)}")
-        except Exception as e:
-            raise Exception(f"Other exception: {str(e)}")
-
-        # Fetch the current study data
-        with DatabaseIO(self.db_name, self.collection_name_studies) as db_io:
-            current_data = db_io.collection.find_one({"_id": study_id})
-
-        if not current_data:
-            raise ValueError(f"No study found with id: {study_id}")
-
-        # Construct a dictionary with only the changed values
-        changes_to_update = {}
-        for key, value in updated_data.items():
-            if current_data.get(key, "") != value:
-                changes_to_update[key] = value
-
-        # If there are changes, update the study in the database
-        if changes_to_update:
-            with DatabaseIO(self.db_name, self.collection_name_studies) as db_io:
-                result = db_io.collection.update_one({"_id": study_id}, {"$set": changes_to_update})
-
-            if result.acknowledged and result.matched_count == 1 and result.modified_count == 1:
-                return_msg = f"Succesfully updated study with id {study_id}"
-            else:
-                return_msg = f"Couldn't confirm updates, please try again or refresh page to see latest"
-        else:
-            return_msg = f"No changes found, study info is up to date"
-
-        return return_msg
-
-
-    def delete_study(self, study_id: str):
-        """
-        Delete a study by its ID.
-
-        :param study_id: The ID of the study to delete.
-        """
-        study_id = ObjectId(study_id)
-        with DatabaseIO(self.db_name, self.collection_name_studies) as db_io:
-            db_io.collection.delete_one({"_id": study_id})