load_test.py

import asyncio
import aiohttp
import json
from tqdm.asyncio import tqdm
import time
from test import test_texts

url = "https://vidhitmakvana1-contact-sharing-recognizer-api.hf.space/detect_contact"
concurrent_requests = 2

async def process_text(session, text, semaphore):
    payload = {"text": text}
    headers = {"Content-Type": "application/json"}

    async with semaphore:
        start_time = time.time()
        while True:
            async with session.post(url, data=json.dumps(payload), headers=headers) as response:
                if response.status == 200:
                    result = await response.json()
                    end_time = time.time()
                    result['response_time'] = end_time - start_time
                    return result
                elif response.status == 429:
                    print(f"Rate limit exceeded. Waiting for 60 seconds before retrying...")
                    await asyncio.sleep(60)
                else:
                    print(f"Error for text: {text}")
                    print(f"Status code: {response.status}")
                    print(f"Response: {await response.text()}")
                    return None

async def main():
    semaphore = asyncio.Semaphore(concurrent_requests)
    async with aiohttp.ClientSession() as session:
        tasks = [process_text(session, text, semaphore) for text in [*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts,*test_texts]]
        results = await tqdm.gather(*tasks)

    correct_predictions = 0
    total_predictions = len(results)
    total_response_time = 0

    for text, result in zip(test_texts, results):
        if result:
            print(f"Text: {result['text']}")
            print(f"Contact Probability: {result['contact_probability']:.4f}")
            print(f"Is Contact Info: {result['is_contact_info']}")
            print(f"Response Time: {result['response_time']:.4f} seconds")
            print("---")
        
            if result['is_contact_info']:
                correct_predictions += 1
        
            total_response_time += result['response_time']

    accuracy = correct_predictions / (total_predictions * 37)
    average_response_time = total_response_time / total_predictions
    print(f"Accuracy: {accuracy:.2f}")
    print(f"Average Response Time: {average_response_time:.4f} seconds")

if __name__ == "__main__":
    while True:
        start_time = time.time()
        asyncio.run(main())
        end_time = time.time()
        total_time = end_time - start_time
        print(f"\nTotal execution time: {total_time:.2f} seconds")