| """ |
| 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'} |