sharoz's picture
initial +
56b65c4
raw
history blame contribute delete
5.59 kB
"""
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 flask import Flask, request, jsonify
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() # Raise an HTTPError for bad responses
# print(response.reason)
try:
return response.json() # Returns the JSON response from the API.
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 generate_design_idea(self, prompt):
# """
# Generates a design idea using OpenAI's API.
# """
# openai.api_key = os.getenv("OPENAI_API_KEY")
# response = openai.ChatCompletion.create(
# model="gpt-3.5-turbo",
# messages=[
# {"role": "system", "content": "You are a helpful assistant."},
# {"role": "user", "content": prompt}
# ],
# max_tokens=150
# )
# return response.choices[0].message['content'].strip()
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 # seconds
while attempt < max_retries:
try:
status_response = requests.get(status_url, headers=self.headers)
status_response.raise_for_status() # Raise an HTTPError for bad responses
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'}