commit 5da81e2ee6155e9c86a63e0ace0cd866071e8ce0 Author: Alex Blank Date: Sun Dec 29 21:27:11 2024 +0100 initial commit diff --git a/main.py b/main.py new file mode 100644 index 0000000..b77817a --- /dev/null +++ b/main.py @@ -0,0 +1,21 @@ +import os + +import yaml + +CONFIG_FILE = os.environ.get('CONFIG_FILE', "config.yml") + +if CONFIG_FILE is None: + raise ValueError('CONFIG_FILE environment variable is not set') +elif not os.path.exists(CONFIG_FILE): + raise ValueError(f'CONFIG_FILE {CONFIG_FILE} does not exist') + +with open(CONFIG_FILE, 'r') as config_file: + config = yaml.safe_load(config_file) + + +def main(): + pass + + +if __name__ == '__main__': + main() diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e69de29 diff --git a/transcription/__init__.py b/transcription/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/transcription/__pycache__/__init__.cpython-311.pyc b/transcription/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..b0cde6c Binary files /dev/null and b/transcription/__pycache__/__init__.cpython-311.pyc differ diff --git a/transcription/__pycache__/job_utils.cpython-311.pyc b/transcription/__pycache__/job_utils.cpython-311.pyc new file mode 100644 index 0000000..c13862e Binary files /dev/null and b/transcription/__pycache__/job_utils.cpython-311.pyc differ diff --git a/transcription/audio_processing.py b/transcription/audio_processing.py new file mode 100644 index 0000000..0c6b7f7 --- /dev/null +++ b/transcription/audio_processing.py @@ -0,0 +1,72 @@ +import os +import random + +from pydub import AudioSegment +from pydub.silence import split_on_silence + + +def split_audio(audio_file_path: str, + chunk_folder: str, + max_num_chunks: int, + min_silence_level: int, + min_silence_length: int, + keep_silence: int) -> list: + """ + Split the audio file into chunks. + :param audio_file_path: path to the audio file + :param chunk_folder: path to the chunk folder + :param max_num_chunks: number of chunks to split the audio file into + :param min_silence_level: minimum silence level + :param min_silence_length: minimum silence length + :param keep_silence: whether to keep silence + :return: list of chunked audio file paths + """ + + # create the chunk folder + if not os.path.exists(chunk_folder): + os.makedirs(chunk_folder) + + # load audio file + audio = AudioSegment.from_file(audio_file_path) + audio_name = os.path.basename(audio_file_path).split('.')[0] + + # split audio file into chunks + chunks = split_on_silence(audio, + silence_thresh=min_silence_level, + min_silence_len=min_silence_length, + keep_silence=keep_silence) + + if len(chunks) > max_num_chunks: + # randomly combine chunks + while len(chunks) > max_num_chunks: + random_chunk_index = random.randint(0, len(chunks) - 2) + chunks[random_chunk_index] += chunks[random_chunk_index + 1] + chunks.pop(random_chunk_index + 1) + + # save chunks + chunked_audio_files = [] + for i, chunk in enumerate(chunks): + chunk_file_path = os.path.join(chunk_folder, f'{audio_name}_{i}.wav') + chunk.export(chunk_file_path, format='wav') + chunked_audio_files.append(chunk_file_path) + + return chunked_audio_files + + +def convert_to_wav(audio_file_path: str, + output_file_path: str) -> str: + """ + Convert an audio file to wav format. + :param audio_file_path: path to the audio file + :param output_folder_path: path to the output folder + :return: path to the converted audio file + """ + + input_audio = AudioSegment.from_file(audio_file_path) + if ".wav" in output_file_path: + new_audio_file_path = output_file_path + else: + new_audio_file_path = output_file_path + ".wav" + input_audio.export(new_audio_file_path, format='wav') + + return new_audio_file_path diff --git a/transcription/job_execution.py b/transcription/job_execution.py new file mode 100644 index 0000000..e69de29 diff --git a/transcription/job_utils.py b/transcription/job_utils.py new file mode 100644 index 0000000..4398468 --- /dev/null +++ b/transcription/job_utils.py @@ -0,0 +1,202 @@ +import os +import shutil +from multiprocessing import Lock + +# multiprocessing lock for thread safety +lock = Lock() + +chunk_input_subdir = 'input_chunks' +chunk_output_subdir = 'output_chunks' + + +def get_job_data(job_name: str, config: dict) -> dict: + """ + Get job data from the database. + :param job_name: name of the job + :param config: app configuration + :return: job data + """ + + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + + if not os.path.exists(job_dir): + raise ValueError(f'Job {job_name} does not exist') + + job_data = dict() + job_data['name'] = job_name + + output_file = os.path.join(job_dir, 'output.txt') + if os.path.exists(output_file): + with open(output_file, 'r') as f: + job_data['output'] = f.read() + job_data["completed"] = True + else: + job_data['output'] = None + job_data["completed"] = False + + processing_file = os.path.join(job_dir, '_PROCESSING') + if os.path.exists(processing_file): + job_data["processing"] = True + else: + job_data["processing"] = False + + # get progress of task + job_data["progress"] = get_progress(job_name, config) + + return job_data + + +def get_existing_jobs(config: dict) -> list: + """ + Get existing jobs from the database. + :param config: app configuration + :return: list of jobs + """ + + job_root_dir = config['jobs']['root_directory'] + + if not os.path.exists(job_root_dir): + return [] + + jobs = list() + for folder in os.listdir(job_root_dir): + job_data = get_job_data(folder, config) + jobs.append(job_data) + return jobs + + +def is_job(job_name: str, config: dict) -> bool: + """ + Check if a job exists. + :param job_name: name of the job + :param config: app configuration + :return: True if the job exists, False otherwise + """ + + try: + get_job_data(job_name, config) + return True + except ValueError: + return False + + +def create_job(job_name: str, config: dict) -> None: + """ + Create a job in the file system. + :param job_name: name of the job + :param config: app configuration + """ + # check, if job root directory exists + if not os.path.exists(config['jobs']['root_directory']): + os.makedirs(config['jobs']['root_directory']) + + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + + if os.path.exists(job_dir): + raise ValueError(f'Job {job_name} already exists') + + lock.acquire() + + os.mkdir(job_dir) + + lock.release() + + +def rename_job(job_name: str, new_job_name: str, config: dict) -> None: + """ + Rename a job in the file system. + :param job_name: name of the job + :param new_job_name: new name of the job + :param config: app configuration + """ + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + new_job_dir = os.path.join(config['jobs']['root_directory'], new_job_name) + + if not os.path.exists(job_dir): + raise ValueError(f'Job {job_name} does not exist') + + if os.path.exists(new_job_dir): + raise ValueError(f'Job {new_job_name} already exists') + + lock.acquire() + + os.rename(job_dir, new_job_dir) + + lock.release() + + +def remove_job(job_name: str, config: dict) -> None: + """ + Remove a job from the file system. + :param job_name: name of the job + :param config: app configuration + """ + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + + if not os.path.exists(job_dir): + raise ValueError(f'Job {job_name} does not exist') + + lock.acquire() + + # remove the directory + shutil.rmtree(job_dir) + + lock.release() + + +def get_processing(job_name: str, config: dict) -> bool: + """ + Get the processing status of a job. + :param job_name: name of the job + :param config: app configuration + :return: True if the job is processing, False otherwise + """ + + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + processing_file = os.path.join(job_dir, '_PROCESSING') + + return os.path.exists(processing_file) + + +def set_processing(job_name: str, config: dict, processing: bool) -> None: + """ + Set the processing status of a job. + :param job_name: name of the job + :param config: app configuration + :param processing: processing status + """ + + lock.acquire() + + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + processing_file = os.path.join(job_dir, '_PROCESSING') + + if processing: + open(processing_file, 'w').close() + else: + os.remove(processing_file) + + lock.release() + + +def get_progress(job_name: str, config: dict) -> int: + """ + Get the progress of a job. + :param job_name: name of the job + :param config: app configuration + :return: progress of the job + """ + + input_dir = os.path.join(config['jobs']['root_directory'], job_name, chunk_input_subdir) + output_dir = os.path.join(config['jobs']['root_directory'], job_name, chunk_output_subdir) + + num_input_files = len(os.listdir(input_dir)) + num_output_files = len(os.listdir(output_dir)) + + if num_input_files == 0 or num_output_files == 0: + return 0 + + if num_output_files >= num_input_files: + return 100 + + return int(num_output_files / num_input_files * 100) diff --git a/transcription/transcription.py b/transcription/transcription.py new file mode 100644 index 0000000..29206ea --- /dev/null +++ b/transcription/transcription.py @@ -0,0 +1,57 @@ +import whisper + + +# clojure for model +def model_loader() -> callable: + """ + Clojure for loading the model + :return: function that loads the model + """ + + model = None + + def load_model(config: dict): + """ + Load the model + :return: model + """ + nonlocal model + + if model is not None: + return model + else: + model_type = config['model']['type'] + model = whisper.load_model(model_type) + return model + + return load_model + + +load_model = model_loader() + + +def transcribe_audio(audio_file_path: str, + output_file_path: str, + config: dict) -> str: + """ + Transcribe an audio file with openai whisper + :param audio_file_path: filepath of the audio file + :param output_file_path: filepath of the output file + :param config: app config + :return: transcribed text + """ + + # load the model + model = load_model(config) + + # move model to desired device + desired_device = config['model']['device'] + model.to(desired_device) + + result = model.transcribe(audio_file_path) + result_text = result['text'] + + # write the result to a text file + with open(output_file_path, 'w') as result_file: + result_file.write(result["text"]) + return result_text diff --git a/transcription/transcription_helper.py b/transcription/transcription_helper.py new file mode 100644 index 0000000..a8ce812 --- /dev/null +++ b/transcription/transcription_helper.py @@ -0,0 +1,155 @@ +import concurrent.futures +import os +from multiprocessing import Lock +from concurrent.futures import ProcessPoolExecutor + +import job_utils +from utils import get_logger +from audio_processing import split_audio, convert_to_wav +from job_utils import is_job, remove_job, create_job, get_processing, set_processing +from transcription import transcribe_audio + +job_update_lock = Lock() +default_input_audio_file_name = 'input_audio' + +logger = get_logger(__name__) + +# define a global executor for non-blocking transcription jobs +executor = ProcessPoolExecutor() + + +def transcribe(audio_file_path: str, + config: dict, + blocking: bool = True, + overwrite: bool = False, + executor: concurrent.futures.Executor = executor) -> str: + """ + Transcribe an audio file. + :param audio_file_path: path to the audio file + :param config: app configuration + :param blocking: block until the job is finished + :param overwrite: overwrite existing job + :param executor: executor for non-blocking transcription + :return: name of the created job + """ + + job_name = os.path.basename(audio_file_path).split('.')[0] + job_dir = os.path.join(config['jobs']['root_directory'], job_name) + + logger.info(f"Transcribing {audio_file_path} to {job_dir}") + + # check, if a job with the same audio file already exists + job_already_exists = is_job(job_name, config) + + if job_already_exists and not overwrite: + logger.error(f'Job for {audio_file_path} already exists and overwrite is not set') + raise ValueError(f'Job for {audio_file_path} already exists') + + # remove, if overwrite is set + if job_already_exists and overwrite: + logger.info(f'Removing existing job {job_name} since overwrite is set') + remove_job(job_name, config) + + # create a new base job + logger.info(f'Creating job {job_name}') + create_job(job_name, config) + + # copy the audio file to the job directory + logger.info(f'Copying audio file {audio_file_path} to {job_dir}') + input_audio_file_name = os.path.basename(audio_file_path).split('.')[0] + input_audio_file_type = os.path.splitext(audio_file_path)[1] + audio_file_destination = os.path.join(job_dir, f"{input_audio_file_name}{input_audio_file_type}") + os.system(f'cp {audio_file_path} {audio_file_destination}') + + # convert the audio file to wav, if it is not already + if not audio_file_destination.endswith('.wav'): + logger.info(f'Converting audio file {audio_file_destination} to wav') + convert_to_wav(audio_file_path, + os.path.join(job_dir, f"{default_input_audio_file_name}.wav")) + + # run the transcription job + logger.info(f'Running transcription job {job_name}') + + if blocking: + logger.info(f'Starting blocking transcription job {job_name}') + run_transcription_job(job_name, config) + else: + logger.info(f'Starting async transcription job {job_name}') + executor.submit(run_transcription_job, job_name, config, 100) + logger.info(f'Started async transcription job {job_name}') + + return job_name + + +def run_transcription_job(job_name: str, + config: dict, + num_chunks=100) -> None: + """ + Start a transcription job. + :param job_name: id of the job + :param config: app configuration + :param num_chunks: number of chunks to split the audio file into + """ + + # check if the job is already running + if get_processing(job_name, config): + raise ValueError(f'Job {job_name} is already running') + + # set the job to processing + set_processing(job_name, config, processing=True) + + try: + # split the audio file into chunks + logger.info(f'Splitting audio file into chunks for job {job_name}') + audio_file_path = os.path.join(config['jobs']['root_directory'], job_name, + f"{default_input_audio_file_name}.wav") + audio_chunk_folder = os.path.join(config['jobs']['root_directory'], job_name, job_utils.chunk_input_subdir) + audio_chunks = split_audio(audio_file_path, + audio_chunk_folder, + num_chunks, + config["audio"]["chunking"]["min_silence_level"], + config["audio"]["chunking"]["min_silence_length"], + config["audio"]["chunking"]["ms_silence_to_keep"]) + + # transcribe the audio chunks + logger.info(f'Transcribing {len(audio_chunks)} audio chunks for job {job_name}') + # create chunk output folder, if not done yet + chunk_output_folder = os.path.join(config['jobs']['root_directory'], job_name, job_utils.chunk_output_subdir) + if not os.path.exists(chunk_output_folder): + os.makedirs(chunk_output_folder) + for i, audio_chunk in enumerate(audio_chunks): + # transcribe the audio + logger.info(f'Transcribing audio chunk {i + 1} of {len(audio_chunks)} for job {job_name}') + output_file_path = os.path.join(chunk_output_folder, f'output_{i}.txt') + transcribe_audio(audio_chunk, + output_file_path, + config) + + # combine the output text chunks + logger.info(f'Combining output text chunks for job {job_name}') + output_chunk_folder = os.path.join(config['jobs']['root_directory'], job_name, job_utils.chunk_output_subdir) + output_text = '' + for text_chunk_file in sorted(os.listdir(output_chunk_folder)): + with open(os.path.join(output_chunk_folder, text_chunk_file), 'r') as text_chunk: + output_text += text_chunk.read() + ' ' + + # remove trailing space + output_text = output_text.strip() + + # write the output to a file + logger.info(f'Writing output text to file for job {job_name}') + output_file = os.path.join(config['jobs']['root_directory'], job_name, 'output.txt') + with open(output_file, 'w') as f: + f.write(output_text) + + # set the job to not processing + set_processing(job_name, config, processing=False) + + logger.info(f'Job {job_name} completed') + + + + except Exception as e: + # set the job to not processing + set_processing(job_name, config, processing=False) + raise e diff --git a/transcription/utils.py b/transcription/utils.py new file mode 100644 index 0000000..13a1dab --- /dev/null +++ b/transcription/utils.py @@ -0,0 +1,32 @@ +import sys +import logging +from logging import Logger + + +def get_logger(name: str) -> Logger: + """ + Get a logger instance. + :param name: name of the logger + :return: logger instance + """ + logger = logging.getLogger(name) + logger.setLevel(logging.INFO) + + # create a file handler + file_handler = logging.FileHandler(f'{name}.log') + file_handler.setLevel(logging.INFO) + + # create stdout handler + stdout_handler = logging.StreamHandler(sys.stdout) + stdout_handler.setLevel(logging.INFO) + + # create a logging format + formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') + file_handler.setFormatter(formatter) + stdout_handler.setFormatter(formatter) + + # add the handlers to the logger + logger.addHandler(file_handler) + logger.addHandler(stdout_handler) + + return logger diff --git a/webapp/__init__.py b/webapp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/webapp/__pycache__/__init__.cpython-311.pyc b/webapp/__pycache__/__init__.cpython-311.pyc new file mode 100644 index 0000000..f0da1a1 Binary files /dev/null and b/webapp/__pycache__/__init__.cpython-311.pyc differ diff --git a/webapp/__pycache__/home.cpython-311.pyc b/webapp/__pycache__/home.cpython-311.pyc new file mode 100644 index 0000000..669bd27 Binary files /dev/null and b/webapp/__pycache__/home.cpython-311.pyc differ diff --git a/webapp/__pycache__/task_list.cpython-311.pyc b/webapp/__pycache__/task_list.cpython-311.pyc new file mode 100644 index 0000000..f58c0df Binary files /dev/null and b/webapp/__pycache__/task_list.cpython-311.pyc differ diff --git a/webapp/home.py b/webapp/home.py new file mode 100644 index 0000000..4c242ac --- /dev/null +++ b/webapp/home.py @@ -0,0 +1,22 @@ +import streamlit as st + +from webapp.task_list import render_task_list + + +def render_home(container, + app_config: dict, + authenticator): + """ + Render the home page + :param container: container object to place the content in + :param app_config: app configuration + :return: None + """ + + with container.container(): + st.title('Annescribe Home') + st.write("Welcome to AnneScribe!") + authenticator.logout() + + # render task list + render_task_list(container, app_config) diff --git a/webapp/login.py b/webapp/login.py new file mode 100644 index 0000000..88e2e50 --- /dev/null +++ b/webapp/login.py @@ -0,0 +1,9 @@ +import os +import yaml + +import streamlit as st +import streamlit_authenticator as st_auth + + + + diff --git a/webapp/task_list.py b/webapp/task_list.py new file mode 100644 index 0000000..60b6ad6 --- /dev/null +++ b/webapp/task_list.py @@ -0,0 +1,97 @@ +from functools import partial + +import streamlit as st +from streamlit_autorefresh import st_autorefresh + + +from transcription.job_utils import get_existing_jobs + +def show_results(job_data: dict): + """ + Show the results of a job + :param job_data: job data + :return: None + """ + if job_data is None: + return + else: + st.session_state[f"results_button_{job_data['name']}"] = True + st.write(f"Transcription Results for job {job_data['name']}") + st.write(f"Output: {job_data['output']}") + +def hide_results(job_data: dict): + """ + Hide the results of a job + :param job_data: job data + :return: None + """ + st.session_state[f"results_button_{job_data['name']}"] = False + +def results_button(job_data: dict, app_config: dict): + """ + Show the results of a job + :param job_data: job data + :param app_config: app configuration + :return: None + """ + is_toggled = st.session_state.get(f"results_button_{job_data['name']}") + other_toggled = any([st.session_state.get(f"results_button_{job['name']}") for job in get_existing_jobs(app_config)]) + + if other_toggled: + for job in get_existing_jobs(app_config): + st.session_state[f"results_button_{job['name']}"] = False + if is_toggled: + return st.button("Show Results", on_click=partial(show_results, job_data)) + else: + return st.button("Hide Results", on_click=partial(hide_results, job_data)) + + +def render_task_list(container, app_config: dict): + """ + Render the task list page + :param container: container object to place the content in + :param app_config: app configuration + :return: None + """ + + # disable all results + for job in get_existing_jobs(app_config): + st.session_state[f"results_button_{job['name']}"] = False + + with container.container(): + st.subheader('Task List') + + existing_jobs = get_existing_jobs(app_config) + + column_config = [ + { + "name": "Name", + "render_func": lambda x: st.markdown(f"**{x['name']}**") + }, + { + "name": "Progress", + "render_func": lambda x: st.progress(x["progress"]) if x["progress"] < 100 else st.markdown(" -- ") + }, + { + "name": "Status", + "render_func": lambda x: st.markdown("Completed") if x["completed"] else st.markdown("Processing" if x["processing"] else "Pending") + }, + { + "name": "Output", + "render_func": lambda x: results_button(x, app_config) + } + ] + + if len(existing_jobs) == 0: + st.write("No tasks available.") + else: + for i, job in enumerate(existing_jobs): + cols = st.columns(len(column_config)) + for j, col in enumerate(column_config): + cols[j].write(col["name"]) + with cols[j]: + col["render_func"](job) + + + + diff --git a/webapp_main.py b/webapp_main.py new file mode 100644 index 0000000..c668074 --- /dev/null +++ b/webapp_main.py @@ -0,0 +1,66 @@ +import os +import tempfile +import json +import yaml +import subprocess + +import streamlit as st +import streamlit_authenticator as st_auth +from streamlit_autorefresh import st_autorefresh + +from webapp.home import render_home + +CONFIG_FILE = os.environ.get('CONFIG_FILE', "config.yml") + +if CONFIG_FILE is None: + raise ValueError('CONFIG_FILE environment variable is not set') +elif not os.path.exists(CONFIG_FILE): + raise ValueError(f'CONFIG_FILE {CONFIG_FILE} does not exist') + +with open(CONFIG_FILE, 'r') as config_file: + config = yaml.safe_load(config_file) + +AUTH_FILE = os.environ.get('AUTH_FILE', "auth.yml") +if AUTH_FILE is None: + raise ValueError('AUTH_YML_FILE environment variable is not set') +elif not os.path.exists(AUTH_FILE): + raise ValueError(f'AUTH_YML_FILE {AUTH_FILE} does not exist') + +with open(AUTH_FILE, 'r') as auth_file: + auth_config = yaml.safe_load(auth_file) + +authenticator = st_auth.Authenticate( + auth_config["credentials"], + auth_config["cookie"]["name"], + auth_config["cookie"]["key"], + auth_config["cookie"]["expiry_days"], +) + +content_wrapper = st.empty() + + +def render_login(): + with content_wrapper.container(): + st.title('Annescribe Login') + st.write("Please login with your credentials.") + authenticator.login(callback=st.session_state.clear()) + + +if "authentication_status" not in st.session_state.keys(): + authentication_status = None + st.session_state["authentication_status"] = authentication_status +else: + authentication_status = st.session_state.get("authentication_status") + username = st.session_state.get("username") +print(authentication_status) +if authentication_status is True: + render_home(content_wrapper, config, authenticator) +elif authentication_status is None: + st.warning('Please enter your username and password') + render_login() +else: + # make sure to remove cookie + render_login() + + +# st_autorefresh(interval=2000, ) \ No newline at end of file