AI toolkit for professional and amateur oral historians
# Configuration file for the Sphinx documentation builder.
#
# This file only contains a selection of the most common options. For a full
# list see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html

# -- Path setup --------------------------------------------------------------

# If extensions (or modules to document with autodoc) are in another directory,
# add these directories to sys.path here. If the directory is relative to the
# documentation root, use os.path.abspath to make it absolute, like shown here.
#
# import os
# import sys
# sys.path.insert(0, os.path.abspath('.'))


# -- Project information -----------------------------------------------------

project = 'HistoryAIToolkit'
copyright = '2023, Audrey Roy Greenfeld and contributors'
author = 'Audrey Roy Greenfeld and contributors'


# -- General configuration ---------------------------------------------------

# Add any Sphinx extension module names here, as strings. They can be
# extensions coming with Sphinx (named 'sphinx.ext.*') or your custom
# ones.
extensions = [
]

# Add any paths that contain templates here, relative to this directory.
templates_path = ['_templates']

# List of patterns, relative to source directory, that match files and
# directories to ignore when looking for source files.
# This pattern also affects html_static_path and html_extra_path.
exclude_patterns = ['_build', 'Thumbs.db', '.DS_Store']


# -- Options for HTML output -------------------------------------------------

# The theme to use for HTML and HTML Help pages.  See the documentation for
# a list of builtin themes.
#
html_theme = 'alabaster'

# Add any paths that contain custom static files (such as style sheets) here,
# relative to this directory. They are copied after the builtin static files,
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ['_static']


# Myst configuration   
extensions = ["myst_parser"]
myst_enable_extensions = ["colon_fence"]
docs/conf.py
historysciencelab/HistoryAIToolkit
import sys
import typer

from pathlib import Path
from typing_extensions import Annotated

from slicer import audio_slicing
from transcript import transcribe_from_paths


__version__ = '0.0.1'

app = typer.Typer()

def version_callback(value: bool):
    if value:
        typer.echo(f"HistoryAIToolKit: {__version__}")
        raise typer.Exit()

@app.command("version")    
def main():
    """Checks the package version"""
    version_callback(sys.argv[1:])

@app.command()
def slice(
    source: Annotated[
        Path,
        typer.Argument(
            exists=True,
            file_okay=True,
            dir_okay=False,
            readable=True,
            resolve_path=True,
            help="Source audio file",
        ),
    ],
    start: Annotated[str, typer.Argument(help="start time in mins:secs")],
    duration: Annotated[str, typer.Argument(help="duration in mins:secs")],
):
    """Slices an audio file into smaller audio files."""
    audio_slicing(source, start, duration)


@app.command()
def transcribe(
    source: Annotated[
        Path,
        typer.Argument(
            exists=True,
            file_okay=True,
            dir_okay=False,
            readable=True,
            resolve_path=True,
            help="Source audio file",
        ),
    ],
    target: Path,
):
    """Transcribes an audio file into text."""
    if source.suffix not in [".mp3", ".wav"]:
        raise ValueError("File must be an .mp3 or .wav file")
    transcribe_from_paths(source, target)


if __name__ == "__main__":
    app()

src/interviewkit/cli.py
historysciencelab/HistoryAIToolkit
from random import choice
from interviewee import Interviewee
from enum import StrEnum, auto
from transcript import Transcript


class Status(StrEnum):
    IN_PROGRESS = auto()
    RECORDED = auto()
    RECORDING = auto()
    TRANSCRIBED = auto()


