Skip to content
Open
98 changes: 82 additions & 16 deletions runware/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,6 @@
getUUID,
fileToBase64,
createImageFromResponse,
createImageToTextFromResponse,
createEnhancedPromptsFromResponse,
instantiateDataclassList,
RunwareAPIError,
Expand Down Expand Up @@ -597,6 +596,14 @@ async def photoMaker(self, requestPhotoMaker: "IPhotoMaker") -> "Union[List[IIma
async def _photoMaker(self, requestPhotoMaker: "IPhotoMaker") -> "Union[List[IImage], IAsyncTaskResponse]":
await self.ensureConnection()

if requestPhotoMaker.model is None or requestPhotoMaker.positivePrompt is None or requestPhotoMaker.height is None or requestPhotoMaker.width is None:
raise ValueError("Standalone photoMaker requires model, positivePrompt, height, and width.")

input_images = requestPhotoMaker.inputImages if requestPhotoMaker.inputImages is not None else []
if not input_images:
raise ValueError("Standalone photoMaker requires at least one image in inputImages.")
requestPhotoMaker.inputImages = input_images

task_uuid = requestPhotoMaker.taskUUID or getUUID()
requestPhotoMaker.taskUUID = task_uuid

Expand Down Expand Up @@ -633,6 +640,16 @@ async def _photoMaker(self, requestPhotoMaker: "IPhotoMaker") -> "Union[List[IIm
request_object["includeCost"] = requestPhotoMaker.includeCost
if requestPhotoMaker.outputType:
request_object["outputType"] = requestPhotoMaker.outputType
if requestPhotoMaker.negativePrompt is not None:
request_object["negativePrompt"] = requestPhotoMaker.negativePrompt
if requestPhotoMaker.CFGScale is not None:
request_object["CFGScale"] = requestPhotoMaker.CFGScale
if requestPhotoMaker.seed is not None:
request_object["seed"] = requestPhotoMaker.seed
if requestPhotoMaker.scheduler is not None:
request_object["scheduler"] = requestPhotoMaker.scheduler
if requestPhotoMaker.checkNsfw is not None:
request_object["checkNSFW"] = requestPhotoMaker.checkNsfw
if requestPhotoMaker.webhookURL:
request_object["webhookURL"] = requestPhotoMaker.webhookURL
return await self._handleWebhookRequest(
Expand All @@ -644,6 +661,7 @@ async def _photoMaker(self, requestPhotoMaker: "IPhotoMaker") -> "Union[List[IIm

numberOfResults = requestPhotoMaker.numberResults


future, should_send = await self._register_pending_operation(
task_uuid,
expected_results=numberOfResults,
Expand Down Expand Up @@ -724,18 +742,30 @@ async def _imageInference(
for k, v in vars(requestImage.instantID).items()
if v is not None
}
if "inputImage" in instant_id_data:
instant_id_data["inputImage"] = await process_image(instant_id_data["inputImage"])

if "poseImage" in instant_id_data:
instant_id_data["poseImage"] = await process_image(instant_id_data["poseImage"])

input_images = instant_id_data.get("inputImages")
single_input = instant_id_data.get("inputImage")

if input_images is None and single_input is not None:
input_images = [single_input]

if input_images is not None:
instant_id_data["inputImages"] = await process_image(input_images)

instant_id_data.pop("inputImage", None)

ip_adapters_data = []
if requestImage.ipAdapters:
for ip_adapter in requestImage.ipAdapters:
ip_adapter_data = {
k: v for k, v in vars(ip_adapter).items() if v is not None
}
if "guideImage" in ip_adapter_data:
if "guideImages" in ip_adapter_data:
ip_adapter_data["guideImages"] = await process_image(ip_adapter_data["guideImages"])
elif "guideImage" in ip_adapter_data:
ip_adapter_data["guideImage"] = await process_image(ip_adapter_data["guideImage"])
ip_adapters_data.append(ip_adapter_data)

Expand Down Expand Up @@ -763,9 +793,20 @@ async def _imageInference(
if requestImage.puLID.inputImages:
pulid_data["inputImages"] = await process_image(requestImage.puLID.inputImages)

photo_maker_data: Dict[str, Any] = {}
if requestImage.photoMaker:
if requestImage.photoMaker.style is not None:
photo_maker_data["style"] = requestImage.photoMaker.style
if requestImage.photoMaker.strength is not None:
photo_maker_data["strength"] = requestImage.photoMaker.strength

image_list = requestImage.photoMaker.images or requestImage.photoMaker.inputImages
if image_list:
photo_maker_data["images"] = await process_image(image_list)

request_object = self._buildImageRequest(
requestImage, prompt, control_net_data_dicts,
instant_id_data, ip_adapters_data, ace_plus_plus_data, pulid_data
instant_id_data, ip_adapters_data, ace_plus_plus_data, pulid_data, photo_maker_data
)

delivery_method_enum = EDeliveryMethod(requestImage.deliveryMethod) if isinstance(requestImage.deliveryMethod,
Expand Down Expand Up @@ -905,9 +946,8 @@ async def _requestImageToText(
# Add template parameter if specified
if requestImageToText.template is not None:
task_params["template"] = requestImageToText.template
# When using template, do NOT include prompt parameter
else:
# Use the provided prompt when no template
# Add prompt only when explicitly provided (API does not support prompt in all cases)
elif requestImageToText.prompt is not None:
task_params["prompt"] = requestImageToText.prompt

# Add optional parameters if they are provided
Expand Down Expand Up @@ -935,7 +975,7 @@ async def _requestImageToText(
results = await asyncio.wait_for(future, timeout=IMAGE_OPERATION_TIMEOUT / 1000)
response = results[0]
self._handle_error_response(response)
return createImageToTextFromResponse(response)
return instantiateDataclass(IImageToText, response)
except asyncio.TimeoutError:
raise Exception(
f"Timeout waiting for image caption | TaskUUID: {taskUUID} | "
Expand Down Expand Up @@ -2326,7 +2366,7 @@ def is_text_complete(r: "Dict[str, Any]") -> bool:
finally:
await self._unregister_pending_operation(task_uuid)

def _buildImageRequest(self, requestImage: IImageInference, prompt: Optional[str], control_net_data_dicts: List[Dict], instant_id_data: Optional[Dict], ip_adapters_data: Optional[List[Dict]], ace_plus_plus_data: Optional[Dict], pulid_data: Optional[Dict]) -> Dict[str, Any]:
def _buildImageRequest(self, requestImage: IImageInference, prompt: Optional[str], control_net_data_dicts: List[Dict], instant_id_data: Optional[Dict], ip_adapters_data: Optional[List[Dict]], ace_plus_plus_data: Optional[Dict], pulid_data: Optional[Dict], photo_maker_data: Optional[Dict]) -> Dict[str, Any]:
request_object = {
"taskType": ETaskType.IMAGE_INFERENCE.value,
"taskUUID": requestImage.taskUUID,
Expand All @@ -2338,7 +2378,16 @@ def _buildImageRequest(self, requestImage: IImageInference, prompt: Optional[str
request_object["positivePrompt"] = prompt

self._addOptionalBuiltInDataTypesFields(request_object, requestImage)
self._addImageSpecialFields(request_object, requestImage, control_net_data_dicts, instant_id_data, ip_adapters_data, ace_plus_plus_data, pulid_data)
self._addImageSpecialFields(
request_object,
requestImage,
control_net_data_dicts,
instant_id_data,
ip_adapters_data,
ace_plus_plus_data,
pulid_data,
photo_maker_data,
)
self._addOptionalField(request_object, requestImage.inputs)
self._addProviderSettings(request_object, requestImage)
self._addOptionalField(request_object, requestImage.ultralytics)
Expand All @@ -2348,7 +2397,17 @@ def _buildImageRequest(self, requestImage: IImageInference, prompt: Optional[str

return request_object

def _addImageSpecialFields(self, request_object: Dict[str, Any], requestImage: IImageInference, control_net_data_dicts: List[Dict], instant_id_data: Optional[Dict], ip_adapters_data: Optional[List[Dict]], ace_plus_plus_data: Optional[Dict], pulid_data: Optional[Dict]) -> None:
def _addImageSpecialFields(
self,
request_object: Dict[str, Any],
requestImage: IImageInference,
control_net_data_dicts: List[Dict],
instant_id_data: Optional[Dict],
ip_adapters_data: Optional[List[Dict]],
ace_plus_plus_data: Optional[Dict],
pulid_data: Optional[Dict],
photo_maker_data: Optional[Dict],
) -> None:
# Add controlNet if present
if control_net_data_dicts:
request_object["controlNet"] = control_net_data_dicts
Expand All @@ -2369,10 +2428,13 @@ def _addImageSpecialFields(self, request_object: Dict[str, Any], requestImage: I

# Add embeddings if present
if requestImage.embeddings:
request_object["embeddings"] = [
{"model": embedding.model}
for embedding in requestImage.embeddings
]
embeddings_payload = []
for embedding in requestImage.embeddings:
d: Dict[str, Any] = {"model": embedding.model}
if embedding.weight is not None:
d["weight"] = embedding.weight
embeddings_payload.append(d)
request_object["embeddings"] = embeddings_payload

# Add refiner if present
if requestImage.refiner:
Expand Down Expand Up @@ -2408,6 +2470,10 @@ def _addImageSpecialFields(self, request_object: Dict[str, Any], requestImage: I
if pulid_data:
request_object["puLID"] = pulid_data

# Add photoMaker if present
if photo_maker_data:
request_object["photoMaker"] = photo_maker_data

# Add referenceImages if present
if requestImage.referenceImages:
request_object["referenceImages"] = requestImage.referenceImages
Expand Down
99 changes: 63 additions & 36 deletions runware/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,7 @@ class ILycoris:
@dataclass
class IEmbedding:
model: str
weight: Optional[float] = None


@dataclass
Expand Down Expand Up @@ -339,38 +340,26 @@ class IPhotoMaker:
numberResults: int = 1
steps: Optional[int] = None
outputType: Optional[IOutputType] = None
inputImages: List[Union[str, File]] = field(default_factory=list)
inputImages: Optional[List[Union[str, File]]] = None
style: Optional[str] = None
strength: Optional[float] = None
outputFormat: Optional[IOutputFormat] = None
includeCost: Optional[bool] = None
taskUUID: Optional[str] = None
webhookURL: Optional[str] = None
negativePrompt: Optional[str] = None
CFGScale: Optional[float] = None
seed: Optional[int] = None
scheduler: Optional[str] = None
checkNsfw: Optional[bool] = None

def __post_init__(self):
# Validate `inputImages` to ensure it has a maximum of 4 elements
if len(self.inputImages) > 4:
raise ValueError("inputImages can contain a maximum of 4 elements.")

# Validate `style` to ensure it matches one of the allowed case-sensitive options
valid_styles = {
"No style",
"Cinematic",
"Disney Character",
"Digital Art",
"Photographic",
"Fantasy art",
"Neonpunk",
"Enhance",
"Comic book",
"Lowpoly",
"Line art",
}
if self.style and self.style not in valid_styles:
raise ValueError(
f"style must be one of the following: {', '.join(valid_styles)}."
)

@dataclass
class IPhotoMakerSettings:
images: Optional[List[Union[str, File]]] = None
inputImages: Optional[List[Union[str, File]]] = None
style: Optional[str] = None
strength: Optional[float] = None

class SerializableMixin:
def serialize(self) -> Dict[str, Any]:
Expand Down Expand Up @@ -405,8 +394,9 @@ class IOutpaint:

@dataclass
class IInstantID:
inputImage: Union[File, str]
inputImage: Optional[Union[File, str]] = None
poseImage: Optional[Union[File, str]] = None
inputImages: Optional[List[Union[str, File]]] = None
identityNetStrength: Optional[float] = None
adapterStrength: Optional[float] = None
controlNetCannyWeight: Optional[float] = None
Expand All @@ -417,8 +407,13 @@ class IInstantID:
@dataclass
class IIpAdapter:
model: Union[int, str]
guideImage: Union[File, str]
guideImage: Optional[Union[File, str]] = None
guideImages: Optional[List[Union[str, File]]] = None
weight: Optional[float] = None
combineMethod: Optional[str] = None
weightType: Optional[str] = None
embedScaling: Optional[str] = None
weightComposition: Optional[float] = None


@dataclass
Expand Down Expand Up @@ -1005,22 +1000,23 @@ class IImageInference:
lycoris: Optional[List[ILycoris]] = field(default_factory=list)
includeCost: Optional[bool] = None
onPartialImages: Optional[Callable[[List[IImage], Optional[IError]], None]] = None
refiner: Optional[IRefiner] = None
refiner: Optional[Union[IRefiner, Dict[str, Any]]] = None
vae: Optional[str] = None
maskMargin: Optional[int] = None
outputQuality: Optional[int] = None
embeddings: Optional[List[IEmbedding]] = field(default_factory=list)
outpaint: Optional[IOutpaint] = None
instantID: Optional[IInstantID] = None
ipAdapters: Optional[List[IIpAdapter]] = field(default_factory=list)
embeddings: Optional[List[Union[IEmbedding, Dict[str, Any]]]] = field(default_factory=list)
outpaint: Optional[Union[IOutpaint, Dict[str, Any]]] = None
instantID: Optional[Union[IInstantID, Dict[str, Any]]] = None
ipAdapters: Optional[List[Union[IIpAdapter, Dict[str, Any]]]] = field(default_factory=list)
referenceImages: Optional[List[Union[str, File]]] = field(default_factory=list)
acePlusPlus: Optional[IAcePlusPlus] = None
puLID: Optional[IPuLID] = None
acePlusPlus: Optional[Union[IAcePlusPlus, Dict[str, Any]]] = None
puLID: Optional[Union[IPuLID, Dict[str, Any]]] = None
photoMaker: Optional[Union[IPhotoMakerSettings, Dict[str, Any]]] = None
providerSettings: Optional[ImageProviderSettings] = None
safety: Optional[Union[ISafety, Dict[str, Any]]] = None
settings: Optional[Union[ISettings, Dict[str, Any]]] = None
inputs: Optional[Union[IInputs, Dict[str, Any]]] = None
ultralytics: Optional[IUltralytics] = None
ultralytics: Optional[Union[IUltralytics, Dict[str, Any]]] = None
useCache: Optional[bool] = None
resolution: Optional[str] = None
extraArgs: Optional[Dict[str, Any]] = field(default_factory=dict)
Expand All @@ -1047,13 +1043,37 @@ def __post_init__(self, checkNsfw: Optional[bool] = None):
self.settings = ISettings(**self.settings)
if self.inputs is not None and isinstance(self.inputs, dict):
self.inputs = IInputs(**self.inputs)
if self.outpaint is not None and isinstance(self.outpaint, dict):
self.outpaint = IOutpaint(**self.outpaint)
if self.refiner is not None and isinstance(self.refiner, dict):
self.refiner = IRefiner(**self.refiner)
if self.embeddings:
self.embeddings = [
IEmbedding(**item) if isinstance(item, dict) else item
for item in self.embeddings
]
if self.photoMaker is not None and isinstance(self.photoMaker, dict):
self.photoMaker = IPhotoMakerSettings(**self.photoMaker)
if self.instantID is not None and isinstance(self.instantID, dict):
self.instantID = IInstantID(**self.instantID)
if self.acePlusPlus is not None and isinstance(self.acePlusPlus, dict):
self.acePlusPlus = IAcePlusPlus(**self.acePlusPlus)
if self.puLID is not None and isinstance(self.puLID, dict):
self.puLID = IPuLID(**self.puLID)
if self.ultralytics is not None and isinstance(self.ultralytics, dict):
self.ultralytics = IUltralytics(**self.ultralytics)
if self.ipAdapters:
self.ipAdapters = [
IIpAdapter(**item) if isinstance(item, dict) else item
for item in self.ipAdapters
]


@dataclass
class IImageCaption:
inputImages: Optional[List[Union[File, str]]] = None # Primary: array of images (UUIDs, URLs, base64, dataURI)
inputImage: Optional[Union[File, str]] = None # Convenience: single image, defaults to inputImages[0] if not provided
prompt: List[str] = field(default_factory=lambda: ["Describe this image in detail"]) # Array of prompts with default
prompt: Optional[List[str]] = None
model: Optional[str] = None # Optional: AIR ID (runware:150@1, runware:150@2) - backend handles default
includeCost: bool = False
template: Optional[str] = None
Expand Down Expand Up @@ -1092,11 +1112,18 @@ class IElevenLabsMusicSettings(SerializableMixin):
compositionPlan: IElevenLabsCompositionPlan # Music composition structure


@dataclass
class IImageToTextStructuredData:
ageGroup: Optional[str] = None
confidence: Optional[float] = None


@dataclass
class IImageToText:
taskType: ETaskType
taskUUID: str
text: str
text: Optional[str] = None
structuredData: Optional[IImageToTextStructuredData] = None
cost: Optional[float] = None


Expand Down
Loading