|
""" |
|
This module provides the DesignProcess class to interact with the homedesigns.ai API |
|
for starting and checking the status of a design process, as well as generating design ideas |
|
using OpenAI's API. |
|
""" |
|
import os |
|
import openai |
|
|
|
from dotenv import load_dotenv |
|
import time |
|
load_dotenv() |
|
import requests |
|
from PIL import Image |
|
|
|
class DesignProcess: |
|
""" |
|
A class to handle the design process using the homedesigns.ai API. |
|
""" |
|
|
|
def __init__(self, image_path, design_type="Interior", ai_intervention="Mid", no_design=1, design_style="Scandinavian", room_type="Bedroom", house_angle=None, garden_type=None, internal_prompt=None): |
|
self.url = os.getenv("S2R_API_URL") |
|
self.headers = { |
|
"Authorization": os.getenv("AUTH_TOKEN"), |
|
} |
|
self.image_path = image_path |
|
self.design_type = design_type |
|
self.ai_intervention = ai_intervention |
|
self.no_design = no_design |
|
self.design_style = design_style |
|
self.room_type = room_type |
|
self.house_angle = house_angle |
|
self.garden_type = garden_type |
|
self.internal_prompt = internal_prompt |
|
|
|
def start_process(self, max_retries=3, delay=2): |
|
""" |
|
Starts the design process by sending a POST request. |
|
Retries the request if it fails. |
|
""" |
|
attempt = 0 |
|
while attempt < max_retries: |
|
try: |
|
with open(self.image_path, "rb") as image_file: |
|
image = Image.open(image_file) |
|
width, height = image.size |
|
if width < 512 or height < 512: |
|
raise ValueError("The image must have a minimum width and height of 512 pixels.") |
|
image_file.seek(0) |
|
|
|
files = {"image": open(self.image_path, "rb")} |
|
data = { |
|
"design_type": self.design_type, |
|
"ai_intervention": self.ai_intervention, |
|
"no_design": self.no_design, |
|
"design_style": self.design_style, |
|
"room_type": self.room_type, |
|
"house_angle": self.house_angle, |
|
"garden_type": self.garden_type, |
|
"internal_prompt": self.internal_prompt |
|
} |
|
print("====================Data before filtering=======================") |
|
print(data) |
|
data = {k: v for k, v in data.items() if v is not None} |
|
print("====================Data after filtering========================") |
|
print(data) |
|
print("================================================================") |
|
response = requests.post(self.url, headers=self.headers, files=files, data=data) |
|
print(response) |
|
print("===========================Text============================") |
|
print(response.text) |
|
print("================================================================") |
|
response.raise_for_status() |
|
|
|
try: |
|
return response.json() |
|
except requests.exceptions.JSONDecodeError: |
|
print("Failed to decode JSON response") |
|
print("Response content:", response.text) |
|
raise requests.RequestException("Invalid JSON response") |
|
except requests.RequestException as e: |
|
print(f"Attempt {attempt + 1} failed: {e}") |
|
attempt += 1 |
|
if attempt < max_retries: |
|
time.sleep(delay) |
|
else: |
|
raise |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def check_status(self, request_id): |
|
""" |
|
Checks the status of the design process. |
|
""" |
|
status_url = f"https://homedesigns.ai/api/v2/perfect_redesign/status_check/{request_id}" |
|
attempt = 0 |
|
max_retries = 3 |
|
delay = 5 |
|
|
|
while attempt < max_retries: |
|
try: |
|
status_response = requests.get(status_url, headers=self.headers) |
|
status_response.raise_for_status() |
|
|
|
try: |
|
return status_response.json() |
|
except requests.exceptions.JSONDecodeError: |
|
print("Failed to decode JSON response") |
|
print("Response content:", status_response.text) |
|
raise requests.RequestException("Invalid JSON response") |
|
except requests.RequestException as e: |
|
print(f"Attempt {attempt + 1} failed: {e}") |
|
attempt += 1 |
|
if attempt < max_retries: |
|
time.sleep(delay) |
|
else: |
|
return {'status': 'error', 'message': 'Failed to get a valid response after multiple attempts'} |