class Interview(object):
    """An interview between an Interviewee and an Interviewer."""

    def __init__(self, interviewee: Interviewee, transcript: Transcript):
        self.interviewee = interviewee
        self.status = Status.IN_PROGRESS

    def start_recording(self) -> None:
        """Start recording audio."""
        self.status = Status.RECORDING

    def stop_recording(self) -> None:
        """Stop recording audio."""
        self.status = Status.RECORDED

    def start_transcription(self) -> None:
        print("transcribing")
        self.status = "Transcribing"

    def add_to_transcript(self, speech: str) -> None:
        self.transcript.append(speech)

    def stop_transcription(self) -> None:
        self.status = Status.TRANSCRIBED

        # Combine list of transcribed text into a single string
        self.transcript = " ".join(self.transcript)

    def suggest_questions(self) -> str:
        """suggest questions based on the transcript content."""
        suggested_questions = self.generate_questions(self.transcript)
        return suggested_questions

    def generate_questions(self, content: str) -> list[str]:
        """generate questions, and return them."""

        #TODO
        response = None

        # Extract and return the generated questions
        generated_questions = [choice["text"].strip() for choice in response.choices]
        return generated_questions

src/interviewkit/interview.py
historysciencelab/HistoryAIToolkit
from enum import StrEnum, auto
from pydantic import BaseModel, NonNegativeInt

class Gender(StrEnum):
    MALE = auto()
    FEMALE = auto()
    OTHER = auto()


class Interviewee(BaseModel):
    """An Interviewee is a person who is interviewed."""

    name: str
    age: NonNegativeInt
    gender: Gender
src/interviewkit/interviewee.py
historysciencelab/HistoryAIToolkit
import os

from interview import Interview
from interviewee import Interviewee, Gender
from transcript import Transcript
from settings import Settings

settings = Settings()

if __name__ == "__main__":
    interviewee = Interviewee(name="John Doe", age=60, gender=Gender.MALE)

    transcript_content = "This is the content of the transcript."
    transcript = Transcript(content=transcript_content)

    interview = Interview(interviewee, transcript)

    interview.start_transcription()

    print(f"Interview Status: {interview.status}")

src/interviewkit/main.py
historysciencelab/HistoryAIToolkit
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    #TODO
    pass
src/interviewkit/settings.py
historysciencelab/HistoryAIToolkit
"""
This takes three args, a filepath to an audio file, audio start and audio end times.

It will sample the file specified for that many minutes and seconds. 

The audio start and end times params currently support time in 2 ways either minutes and seconds or minutes only.

`ffmpeg` is a requirement for this to work.

Example usage:
    Format:  python interviewkit/slicer.py path_to_audio_file audio_slice_start_time audio_slice_end_time
    Example 1: python interviewkit/slicer.py data/MartineBarrat_FINAL.mp3 2 3 
             (Audio will be sliced from 2mins to 4mins)
    Example 2: python interviewkit/slicer.py data/Martine+Barrat_FINAL.mp3 70:40 80:40 
             (Audio will be sliced from 105mins 40secs to 107mins and 40secs)

This generates:
    data/sampled-70m40s-80m40s-MartineBarrat_FINAL.mp3
    data/sampled-2m-4m-MartineBarrat_FINAL.mp3

Note: 
The Script currently doesn't support input in hours format, so hours will need to converted to mins:secs format.
Example: If we want audio from 1hr 20mins 30 secs to 1hr 30mins 40secs, it can be done as shown below:
1hr 20mins 30 secs = 80mins 30secs
1hr 30mins 40sec = 90mins 40secs

python interviewkit/slicer.py data/Martine+Barrat_FINAL.mp3 80:30 90:40

"""
import sys
from pathlib import Path
import shutil

try:
    import pydub
except ImportError:
    print("Please install pydub: pip install pydub")
    exit(1)

if shutil.which("ffmpeg") is None:
    print("Please install ffmpeg: https://ffmpeg.org/download.html")
    print("  On mac you can: brew install ffmpeg")
    exit(1)

