AgentsCourseFinalProject / auxiliary_fns.py
VicBeltran's picture
working agent local version
de814df
import os
import requests
import subprocess
import pandas as pd
from PIL import Image
from io import BytesIO
import soundfile as sf
DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
IMAGE_FILES = ["png", "jpg", "tiff", "jpeg", "bmp"]
AUDIO_FILES = ["wav", "mp3", "aac", "ogg"]
TABULAR_FILES = ["csv", "xlsx"]
def read_audio_file(audio_bytes, file_extension):
"""
Reads audio data from in-memory bytes.
Args:
audio_bytes (bytes): The audio data as bytes.
file_extension (str): The extension of the audio file (e.g., 'wav', 'mp3').
"""
try:
audio_buffer = BytesIO(audio_bytes)
format_string = file_extension.lower()
data, samplerate = sf.read(audio_buffer, format=format_string)
return (data, samplerate)
except sf.LibsndfileError:
print(f"Error: Could not read the audio data from memory with the specified format: {file_extension}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
def read_tabular_data(file_bytes, file_extension):
file_bytes.seek(0)
if file_extension == "csv":
return (pd.read_csv(file_bytes))
elif file_extension == "xlsx":
return (pd.read_excel(file_bytes))
def read_image_data(file_bytes, file_extension):
return Image.open(file_bytes)
def write_and_execute_file(text):
with open(f"file_to_execute.{file_extension}", "wb") as f:
f.write(text)
result = subprocess.run(['python', 'file_to_execute.py'], capture_output=True, text=True, check=True)
return result.stdout
def file_handler(task_id, file_name):
response = requests.get(f"{DEFAULT_API_URL}/files/{task_id}")
response.raise_for_status()
data = response.content
ext = file_name.split(".")[-1]
if ext in AUDIO_FILES:
file_data = read_audio_file(data, ext)
elif ext in TABULAR_FILES:
file_data = read_tabular_file(data, ext)
elif ext in IMAGE_FILES:
file_data = read_image_file(data, ext)
elif ext == "py":
file_data = (data, ext)
return file_data, ext