File size: 5,587 Bytes
56b65c4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
"""
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'} |