def convert_audio_time_to_msec(audio_time_split_list):
    """ Converting mins and secs to msecs for pydub computation """

    if(audio_time_split_list):
        if(len(audio_time_split_list) == 1):
            return int(audio_time_split_list[0]) * 60 * 1000
        elif(len(audio_time_split_list) == 2):
            return int(audio_time_split_list[0]) * 60 * 1000 + int(audio_time_split_list[1]) * 1000
        else:
            print("Error! Audio slice input params invalid. Audio slice supports start/end time in mins or mins:secs format. Please try again with correct input times.")
            print("Error inside convert_audio_time_to_msec(audio_time_split_list) funtion.")
            exit(1)
    else:
        print("Error! Audio slice input params invalid. Please try again with correct parameters.")
        print("Error inside convert_audio_time_to_msec(audio_time_split_list) funtion.")
        exit(1)

def export_filename(audio_time_list):
    """ Filename for exported file """
    
    if audio_time_list and len(audio_time_list) == 2:
        return f"{audio_time_list[0]}m{audio_time_list[1]}s"
    elif audio_time_list and len(audio_time_list) == 1:
        return f"{audio_time_list[0]}m"
    else:
        print("Error! Audio slice input params invalid. Please try again with correct parameters.")
        print("Error inside export_filename(audio_time_list) funtion.")
        exit(1)


def audio_slicing(path, audio_slice_start_time, audio_slice_end_time):
    """ It reads the original audio and uses start and end input time params to generate sliced audio. """
    
    print("Sampling {} from {} to {}".format(path, audio_slice_start_time, audio_slice_end_time))

    # Reading original audio file
    audio = pydub.AudioSegment.from_file(path)
    original_audio_size_ms = audio.duration_seconds * 1000

    # Fetching mins and secs from audio input
    audio_start_time_list = audio_slice_start_time.split(":")
    audio_end_time_list = audio_slice_end_time.split(":")

    # Converting audio start and end times in msecs
    audio_start_time = convert_audio_time_to_msec(audio_start_time_list)
    audio_end_time = convert_audio_time_to_msec(audio_end_time_list)
    
    # Check if audio start and end times are within original audio size limits
    if(audio_start_time > original_audio_size_ms or audio_end_time > original_audio_size_ms):
        print("Error! Audio slice input params cannot be greater than original audio size. Please try again with correct parameters.")
        exit(1)

    # Audio slicing process
    audio = audio[audio_start_time:audio_end_time]

    # Filename for exported file
    audio_start_time_name = export_filename(audio_start_time_list)
    audio_end_time_name = export_filename(audio_end_time_list)
    new_filename =  f"{path.parent}/sampled-{audio_start_time_name}-{audio_end_time_name}-{path.name}"
    audio.export(new_filename, format="mp3")
    print("Created new file: ", new_filename)

def main():
    if len(sys.argv) != 4:
        print("Usage: python3 slicer.py <filepath> <audio start time in minutes> <audio end time in minutes>")
        return

    path = Path(sys.argv[1])
    audio_slice_start_time = (sys.argv[2])
    audio_slice_end_time = (sys.argv[3])

    audio_slicing(path, audio_slice_start_time, audio_slice_end_time)

if __name__ == '__main__':
    main()

src/interviewkit/slicer.py
historysciencelab/HistoryAIToolkit
from pathlib import Path
from rich.console import Console
import sys
from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

try:
    import whisper
except ImportError:
    print("Please install Whisper: pip install openai-whisper")
    exit(1)

from whisper.utils import get_writer
from pydantic import BaseModel

console = Console()

class Transcript(BaseModel):
    """The Transcript entity represents the transcript of an interview."""
    content: str

# # Securely get your credentials
# PAT = os.getenv('CLARIFAI_PAT')
# USER_ID = os.getenv('CLARIFAI_USER_ID')
# APP_ID = os.getenv('CLARIFAI_APP_ID')
# MODEL_ID = os.getenv('CLARIFAI_MODEL_ID')
# MODEL_VERSION_ID = os.getenv('CLARIFAI_MODEL_VERSION_ID')
PAT = ''
# Specify the correct user_id/app_id pairings
# Since you're making inferences outside your app's scope
USER_ID = 'meta'
APP_ID = 'Llama-2'
# Change these to whatever model and text URL you want to use
MODEL_ID = 'llama2-70b-chat'
MODEL_VERSION_ID = 'acba9c1995f8462390d7cb77d482810b'

def generate_questions(transcript_chunk):
    channel = ClarifaiChannel.get_grpc_channel()
    stub = service_pb2_grpc.V2Stub(channel)

    metadata = (('authorization', 'Key ' + PAT),)
    userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

    post_model_outputs_response = stub.PostModelOutputs(
        service_pb2.PostModelOutputsRequest(
            user_app_id=userDataObject,
            model_id=MODEL_ID,
            version_id=MODEL_VERSION_ID,
            inputs=[
                resources_pb2.Input(
                    data=resources_pb2.Data(
                        text=resources_pb2.Text(
                            raw=transcript_chunk
                        )
                    )
                )
            ]
        ),
        metadata=metadata
    )

    if post_model_outputs_response.status.code != status_code_pb2.SUCCESS:
        print(post_model_outputs_response.status)
        raise Exception(f"Post model outputs failed, status: {post_model_outputs_response.status.description}")

    output = post_model_outputs_response.outputs[0]
    return output.data.text.raw

def transcribe_from_paths(source: Path, target: Path) -> None:
    console.print("Loading whisper base model...")
    model = whisper.load_model("base")

    with console.status("Transcribing audio file..."):
        result = model.transcribe((str(source)), fp16=False)

    # Setting some initial options values for the .txt output file
    txt_file_options = {
        "max_line_width": 50,  
        "max_line_count": 1,
        "highlight_words": False,  
    }

    # Save as a .txt file with hard breaks, added for readability to user
    console.print("Saving transcript as a .txt file...")
    txt_writer = get_writer("txt", str(target))
    txt_writer(result, source.name, txt_file_options)

    console.print("Transcript saved to:")
    console.print(f"    [green bold]{target / source.name}.txt[/green bold]")

    # Generate questions from the transcript
    transcript_chunk = result['text']  # Assuming 'result' contains the transcribed text
    # Debug: Print type and value of transcript_chunk
    print(f"Type of transcript_chunk: {type(transcript_chunk)}")
    print(f"Value of transcript_chunk: {transcript_chunk}")
    
    # Ensure transcript_chunk is a string
    if not isinstance(transcript_chunk, str):
        print("Warning: transcript_chunk is not a string. Trying to convert...")
        transcript_chunk = str(transcript_chunk)
    
    questions = generate_questions(transcript_chunk)
    console.print("Generated Questions:\n", questions)

if __name__ == "__main__":
    source = Path(sys.argv[1])
    target = Path(sys.argv[2])
    if source.suffix not in [".mp3", ".wav"]:
        raise ValueError("File must be an .mp3 or .wav file")
    transcribe_from_paths(source, target)

src/interviewkit/transcript.py
historysciencelab/HistoryAIToolkit
from pathlib import Path
from rich.console import Console
import sys
from transformers import T5Tokenizer, T5ForConditionalGeneration

try:
    import whisper
except ImportError:
    print("Please install Whisper: pip install openai-whisper")
    exit(1)

from whisper.utils import get_writer
from pydantic import BaseModel

console = Console()

class Transcript(BaseModel):
    """The Transcript entity represents the transcript of an interview."""
    content: str

# Load T5 model and tokenizer
tokenizer = T5Tokenizer.from_pretrained("t5-base")
model = T5ForConditionalGeneration.from_pretrained("t5-base")
def chunk_text(text, max_length):
    """Split the text into chunks of max_length."""
    words = text.split()
    chunks = []
    current_chunk = []
    current_length = 0
    
    for word in words:
        if current_length + len(word) <= max_length:
            current_length += len(word) + 1  # +1 for the space
            current_chunk.append(word)
        else:
            chunks.append(" ".join(current_chunk))
            current_chunk = [word]
            current_length = len(word)
    
    chunks.append(" ".join(current_chunk))  # Don't forget the last chunk
    return chunks

def generate_questions_for_all_chunks(chunks):
    all_questions = []
    for chunk in chunks:
        questions = generate_questions(chunk)
        all_questions.append(questions)
    return all_questions

def generate_questions(transcript_chunk):
    # Ensure transcript_chunk is a string
    if not isinstance(transcript_chunk, str):
        print("Warning: transcript_chunk is not a string. Trying to convert...")
        transcript_chunk = str(transcript_chunk)

    input_text = "Generate questions based on the Interview texts: " + transcript_chunk
    input_ids = tokenizer(input_text, return_tensors="pt").input_ids

    outputs = model.generate(input_ids)
    questions = tokenizer.decode(outputs[0])
    
    return questions

def transcribe_from_paths(source: Path, target: Path) -> None:
    console.print("Loading whisper base model...")
    model = whisper.load_model("base")

    with console.status("Transcribing audio file..."):
        result = model.transcribe((str(source)), fp16=False)

    txt_file_options = {
        "max_line_width": 50,  
        "max_line_count": 1,
        "highlight_words": False,  
    }

    console.print("Saving transcript as a .txt file...")
    txt_writer = get_writer("txt", str(target))
    txt_writer(result, source.name, txt_file_options)

    console.print("Transcript saved to:")
    console.print(f"    [green bold]{target / source.name}.txt[/green bold]")

    transcript_chunk = result['text']
    max_length = 512
    chunks = chunk_text(transcript_chunk, max_length)
    
    all_questions = generate_questions_for_all_chunks(chunks)
    
    for i, questions in enumerate(all_questions):
        console.print(f"Generated Questions for chunk {i+1}:\n", questions)

    
    # chunks = chunk_text(transcript_chunk, max_length)
    # questions = generate_questions(chunks[0])
    # console.print("Generated Questions:\n", questions)

if __name__ == "__main__":
    source = Path(sys.argv[1])
    target = Path(sys.argv[2])
    if source.suffix not in [".mp3", ".wav"]:
        raise ValueError("File must be an .mp3 or .wav file")
    transcribe_from_paths(source, target)

src/interviewkit/transcript_using_m5.py
historysciencelab/HistoryAIToolkit
import pytest
import subprocess


@pytest.fixture
def run_cli_command():
    # This fixture runs the CLI command and captures its output
    def run_command():
        # Run the CLI command and capture its output
        result = subprocess.run(["python", "-m", "src.interviewkit.cli", "version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)
        return result
    
    return run_command


class TestCLIVersion():
    
    def test_version_option(self, run_cli_command):
        result = run_cli_command()

        assert result.returncode == 0

        # Check if the version number is correct
        assert f"HistoryAIToolKit: 0.0.1" in result.stdout
tests/test_cli_version.py
historysciencelab/HistoryAIToolkit
import subprocess
import sys
import pytest
import shutil

@pytest.fixture
def create_venv(tmp_path):
    venv_dir = tmp_path / ".venv"
    subprocess.run([sys.executable, "-m", "venv", str(venv_dir)], check=True)
    yield venv_dir  
    shutil.rmtree(venv_dir)  


def test_install_in_editable_mode(create_venv):
    activate_script = create_venv / "bin" / "activate" if sys.platform != "win32" else create_venv / "Scripts" / "activate"
    shell = False if sys.platform != "win32" else True
    
   
    commands = [
        f"source {activate_script}",
        "pip install -e '.[test]'"
    ]
    
    
    process = subprocess.Popen(["/bin/bash", "-c", " && ".join(commands)], shell=shell)
    process.communicate()
    
    
    assert process.returncode == 0, "Installation in editable mode failed"


tests/test_install_package.py
historysciencelab/HistoryAIToolkit