ExternalVendorCode/lollms
1
0
Fork 0
mirror of https://github.com/ParisNeo/lollms.git synced 2025-03-13 15:56:37 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Moving to the new Lollms style V19

Browse Source
This commit is contained in:
Saifeddine ALOUI 2025-03-06 14:00:02 +01:00
parent 0739f1c279
commit 324b8d220d
35 changed files with 1584 additions and 610 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
configs/config.yaml
View File

@ -1,5 +1,5 @@
# =================== Lord Of Large Language Multimodal Systems Configuration file ===========================
version: 157
version: 159
# video viewing and news recovering
last_viewed_video: null
@ -31,6 +31,11 @@ host: localhost
port: 9600
app_custom_logo: ""
app_custom_name: "LoLLMS"
app_show_changelogs: true
app_custom_welcome_message: ""
app_custom_slogan: ""
app_show_fun_facts: true
# Genreration parameters
discussion_prompt_separator: "!@>"

118
lollms/app.py
View File

@ -87,12 +87,6 @@ class LollmsApplication(LoLLMsCom):
# services
self.ollama = None
self.vllm = None
self.whisper = None
self.xtts = None
self.sd = None
self.comfyui = None
self.motion_ctrl = None
self.tti = None
self.tts = None
self.stt = None
@ -535,16 +529,16 @@ class LollmsApplication(LoLLMsCom):
if self.config.whisper_activate or self.config.active_stt_service == "whisper":
try:
from lollms.services.stt.whisper.lollms_whisper import LollmsWhisper
self.whisper = LollmsWhisper(self, self.config.whisper_model, self.lollms_paths.personal_outputs_path)
self.whisper = LollmsWhisper(self)
stt_services.append("whisper")
except Exception as ex:
trace_exception(ex)
if self.config.active_stt_service == "openai_whisper":
from lollms.services.stt.openai_whisper.lollms_openai_whisper import LollmsOpenAIWhisper
self.stt = LollmsOpenAIWhisper(self, self.config.openai_whisper_model, self.config.openai_whisper_key)
self.stt = LollmsOpenAIWhisper(self)
elif self.config.active_stt_service == "whisper":
from lollms.services.stt.whisper.lollms_whisper import LollmsWhisper
self.stt = LollmsWhisper(self, self.config.whisper_model)
self.stt = LollmsWhisper(self)
ASCIIColors.execute_with_animation("Loading STT services", start_stt, ASCIIColors.color_blue)
@ -553,31 +547,22 @@ class LollmsApplication(LoLLMsCom):
ASCIIColors.yellow("Loading XTTS")
try:
from lollms.services.tts.xtts.lollms_xtts import LollmsXTTS
voice=self.config.xtts_current_voice
if voice!="main_voice":
voices_folder = self.lollms_paths.custom_voices_path
else:
voices_folder = Path(__file__).parent.parent.parent/"services/xtts/voices"
self.xtts = LollmsXTTS(
self,
voices_folders=[voices_folder, self.lollms_paths.custom_voices_path],
freq=self.config.xtts_freq
self.tts = LollmsXTTS(
self
)
except Exception as ex:
trace_exception(ex)
self.warning(f"Couldn't load XTTS")
if self.config.active_tts_service == "eleven_labs_tts":
from lollms.services.tts.eleven_labs_tts.lollms_eleven_labs_tts import LollmsElevenLabsTTS
self.tts = LollmsElevenLabsTTS(self, self.config.elevenlabs_tts_model_id, self.config.elevenlabs_tts_voice_id, self.config.elevenlabs_tts_key, stability=self.config.elevenlabs_tts_voice_stability, similarity_boost=self.config.elevenlabs_tts_voice_boost)
self.tts = LollmsElevenLabsTTS(self)
elif self.config.active_tts_service == "openai_tts":
from lollms.services.tts.open_ai_tts.lollms_openai_tts import LollmsOpenAITTS
self.tts = LollmsOpenAITTS(self, self.config.openai_tts_model, self.config.openai_tts_voice, self.config.openai_tts_key)
self.tts = LollmsOpenAITTS(self)
elif self.config.active_tts_service == "fish_tts":
from lollms.services.tts.fish.lollms_fish_tts import LollmsFishAudioTTS
self.tts = LollmsFishAudioTTS(self, self.config.fish_tts_voice, self.config.fish_tts_key)
elif self.config.active_tts_service == "xtts" and self.xtts:
self.tts = self.xtts
self.tts = LollmsFishAudioTTS(self)
ASCIIColors.execute_with_animation("Loading TTS services", start_tts, ASCIIColors.color_blue)
@ -585,14 +570,14 @@ class LollmsApplication(LoLLMsCom):
if self.config.enable_sd_service:
try:
from lollms.services.tti.sd.lollms_sd import LollmsSD
self.sd = LollmsSD(self, auto_sd_base_url=self.config.sd_base_url)
self.sd = LollmsSD(self)
except:
self.warning(f"Couldn't load SD")
if self.config.enable_comfyui_service:
try:
from lollms.services.tti.comfyui.lollms_comfyui import LollmsComfyUI
self.comfyui = LollmsComfyUI(self, comfyui_base_url=self.config.comfyui_base_url)
self.comfyui = LollmsComfyUI(self)
except:
self.warning(f"Couldn't load SD")
@ -607,35 +592,37 @@ class LollmsApplication(LoLLMsCom):
self.tti = self.sd
else:
from lollms.services.tti.sd.lollms_sd import LollmsSD
self.tti = LollmsSD(self, auto_sd_base_url = self.config.sd_base_url)
self.tti = LollmsSD(self)
elif self.config.active_tti_service == "dall-e":
from lollms.services.tti.dalle.lollms_dalle import LollmsDalle
self.tti = LollmsDalle(self, self.config.dall_e_key)
self.tti = LollmsDalle(self)
elif self.config.active_tti_service == "midjourney":
from lollms.services.tti.midjourney.lollms_midjourney import LollmsMidjourney
self.tti = LollmsMidjourney(self, self.config.midjourney_key, self.config.midjourney_timeout, self.config.midjourney_retries)
self.tti = LollmsMidjourney(self)
elif self.config.active_tti_service == "comfyui" and (self.tti is None or self.tti.name!="comfyui"):
if self.comfyui:
self.tti = self.comfyui
else:
from lollms.services.tti.comfyui.lollms_comfyui import LollmsComfyUI
self.tti = LollmsComfyUI(self, comfyui_base_url=self.config.comfyui_base_url)
self.tti = LollmsComfyUI(self)
ASCIIColors.execute_with_animation("Loading loacal TTI services", start_tti, ASCIIColors.color_blue)
def start_ttv(*args, **kwargs):
if self.config.active_ttv_service == "lumalabs" and (self.ttv is None or self.ttv.name!="lumalabs"):
if self.config.active_ttv_service == "lumalabs":
try:
from lollms.services.ttv.lumalabs.lollms_lumalabs import LollmsLumaLabs
self.ttv = LollmsLumaLabs(self.config.lumalabs_key)
except:
self.ttv = LollmsLumaLabs(self)
except Exception as ex:
trace_exception(ex)
self.warning(f"Couldn't create lumalabs binding")
if self.config.active_ttv_service == "novita_ai" and (self.ttv is None or self.ttv.name!="novita_ai"):
try:
from lollms.services.ttv.novita_ai.lollms_novita_ai import LollmsNovitaAITextToVideo
self.ttv = LollmsNovitaAITextToVideo(self.config.novita_ai_key)
except:
self.warning(f"Couldn't create novita ai bvinding")
self.ttv = LollmsNovitaAITextToVideo(self)
except Exception as ex:
trace_exception(ex)
self.warning(f"Couldn't create novita ai binding")
ASCIIColors.execute_with_animation("Loading loacal TTV services", start_ttv, ASCIIColors.color_blue)
@ -670,12 +657,12 @@ class LollmsApplication(LoLLMsCom):
if self.config.whisper_activate and self.whisper is None:
try:
from lollms.services.stt.whisper.lollms_whisper import LollmsWhisper
self.whisper = LollmsWhisper(self, self.config.whisper_model, self.lollms_paths.personal_outputs_path)
self.whisper = LollmsWhisper(self)
except Exception as ex:
trace_exception(ex)
ASCIIColors.blue("Loading loacal TTS services")
if self.config.active_tts_service == "xtts" and self.xtts is None:
if self.config.active_tts_service == "xtts" and (self.tts is None or self.tts.name!="xtts"):
ASCIIColors.yellow("Loading XTTS")
try:
from lollms.services.tts.xtts.lollms_xtts import LollmsXTTS
@ -685,10 +672,8 @@ class LollmsApplication(LoLLMsCom):
else:
voices_folder = Path(__file__).parent.parent.parent/"services/xtts/voices"
self.xtts = LollmsXTTS(
self,
voices_folders=[voices_folder, self.lollms_paths.custom_voices_path],
freq=self.config.xtts_freq
self.tts = LollmsXTTS(
self
)
except Exception as ex:
trace_exception(ex)
@ -698,14 +683,14 @@ class LollmsApplication(LoLLMsCom):
if self.config.enable_sd_service and self.sd is None:
try:
from lollms.services.tti.sd.lollms_sd import LollmsSD
self.sd = LollmsSD(self, auto_sd_base_url=self.config.sd_base_url)
self.sd = LollmsSD(self)
except:
self.warning(f"Couldn't load SD")
if self.config.enable_comfyui_service and self.comfyui is None:
try:
from lollms.services.tti.comfyui.lollms_comfyui import LollmsComfyUI
self.comfyui = LollmsComfyUI(self, comfyui_base_url=self.config.comfyui_base_url)
self.comfyui = LollmsComfyUI(self)
except:
self.warning(f"Couldn't load Comfyui")
@ -713,21 +698,21 @@ class LollmsApplication(LoLLMsCom):
if self.config.active_tti_service == "diffusers" and (self.tti is None or self.tti.name!="diffusers" or self.tti.model!=self.config.diffusers_model):
from lollms.services.tti.diffusers.lollms_diffusers import LollmsDiffusers
self.tti = LollmsDiffusers(self)
elif self.config.active_tti_service == "diffusers_client" and (self.tti.base_url!=self.config.diffusers_client_base_url or self.tti.name!="diffusers_client"):
elif self.config.active_tti_service == "diffusers_client" and (self.tti.name!="diffusers_client"):
from lollms.services.tti.diffusers_client.lollms_diffusers_client import LollmsDiffusersClient
self.tti = LollmsDiffusersClient(self, base_url=self.config.diffusers_client_base_url)
self.tti = LollmsDiffusersClient(self)
elif self.config.active_tti_service == "autosd" and (self.tti is None or self.tti.name!="stable_diffusion"):
if self.sd:
self.tti = self.sd
else:
from lollms.services.tti.sd.lollms_sd import LollmsSD
self.tti = LollmsSD(self, auto_sd_base_url=self.config.sd_base_url)
self.tti = LollmsSD(self)
elif self.config.active_tti_service == "dall-e" and (self.tti is None or self.tti.name!="dall-e-2" or type(self.tti.name)!="dall-e-3"):
from lollms.services.tti.dalle.lollms_dalle import LollmsDalle
self.tti = LollmsDalle(self, self.config.dall_e_key)
self.tti = LollmsDalle(self)
elif self.config.active_tti_service == "midjourney" and (self.tti is None or self.tti.name!="midjourney"):
from lollms.services.tti.midjourney.lollms_midjourney import LollmsMidjourney
self.tti = LollmsMidjourney(self, self.config.midjourney_key, self.config.midjourney_timeout, self.config.midjourney_retries)
self.tti = LollmsMidjourney(self)
elif self.config.active_tti_service == "comfyui" and (self.tti is None or self.tti.name!="comfyui"):
if self.comfyui:
self.tti = self.comfyui
@ -738,31 +723,42 @@ class LollmsApplication(LoLLMsCom):
ASCIIColors.blue("Activating TTS service")
if self.config.active_tts_service == "eleven_labs_tts":
from lollms.services.tts.eleven_labs_tts.lollms_eleven_labs_tts import LollmsElevenLabsTTS
self.tts = LollmsElevenLabsTTS(self, self.config.elevenlabs_tts_model_id, self.config.elevenlabs_tts_voice_id, self.config.elevenlabs_tts_key, stability=self.config.elevenlabs_tts_voice_stability, similarity_boost=self.config.elevenlabs_tts_voice_boost)
self.tts = LollmsElevenLabsTTS(self)
elif self.config.active_tts_service == "openai_tts" and (self.tts is None or self.tts.name!="openai_tts"):
from lollms.services.tts.open_ai_tts.lollms_openai_tts import LollmsOpenAITTS
self.tts = LollmsOpenAITTS(self, self.config.openai_tts_model, self.config.openai_tts_voice, self.config.openai_tts_key)
self.tts = LollmsOpenAITTS(self)
elif self.config.active_tts_service == "fish_tts":
from lollms.services.tts.fish.lollms_fish_tts import LollmsFishAudioTTS
self.tts = LollmsFishAudioTTS(self, self.config.fish_tts_voice, self.config.fish_tts_key)
elif self.config.active_tts_service == "xtts" and self.xtts:
self.tts = self.xtts
self.tts = LollmsFishAudioTTS(self)
ASCIIColors.blue("Activating STT service")
if self.config.active_stt_service == "openai_whisper" and (self.tts is None or self.tts.name!="openai_whisper"):
from lollms.services.stt.openai_whisper.lollms_openai_whisper import LollmsOpenAIWhisper
self.stt = LollmsOpenAIWhisper(self, self.config.openai_whisper_model, self.config.openai_whisper_key)
self.stt = LollmsOpenAIWhisper(self)
elif self.config.active_stt_service == "whisper" and (self.tts is None or self.tts.name!="whisper") :
from lollms.services.stt.whisper.lollms_whisper import LollmsWhisper
self.stt = LollmsWhisper(self, self.config.whisper_model)
self.stt = LollmsWhisper(self)
if self.config.active_ttv_service == "lumalabs" and (self.ttv is None or self.tti.name!="lumalabs"):
try:
from lollms.services.ttv.lumalabs.lollms_lumalabs import LollmsLumaLabs
self.sd = LollmsLumaLabs(self.config.lumalabs_key)
except:
self.warning(f"Couldn't load SD")
def start_ttv(*args, **kwargs):
if self.config.active_ttv_service == "lumalabs" and (self.ttv is None or self.ttv.name!="lumalabs"):
try:
from lollms.services.ttv.lumalabs.lollms_lumalabs import LollmsLumaLabs
self.ttv = LollmsLumaLabs(self, self.config.lumalabs_key)
except:
self.warning(f"Couldn't create lumalabs binding")
if self.config.active_ttv_service == "novita_ai" and (self.ttv is None or self.ttv.name!="novita_ai"):
try:
from lollms.services.ttv.novita_ai.lollms_novita_ai import LollmsNovitaAITextToVideo
self.ttv = LollmsNovitaAITextToVideo(self,None)
except Exception as ex:
trace_exception(ex)
self.warning(f"Couldn't create novita ai binding")
ASCIIColors.execute_with_animation("Loading loacal TTV services", start_ttv, ASCIIColors.color_blue)
print("OK")
except Exception as ex:
trace_exception(ex)

1
lollms/config.py
View File

@ -586,6 +586,7 @@ class TypedConfig:
def save(self, file_path:str|Path|None=None):
self.config.save_config(file_path=file_path)
def to_dict(self, use_template=False):
if not use_template:
return self.config

7
lollms/configs/config.yaml
View File

@ -1,5 +1,5 @@
# =================== Lord Of Large Language Multimodal Systems Configuration file ===========================
version: 157
version: 159
# video viewing and news recovering
last_viewed_video: null
@ -31,6 +31,11 @@ host: localhost
port: 9600
app_custom_logo: ""
app_custom_name: "LoLLMS"
app_show_changelogs: true
app_custom_welcome_message: ""
app_custom_slogan: ""
app_show_fun_facts: true
# Genreration parameters
discussion_prompt_separator: "!@>"

4
lollms/databases/discussions_database.py
View File

@ -911,9 +911,9 @@ class Discussion:
pth = str(view_file).replace("\\","/").split('/')
if "discussion_databases" in pth:
pth = discussion_path_to_url(view_file)
self.lollms.new_message(client.client_id if client is not None else 0, content = "", message_type = MSG_OPERATION_TYPE.MSG_OPERATION_TYPE_SET_CONTENT)
self.lollms.personality.new_message()
output = f'<img src="{pth}" width="800">\n\n'
self.lollms.personality.set_message_html(output, client_id=client.client_id if client is not None else 0)
self.lollms.personality.set_message_html(output)
self.lollms.close_message(client.client_id if client is not None else 0)
if self.lollms.model.binding_type not in [BindingType.TEXT_IMAGE, BindingType.TEXT_IMAGE_VIDEO]:

7
lollms/server/configs/config.yaml
View File

@ -1,5 +1,5 @@
# =================== Lord Of Large Language Multimodal Systems Configuration file ===========================
version: 157
version: 159
# video viewing and news recovering
last_viewed_video: null
@ -31,6 +31,11 @@ host: localhost
port: 9600
app_custom_logo: ""
app_custom_name: "LoLLMS"
app_show_changelogs: true
app_custom_welcome_message: ""
app_custom_slogan: ""
app_show_fun_facts: true
# Genreration parameters
discussion_prompt_separator: "!@>"

7
lollms/server/endpoints/lollms_binding_infos.py
View File

@ -229,8 +229,10 @@ def unInstall_binding(data:BindingInstallParams):
# ----------------------------------- Bet binding settings ----------------------------------------
@router.get("/get_active_binding_settings")
def get_active_binding_settings():
@router.post("/get_active_binding_settings")
async def get_active_binding_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
print("- Retreiving binding settings")
if lollmsElfServer.binding is not None:
if hasattr(lollmsElfServer.binding,"binding_config"):
@ -245,6 +247,7 @@ def get_active_binding_settings():
# @router.post("/set_active_binding_settings")
# async def set_active_binding_settings(request: BindingSettingsRequest):
@router.post("/set_active_binding_settings")
async def set_active_binding_settings(request: Request):
data = await request.json()

129
lollms/server/endpoints/lollms_stt.py Normal file
View File

@ -0,0 +1,129 @@
"""
project: lollms_webui
file: lollms_xstt.py
author: ParisNeo
description:
This module contains a set of FastAPI routes that provide information about the Lord of Large Language and Multimodal Systems (LoLLMs) Web UI
application. These routes allow users to
"""
from fastapi import APIRouter, Request, UploadFile, File, HTTPException
from fastapi.responses import PlainTextResponse
from lollms_webui import LOLLMSWebUI
from pydantic import BaseModel
from starlette.responses import StreamingResponse
from lollms.types import MSG_OPERATION_TYPE
from lollms.main_config import BaseConfig
from lollms.utilities import find_next_available_filename, output_file_path_to_url, detect_antiprompt, remove_text_from_string, trace_exception, find_first_available_file_index, add_period, PackageManager
from lollms.security import sanitize_path, validate_path, check_access
from pathlib import Path
from ascii_colors import ASCIIColors
import os
import platform
# ----------------------- Defining router and main class ------------------------------
router = APIRouter()
lollmsElfServer:LOLLMSWebUI = LOLLMSWebUI.get_instance()
class Identification(BaseModel):
client_id: str
class ServiceListingRequest(BaseModel):
client_id: str
@router.get("/list_stt_models")
def list_stt_models():
if lollmsElfServer.config.headless_server_mode:
return {"status":False,"error":"Code execution is blocked when in headless mode for obvious security reasons!"}
if lollmsElfServer.config.host!="localhost" and lollmsElfServer.config.host!="127.0.0.1":
return {"status":False,"error":"Code execution is blocked when the server is exposed outside for very obvious reasons!"}
ASCIIColors.yellow("Listing voices")
return {"voices":lollmsElfServer.stt.get_models()}
class LollmsAudio2TextRequest(BaseModel):
wave_file_path: str
model: str = None
fn:str = None
@router.post("/audio2text")
async def audio2text(request: LollmsAudio2TextRequest):
if lollmsElfServer.config.headless_server_mode:
return {"status":False,"error":"Code execution is blocked when in headless mode for obvious security reasons!"}
if lollmsElfServer.config.host!="localhost" and lollmsElfServer.config.host!="127.0.0.1":
return {"status":False,"error":"Code execution is blocked when the server is exposed outside for very obvious reasons!"}
result = lollmsElfServer.whisper.transcribe(str(request.wave_file_path))
return PlainTextResponse(result)
@router.post("/list_stt_services")
async def list_stt_services(request: ServiceListingRequest):
"""
Dumb endpoint that returns a static list of STT services.
Args:
request (ServiceListingRequest): The request body containing the client_id.
Returns:
List[str]: A list of STT service names.
"""
# Validate the client_id (dumb validation for demonstration)
check_access(lollmsElfServer, request.client_id)
# Static list of STT services
stt_services = [
{"name": "whisper", "caption":"Whisper", "help":"Whisper local speech to text service"},
{"name": "openai_whisper", "caption":"Open AI Whisper STT", "help":"Open ai remote speech to text service"},
]
return stt_services
@router.post("/get_active_stt_settings")
async def get_active_stt_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
print("- Retreiving stt settings")
if lollmsElfServer.stt is not None:
if hasattr(lollmsElfServer.stt,"service_config"):
return lollmsElfServer.stt.service_config.config_template.template
else:
return {}
else:
return {}
@router.post("/set_active_stt_settings")
async def set_active_stt_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
settings = data["settings"]
"""
Sets the active stt settings.
:param request: The sttSettingsRequest object.
:return: A JSON response with the status of the operation.
"""
try:
print("- Setting stt settings")
if lollmsElfServer.stt is not None:
if hasattr(lollmsElfServer.stt,"service_config"):
lollmsElfServer.stt.service_config.update_template(settings)
lollmsElfServer.stt.service_config.config.save_config()
lollmsElfServer.stt.settings_updated()
return {'status':True}
else:
return {'status':False}
else:
return {'status':False}
except Exception as ex:
trace_exception(ex)
lollmsElfServer.error(ex)
return {"status":False,"error":str(ex)}

80
lollms/server/endpoints/lollms_tti.py
View File

@ -1,4 +1,4 @@
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel
from typing import Optional
from base64 import b64encode
@ -7,9 +7,10 @@ from PIL import Image
from fastapi import APIRouter
from lollms_webui import LOLLMSWebUI
from pydantic import BaseModel
from typing import List
from ascii_colors import trace_exception
from lollms.security import check_access
from ascii_colors import ASCIIColors
router = APIRouter()
lollmsElfServer = LOLLMSWebUI.get_instance()
@ -23,6 +24,36 @@ class ImageRequest(BaseModel):
class ImageResponse(BaseModel):
image: str
class ServiceListingRequest(BaseModel):
client_id: str
@router.post("/list_tti_services")
async def list_tti_services(request: ServiceListingRequest):
"""
Dumb endpoint that returns a static list of TTI services.
Args:
request (ServiceListingRequest): The request body containing the client_id.
Returns:
List[str]: A list of TTV service names.
"""
# Validate the client_id (dumb validation for demonstration)
check_access(lollmsElfServer, request.client_id)
# Static list of TTV services
ttv_services = [
{"name": "diffusers", "caption":"Diffusers", "help":"Diffusers text to image service"},
{"name": "diffusers_client", "caption":"Diffusers Client", "help":"Diffusers Client text to image service"},
{"name": "autosd", "caption":"AUTO1111's SD", "help":"AUTO1111's SD text to image service"},
{"name": "dall-e", "caption":"Open AI's DALL-E", "help":"Open AI's DALL-E text to image service"},
{"name": "midjourney", "caption":"Midjourney", "help":"Midjourney text to image service"},
{"name": "comfyui", "caption":"Comfyui", "help":"Comfyui text to image service"},
{"name": "fooocus", "caption":"Fooocus", "help":"Fooocus text to image service"},
]
return ttv_services
@router.post("/generate_image", response_model=ImageResponse)
async def generate_image(request: ImageRequest):
try:
@ -89,3 +120,46 @@ def build_image(prompt, negative_prompt, width, height, return_format="markdown"
trace_exception(ex)
print(f"Error in build_image: {str(ex)}")
return None
@router.post("/get_active_tti_settings")
async def get_active_tti_settings(request: ServiceListingRequest):
check_access(lollmsElfServer,request.client_id)
ASCIIColors.info("Retreiving tti settings")
if lollmsElfServer.tti is not None:
if hasattr(lollmsElfServer.tti,"service_config"):
return lollmsElfServer.tti.service_config.config_template.template
else:
return {}
else:
return {}
@router.post("/set_active_tti_settings")
async def set_active_ttv_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
settings = data["settings"]
"""
Sets the active ttv settings.
:param request: The ttvSettingsRequest object.
:return: A JSON response with the status of the operation.
"""
try:
print("- Setting ttv settings")
if lollmsElfServer.ttv is not None:
if hasattr(lollmsElfServer.tti,"service_config"):
lollmsElfServer.tti.service_config.update_template(settings)
lollmsElfServer.tti.service_config.config.save_config()
lollmsElfServer.tti.settings_updated()
return {'status':True}
else:
return {'status':False}
else:
return {'status':False}
except Exception as ex:
trace_exception(ex)
lollmsElfServer.error(ex)
return {"status":False,"error":str(ex)}

71
lollms/server/endpoints/lollms_ttm.py
View File

@ -1,4 +1,4 @@
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, Request
from pydantic import BaseModel
from typing import Optional
from base64 import b64encode
@ -14,27 +14,74 @@ from lollms.security import check_access
router = APIRouter()
lollmsElfServer = LOLLMSWebUI.get_instance()
# Define a Pydantic model for the request body
class TTMServiceRequest(BaseModel):
class ServiceListingRequest(BaseModel):
client_id: str
@router.post("/list_ttm_services", response_model=List[str])
async def list_ttm_services(request: TTMServiceRequest):
# Define a Pydantic model for the request body
@router.post("/list_ttm_services")
async def list_ttm_services(request: ServiceListingRequest):
"""
Dumb endpoint that returns a static list of TTM services.
Args:
request (TTMServiceRequest): The request body containing the client_id.
request (ServiceListingRequest): The request body containing the client_id.
Returns:
List[str]: A list of TTV service names.
List[str]: A list of TTM service names.
"""
# Validate the client_id (dumb validation for demonstration)
check_access(lollmsElfServer, request.client_id)
# Static list of TTV services
ttm_services = ["suno"]
return ttm_services
# Static list of TTM services
ttm_services = [
{"name": "suno", "caption":"Suno AI", "help":"Suno ai"},
{"name": "music_gen", "caption":"Music Gen", "help":"Music Gen"},
]
return ttm_services
@router.post("/get_active_ttm_settings")
async def get_active_ttm_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
print("- Retreiving ttm settings")
if lollmsElfServer.ttm is not None:
if hasattr(lollmsElfServer.ttm,"service_config"):
return lollmsElfServer.ttm.service_config.config_template.template
else:
return {}
else:
return {}
@router.post("/set_active_ttm_settings")
async def set_active_ttm_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
settings = data["settings"]
"""
Sets the active ttm settings.
:param request: The Request object.
:return: A JSON response with the status of the operation.
"""
try:
print("- Setting ttm settings")
if lollmsElfServer.ttm is not None:
if hasattr(lollmsElfServer.ttm,"service_config"):
lollmsElfServer.ttm.service_config.update_template(settings)
lollmsElfServer.ttm.service_config.config.save_config()
lollmsElfServer.ttm.settings_updated()
return {'status':True}
else:
return {'status':False}
else:
return {'status':False}
except Exception as ex:
trace_exception(ex)
lollmsElfServer.error(ex)
return {"status":False,"error":str(ex)}

98
lollms/server/endpoints/lollms_tts.py
View File

@ -29,6 +29,8 @@ lollmsElfServer:LOLLMSWebUI = LOLLMSWebUI.get_instance()
class Identification(BaseModel):
client_id: str
class ServiceListingRequest(BaseModel):
client_id: str
# ----------------------- voice ------------------------------
@router.get("/list_voices")
@ -45,17 +47,6 @@ def list_voices():
ASCIIColors.yellow("Listing voices")
return {"voices":lollmsElfServer.tts.get_voices()}
@router.get("/list_stt_models")
def list_stt_models():
if lollmsElfServer.config.headless_server_mode:
return {"status":False,"error":"Code execution is blocked when in headless mode for obvious security reasons!"}
if lollmsElfServer.config.host!="localhost" and lollmsElfServer.config.host!="127.0.0.1":
return {"status":False,"error":"Code execution is blocked when the server is exposed outside for very obvious reasons!"}
ASCIIColors.yellow("Listing voices")
return {"voices":lollmsElfServer.stt.get_models()}
@router.get("/list_tts_models")
def list_tts_models():
if lollmsElfServer.config.headless_server_mode:
@ -93,23 +84,6 @@ async def set_voice(request: Request):
return {"status":False,"error":str(ex)}
class LollmsAudio2TextRequest(BaseModel):
wave_file_path: str
model: str = None
fn:str = None
@router.post("/audio2text")
async def audio2text(request: LollmsAudio2TextRequest):
if lollmsElfServer.config.headless_server_mode:
return {"status":False,"error":"Code execution is blocked when in headless mode for obvious security reasons!"}
if lollmsElfServer.config.host!="localhost" and lollmsElfServer.config.host!="127.0.0.1":
return {"status":False,"error":"Code execution is blocked when the server is exposed outside for very obvious reasons!"}
result = lollmsElfServer.whisper.transcribe(str(request.wave_file_path))
return PlainTextResponse(result)
class LollmsText2AudioRequest(BaseModel):
client_id: str
@ -276,3 +250,71 @@ def get_snd_output_devices():
else:
return []
@router.post("/list_tts_services")
async def list_tts_services(request: ServiceListingRequest):
"""
Dumb endpoint that returns a static list of TTS services.
Args:
request (ServiceListingRequest): The request body containing the client_id.
Returns:
List[str]: A list of TTS service names.
"""
# Validate the client_id (dumb validation for demonstration)
check_access(lollmsElfServer, request.client_id)
# Static list of TTS services
tts_services = [
{"name": "xtts", "caption":"XTTS", "help":"Xtts local text to speach service"},
{"name": "eleven_labs", "caption":"Eleven labs", "help":"Eleven labs remote text to speach service"},
{"name": "lollms_fish_tts", "caption":"Fish TTS", "help":"Fish remote text to speach service"},
{"name": "lollms_openai_tts", "caption":"Open AI TTS", "help":"Open ai remote text to speach service"},
]
return tts_services
@router.post("/get_active_tts_settings")
async def get_active_tts_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
print("- Retreiving tts settings")
if lollmsElfServer.tts is not None:
if hasattr(lollmsElfServer.tts,"service_config"):
return lollmsElfServer.tts.service_config.config_template.template
else:
return {}
else:
return {}
@router.post("/set_active_tts_settings")
async def set_active_tts_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
settings = data["settings"]
"""
Sets the active tts settings.
:param request: The ttsSettingsRequest object.
:return: A JSON response with the status of the operation.
"""
try:
print("- Setting tts settings")
if lollmsElfServer.tts is not None:
if hasattr(lollmsElfServer.tts,"service_config"):
lollmsElfServer.tts.service_config.update_template(settings)
lollmsElfServer.tts.service_config.config.save_config()
lollmsElfServer.tts.settings_updated()
return {'status':True}
else:
return {'status':False}
else:
return {'status':False}
except Exception as ex:
trace_exception(ex)
lollmsElfServer.error(ex)
return {"status":False,"error":str(ex)}

71
lollms/server/endpoints/lollms_ttv.py
View File

@ -1,4 +1,4 @@
from fastapi import APIRouter, HTTPException
from fastapi import APIRouter, Request
from pydantic import BaseModel
from typing import Optional
from base64 import b64encode
@ -10,17 +10,27 @@ from pydantic import BaseModel
from typing import List
from ascii_colors import trace_exception
from lollms.security import check_access
from ascii_colors import ASCIIColors
router = APIRouter()
lollmsElfServer = LOLLMSWebUI.get_instance()
# Define a Pydantic model for the request body
class TTVServiceRequest(BaseModel):
class ServiceListingRequest(BaseModel):
client_id: str
@router.post("/list_ttv_services", response_model=List[str])
async def list_ttv_services(request: TTVServiceRequest):
class TTVServiceGetConfigRequest(BaseModel):
client_id: str
service_name: str
class TTVServiceSetConfigRequest(BaseModel):
client_id: str
service_name: str
@router.post("/list_ttv_services")
async def list_ttv_services(request: ServiceListingRequest):
"""
Dumb endpoint that returns a static list of TTV services.
@ -35,6 +45,55 @@ async def list_ttv_services(request: TTVServiceRequest):
# Static list of TTV services
ttv_services = ["novita_ai", "cog_video_x", "diffusers", "lumalab"]
ttv_services = [
{"name": "novita_ai", "caption":"Novita AI", "help":"Novita ai text to video services"},
{"name": "diffusers", "caption":"Diffusers", "help":"Diffusers based Local text to video services"},
{"name": "lumalabs", "caption":"Luma labs", "help":"Luma labs text to video services"},
{"name": "cog_video_x", "caption":"Cog Video", "help":"Cog video"},
]
return ttv_services
return ttv_services
@router.post("/get_active_ttv_settings")
async def get_active_ttv_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
print("- Retreiving ttv settings")
if lollmsElfServer.ttv is not None:
if hasattr(lollmsElfServer.ttv,"service_config"):
return lollmsElfServer.ttv.service_config.config_template.template
else:
return {}
else:
return {}
@router.post("/set_active_ttv_settings")
async def set_active_ttv_settings(request: Request):
data = await request.json()
check_access(lollmsElfServer,data["client_id"])
settings = data["settings"]
"""
Sets the active ttv settings.
:param request: The ttvSettingsRequest object.
:return: A JSON response with the status of the operation.
"""
try:
print("- Setting ttv settings")
if lollmsElfServer.ttv is not None:
if hasattr(lollmsElfServer.ttv,"service_config"):
lollmsElfServer.ttv.service_config.update_template(settings)
lollmsElfServer.ttv.service_config.config.save_config()
lollmsElfServer.ttv.settings_updated()
return {'status':True}
else:
return {'status':False}
else:
return {'status':False}
except Exception as ex:
trace_exception(ex)
lollmsElfServer.error(ex)
return {"status":False,"error":str(ex)}

3
lollms/server/events/lollms_personality_events.py
View File

@ -65,6 +65,7 @@ def add_events(sio:socketio):
@sio.on('send_file_chunk')
def send_file_chunk(sid, data):
client_id = sid
client = lollmsElfServer.session.get_client(client_id)
@ -106,7 +107,7 @@ def add_events(sio:socketio):
if is_last_chunk:
lollmsElfServer.success('File received and saved successfully')
lollmsElfServer.HideBlockingMessage()
lollmsElfServer.ShowBlockingMessage(f"File received {file_path.name}.\nVectorizing the data ...")
lollmsElfServer.ShowBlockingMessage(f"File received {file_path.name}.\nProcessing the file ...")
if lollmsElfServer.personality.processor:
result = client.discussion.add_file(file_path, client, lollmsElfServer.tasks_library, partial(lollmsElfServer.process_data, client_id=client_id))

58
lollms/service.py Normal file
View File

@ -0,0 +1,58 @@
"""
Lollms TTI Module
=================
This module is part of the Lollms library, designed to provide Text-to-Image (TTI) functionalities within the LollmsApplication framework. The base class `LollmsTTI` is intended to be inherited and implemented by other classes that provide specific TTI functionalities.
Author: ParisNeo, a computer geek passionate about AI
"""
from abc import ABC, abstractmethod
from lollms.app import LollmsApplication
from pathlib import Path
from typing import List, Dict
from lollms.main_config import LOLLMSConfig
from lollms.config import TypedConfig
class LollmsSERVICE(ABC):
"""
LollmsSERVICE is a base class for implementing Lollms services.
Attributes:
name (str): The name if the service.
app (LollmsApplication): The instance of the main Lollms application.
service_config (TypedConfig): Specific configurations for the current service
"""
def __init__(
self,
name:str,
app: LollmsApplication,
service_config: TypedConfig,
):
"""
Initializes the LollmsTTI class with the given parameters.
Args:
app (LollmsApplication): The instance of the main Lollms application.
model (str, optional): The TTI model to be used for image generation. Defaults to an empty string.
api_key (str, optional): API key for accessing external TTI services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output image files will be saved. Defaults to None.
"""
self.ready = False
self.name = name
self.app = app
self.service_config = service_config
self.sync_configuration()
def sync_configuration(self):
self.configuration_file_path = self.app.lollms_paths.personal_configuration_path/"services"/self.name/f"config.yaml"
self.configuration_file_path.parent.mkdir(parents=True, exist_ok=True)
self.service_config.config.file_path = self.configuration_file_path
try:
self.service_config.config.load_config()
except:
self.service_config.config.save_config()
self.service_config.sync()
@abstractmethod
def settings_updated(self):
pass

38
lollms/services/stt/openai_whisper/lollms_openai_whisper.py
View File

@ -39,14 +39,40 @@ class LollmsOpenAIWhisper(LollmsSTT):
def __init__(
self,
app:LollmsApplication,
model="whisper-1",
api_key="",
output_path=None
output_folder:str|Path=None
):
super().__init__("openai_whisper",app, model, output_path)
self.client = OpenAI(api_key=api_key)
self.ready = True
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
api_key = os.getenv("OPENAI_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{"name":"api_key", "type":"str", "value":api_key, "help":"A valid Open AI key to generate text using anthropic api"},
{
"name": "model",
"type": "str",
"value": "whisper-1",
"options": ["whisper-1"],
"help": "The model to be used"
},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("openai_whisper",app, service_config, output_folder)
self.client = OpenAI(api_key=self.service_config.api_key)
self.ready = True
def settings_updated(self):
self.client = OpenAI(api_key=self.service_config.api_key)
self.ready = True
def transcribe(
self,
wav_path: str|Path,

35
lollms/services/stt/whisper/lollms_whisper.py
View File

@ -15,6 +15,7 @@ from ascii_colors import ASCIIColors, trace_exception
from lollms.paths import LollmsPaths
import subprocess
import pipmaster as pm
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
try:
if not pm.is_installed("openai-whisper"):
pm.install("openai-whisper")
@ -40,17 +41,41 @@ class LollmsWhisper(LollmsSTT):
def __init__(
self,
app:LollmsApplication,
model="small",
output_path=None
output_folder:str|Path=None
):
super().__init__("whisper",app, model, output_path)
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{
"name": "model_name",
"type": "str",
"value": "base",
"options": ["tiny", "tiny.en", "base", "base.en", "small", "small.en", "medium", "medium.en", "large", "large-v2", "large-v3", "turbo"],
"help": "The engine to be used"
},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("whisper",app, service_config, output_folder)
try:
self.whisper = whisper.load_model(model)
self.whisper = whisper.load_model(service_config.model_name)
except:
ASCIIColors.red("Couldn't load whisper model!\nWhisper will be disabled")
self.whisper = None
self.ready = True
def settings_updated(self):
pass
def transcribe(
self,
wave_path: str|Path

266
lollms/services/tti/comfyui/lollms_comfyui.py
View File

@ -28,26 +28,22 @@ from typing import List, Dict, Any
import uuid
from ascii_colors import ASCIIColors, trace_exception
from lollms.paths import LollmsPaths
from lollms.utilities import git_pull, show_yes_no_dialog, run_script_in_env, create_conda_env, run_python_script_in_env, PackageManager
from lollms.utilities import git_pull, show_yes_no_dialog, PackageManager
from lollms.tti import LollmsTTI
import subprocess
import shutil
from tqdm import tqdm
import threading
import pipmaster as pm
if not PackageManager.check_package_installed("websocket"):
PackageManager.install_or_update("websocket-client")
if not pm.is_installed("websocket"):
pm.install("websocket-client")
import websocket
if not PackageManager.check_package_installed("urllib"):
PackageManager.install_or_update("urllib")
if not pm.is_installed("urllib"):
pm.install("urllib")
from urllib import request, parse
def verify_comfyui(lollms_paths:LollmsPaths):
# Clone repository
root_dir = lollms_paths.personal_path
shared_folder = root_dir/"shared"
comfyui_folder = shared_folder / "comfyui"
return comfyui_folder.exists()
def download_file(url, folder_path, local_filename):
# Make sure 'folder_path' exists
@ -65,76 +61,7 @@ def download_file(url, folder_path, local_filename):
return local_filename
def install_comfyui(lollms_app:LollmsApplication):
root_dir = lollms_app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
comfyui_folder = shared_folder / "comfyui"
if comfyui_folder.exists():
if show_yes_no_dialog("warning!","I have detected that there is a previous installation of Comfyui.\nShould I remove it and continue installing?"):
shutil.rmtree(comfyui_folder)
elif show_yes_no_dialog("warning!","Continue installation?"):
ASCIIColors.cyan("Installing comfyui conda environment with python 3.10")
create_conda_env("comfyui","3.10")
ASCIIColors.cyan("Done")
return
else:
return
subprocess.run(["git", "clone", "https://github.com/ParisNeo/ComfyUI.git", str(comfyui_folder)])
subprocess.run(["git", "clone", "https://github.com/ParisNeo/ComfyUI-Manager.git", str(comfyui_folder/"custom_nodes/ComfyUI-Manager")])
subprocess.run(["git", "clone", "https://github.com/AlekPet/ComfyUI_Custom_Nodes_AlekPet.git", str(comfyui_folder/"custom_nodes/ComfyUI_Custom_Nodes_AlekPet")])
subprocess.run(["git", "clone", "https://github.com/ParisNeo/lollms_nodes_suite.git", str(comfyui_folder/"custom_nodes/lollms_nodes_suite")])
subprocess.run(["git", "clone", "https://github.com/jags111/efficiency-nodes-comfyui.git", str(comfyui_folder/"custom_nodes/efficiency-nodes-comfyui")])
subprocess.run(["git", "clone", "https://github.com/Kosinkadink/ComfyUI-Advanced-ControlNet.git", str(comfyui_folder/"custom_nodes/ComfyUI-Advanced-ControlNet")])
subprocess.run(["git", "clone", "https://github.com/Kosinkadink/ComfyUI-VideoHelperSuite.git", str(comfyui_folder/"custom_nodes/ComfyUI-VideoHelperSuite")])
subprocess.run(["git", "clone", "https://github.com/LykosAI/ComfyUI-Inference-Core-Nodes.git", str(comfyui_folder/"custom_nodes/ComfyUI-Inference-Core-Nodes")])
subprocess.run(["git", "clone", "https://github.com/Fannovel16/comfyui_controlnet_aux.git", str(comfyui_folder/"custom_nodes/comfyui_controlnet_aux")])
subprocess.run(["git", "clone", "https://github.com/Kosinkadink/ComfyUI-AnimateDiff-Evolved.git", str(comfyui_folder/"custom_nodes/ComfyUI-AnimateDiff-Evolved")])
if show_yes_no_dialog("warning!","You will need to install an image generation model.\nDo you want to install an image model from civitai?\nI suggest Juggernaut XL.\nIt is a very good model.\nyou can always install more models afterwards in your comfyui folder/models.checkpoints"):
download_file("https://civitai.com/api/download/models/357609", comfyui_folder/"models/checkpoints","Juggernaut_XL.safetensors")
if show_yes_no_dialog("warning!","Do you want to install a video model from hugging face?\nIsuggest SVD XL."):
download_file("https://huggingface.co/stabilityai/stable-video-diffusion-img2vid-xt/blob/main/svd_xt.safetensors", comfyui_folder/"models/checkpoints","svd_xt.safetensors")
if show_yes_no_dialog("warning!","Do you want to install all control net models?"):
(comfyui_folder/"models/controlnet").mkdir(parents=True, exist_ok=True)
download_file("https://huggingface.co/thibaud/controlnet-openpose-sdxl-1.0/resolve/main/OpenPoseXL2.safetensors", comfyui_folder/"models/controlnet","OpenPoseXL2.safetensors")
download_file("https://huggingface.co/diffusers/controlnet-depth-sdxl-1.0/resolve/main/diffusion_pytorch_model.safetensors", comfyui_folder/"models/controlnet","DepthMap_XL.safetensors")
if show_yes_no_dialog("warning!","Do you want to install all animation models?"):
(comfyui_folder/"models/animatediff_models").mkdir(parents=True, exist_ok=True)
download_file("https://huggingface.co/guoyww/animatediff/resolve/cd71ae134a27ec6008b968d6419952b0c0494cf2/mm_sdxl_v10_beta.ckpt", comfyui_folder/"models/animatediff_models","mm_sdxl_v10_beta.ckpt")
create_conda_env("comfyui","3.10")
if lollms_app.config.hardware_mode in ["nvidia", "nvidia-tensorcores"]:
run_python_script_in_env("comfyui", "-m pip install --pre torch torchvision torchaudio --index-url https://download.pytorch.org/whl/nightly/cu121")
if lollms_app.config.hardware_mode in ["amd", "amd-noavx"]:
run_python_script_in_env("comfyui", "-m pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/rocm5.7")
elif lollms_app.config.hardware_mode in ["cpu", "cpu-noavx"]:
run_python_script_in_env("comfyui", "-m pip install --pre torch torchvision torchaudio")
run_python_script_in_env("comfyui", f"-m pip install -r {comfyui_folder}/requirements.txt")
lollms_app.comfyui = LollmsComfyUI(lollms_app)
ASCIIColors.green("Comfyui installed successfully")
lollms_app.HideBlockingMessage()
def upgrade_comfyui(lollms_app:LollmsApplication):
root_dir = lollms_app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
comfyui_folder = shared_folder / "comfyui"
if not comfyui_folder.exists():
lollms_app.InfoMessage("Comfyui is not installed, install it first")
return
subprocess.run(["git", "pull", str(comfyui_folder)])
subprocess.run(["git", "pull", str(comfyui_folder/"custom_nodes/ComfyUI-Manager")])
subprocess.run(["git", "pull", str(comfyui_folder/"custom_nodes/efficiency-nodes-comfyui")])
ASCIIColors.success("DONE")
def get_comfyui(lollms_paths:LollmsPaths):
@ -154,33 +81,68 @@ def get_comfyui(lollms_paths:LollmsPaths):
class LollmsComfyUI(LollmsTTI):
has_controlnet = False
def __init__(
self,
app:LollmsApplication,
wm = "Artbot",
max_retries=50,
comfyui_base_url=None,
share=False,
wait_for_service=False
):
super().__init__("comfyui", app)
if comfyui_base_url=="" or comfyui_base_url=="http://127.0.0.1:8188/":
comfyui_base_url = None
def __init__(self, app:LollmsApplication, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{
"name": "base_url",
"type": "str",
"value": "http://127.0.0.1:8188/",
"help": "The base URL for the service. This is the address where the service is hosted (e.g., http://127.0.0.1:8188/)."
},
{
"name": "wm",
"type": "str",
"value": "lollms",
"help": "Watermarking text or identifier to be used in the service."
},
{
"name": "max_retries",
"type": "int",
"value": 50,
"help": "The maximum number of retries to attempt before determining that the service is unavailable."
},
{
"name": "local_service",
"type": "bool",
"value": False,
"help": "If set to true, a local instance of the service will be installed and used."
},
{
"name": "start_service_at_startup",
"type": "bool",
"value": False,
"help": "If set to true, the service will automatically start at startup. This also enables the local service option."
},
{
"name": "share",
"type": "bool",
"value": False,
"help": "If set to true, the server will be accessible from outside your local machine (e.g., over the internet)."
}
]),
BaseConfig(config={
})
)
super().__init__("comfyui",app, service_config)
# Get the current directory
lollms_paths = app.lollms_paths
self.app = app
root_dir = lollms_paths.personal_path
self.wm = wm
# Store the path to the script
if comfyui_base_url is None:
self.comfyui_base_url = "http://127.0.0.1:8188/"
if not verify_comfyui(lollms_paths):
install_comfyui(app.lollms_paths)
else:
self.comfyui_base_url = comfyui_base_url
# If this is requiring a local service then verify if it is on
if self.service_config.local_service:
if not self.verify_comfyui():
self.install()
self.comfyui_url = self.comfyui_base_url+"/comfyuiapi/v1"
self.comfyui_url = self.service_config.base_url+"/comfyuiapi/v1"
shared_folder = root_dir/"shared"
self.comfyui_folder = shared_folder / "comfyui"
self.output_dir = root_dir / "outputs/comfyui"
@ -198,17 +160,19 @@ class LollmsComfyUI(LollmsTTI):
ASCIIColors.red(" Forked from comfyanonymous's Comfyui nodes system")
ASCIIColors.red(" Integration in lollms by ParisNeo")
if not self.wait_for_service(1,False) and comfyui_base_url is None:
if not self.wait_for_service(1,False) and self.service_config.local_service and self.service_config.start_service_at_startup and self.service_config.base_url is None:
ASCIIColors.info("Loading lollms_comfyui")
if platform.system() == "Windows":
ASCIIColors.info("Running on windows")
script_path = self.comfyui_folder / "main.py"
if share:
run_python_script_in_env("comfyui", str(script_path), cwd=self.comfyui_folder, wait=False)
if self.service_config.share:
pass # TODO: implement
#run_python_script_in_env("comfyui", str(script_path), cwd=self.comfyui_folder, wait=False)
# subprocess.Popen("conda activate " + str(script_path) +" --share", cwd=self.comfyui_folder)
else:
run_python_script_in_env("comfyui", str(script_path), cwd=self.comfyui_folder, wait=False)
pass # TODO: implement
# run_python_script_in_env("comfyui", str(script_path), cwd=self.comfyui_folder, wait=False)
# subprocess.Popen(script_path, cwd=self.comfyui_folder)
else:
ASCIIColors.info("Running on linux/MacOs")
@ -216,19 +180,97 @@ class LollmsComfyUI(LollmsTTI):
ASCIIColors.info(f"launcher path: {script_path}")
ASCIIColors.info(f"comfyui path: {self.comfyui_folder}")
if share:
run_script_in_env("comfyui","bash " + script_path +" --share", cwd=self.comfyui_folder)
if self.service_config.share:
pass # TODO: implement
# run_script_in_env("comfyui","bash " + script_path +" --share", cwd=self.comfyui_folder)
# subprocess.Popen("conda activate " + str(script_path) +" --share", cwd=self.comfyui_folder)
else:
run_script_in_env("comfyui","bash " + script_path, cwd=self.comfyui_folder)
pass # TODO: implement
# run_script_in_env("comfyui","bash " + script_path, cwd=self.comfyui_folder)
ASCIIColors.info("Process done")
ASCIIColors.success("Launching Comfyui succeeded")
# Wait until the service is available at http://127.0.0.1:8188//
if wait_for_service:
self.wait_for_service()
else:
self.wait_for_service_in_another_thread(max_retries=max_retries)
def verify_comfyui(self):
# Clone repository
root_dir = self.app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
comfyui_folder = shared_folder / "comfyui"
return comfyui_folder.exists()
def install(self):
root_dir = self.app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
comfyui_folder = shared_folder / "comfyui"
if comfyui_folder.exists():
if show_yes_no_dialog("warning!","I have detected that there is a previous installation of Comfyui.\nShould I remove it and continue installing?"):
shutil.rmtree(comfyui_folder)
elif show_yes_no_dialog("warning!","Continue installation?"):
ASCIIColors.cyan("Installing comfyui conda environment with python 3.10")
create_conda_env("comfyui","3.10")
ASCIIColors.cyan("Done")
return
else:
return
subprocess.run(["git", "clone", "https://github.com/ParisNeo/ComfyUI.git", str(comfyui_folder)])
subprocess.run(["git", "clone", "https://github.com/ParisNeo/ComfyUI-Manager.git", str(comfyui_folder/"custom_nodes/ComfyUI-Manager")])
subprocess.run(["git", "clone", "https://github.com/AlekPet/ComfyUI_Custom_Nodes_AlekPet.git", str(comfyui_folder/"custom_nodes/ComfyUI_Custom_Nodes_AlekPet")])
subprocess.run(["git", "clone", "https://github.com/ParisNeo/lollms_nodes_suite.git", str(comfyui_folder/"custom_nodes/lollms_nodes_suite")])
subprocess.run(["git", "clone", "https://github.com/jags111/efficiency-nodes-comfyui.git", str(comfyui_folder/"custom_nodes/efficiency-nodes-comfyui")])
subprocess.run(["git", "clone", "https://github.com/Kosinkadink/ComfyUI-Advanced-ControlNet.git", str(comfyui_folder/"custom_nodes/ComfyUI-Advanced-ControlNet")])
subprocess.run(["git", "clone", "https://github.com/Kosinkadink/ComfyUI-VideoHelperSuite.git", str(comfyui_folder/"custom_nodes/ComfyUI-VideoHelperSuite")])
subprocess.run(["git", "clone", "https://github.com/LykosAI/ComfyUI-Inference-Core-Nodes.git", str(comfyui_folder/"custom_nodes/ComfyUI-Inference-Core-Nodes")])
subprocess.run(["git", "clone", "https://github.com/Fannovel16/comfyui_controlnet_aux.git", str(comfyui_folder/"custom_nodes/comfyui_controlnet_aux")])
subprocess.run(["git", "clone", "https://github.com/Kosinkadink/ComfyUI-AnimateDiff-Evolved.git", str(comfyui_folder/"custom_nodes/ComfyUI-AnimateDiff-Evolved")])
if show_yes_no_dialog("warning!","You will need to install an image generation model.\nDo you want to install an image model from civitai?\nI suggest Juggernaut XL.\nIt is a very good model.\nyou can always install more models afterwards in your comfyui folder/models.checkpoints"):
download_file("https://civitai.com/api/download/models/357609", comfyui_folder/"models/checkpoints","Juggernaut_XL.safetensors")
if show_yes_no_dialog("warning!","Do you want to install a video model from hugging face?\nIsuggest SVD XL."):
download_file("https://huggingface.co/stabilityai/stable-video-diffusion-img2vid-xt/blob/main/svd_xt.safetensors", comfyui_folder/"models/checkpoints","svd_xt.safetensors")
if show_yes_no_dialog("warning!","Do you want to install all control net models?"):
(comfyui_folder/"models/controlnet").mkdir(parents=True, exist_ok=True)
download_file("https://huggingface.co/thibaud/controlnet-openpose-sdxl-1.0/resolve/main/OpenPoseXL2.safetensors", comfyui_folder/"models/controlnet","OpenPoseXL2.safetensors")
download_file("https://huggingface.co/diffusers/controlnet-depth-sdxl-1.0/resolve/main/diffusion_pytorch_model.safetensors", comfyui_folder/"models/controlnet","DepthMap_XL.safetensors")
if show_yes_no_dialog("warning!","Do you want to install all animation models?"):
(comfyui_folder/"models/animatediff_models").mkdir(parents=True, exist_ok=True)
download_file("https://huggingface.co/guoyww/animatediff/resolve/cd71ae134a27ec6008b968d6419952b0c0494cf2/mm_sdxl_v10_beta.ckpt", comfyui_folder/"models/animatediff_models","mm_sdxl_v10_beta.ckpt")
# TODO: fix
# create_conda_env("comfyui","3.10")
# if self.app.config.hardware_mode in ["nvidia", "nvidia-tensorcores"]:
# run_python_script_in_env("comfyui", "-m pip install --pre torch torchvision torchaudio --index-url https://download.pytorch.org/whl/nightly/cu121")
# if self.app.config.hardware_mode in ["amd", "amd-noavx"]:
# run_python_script_in_env("comfyui", "-m pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/rocm5.7")
# elif self.app.config.hardware_mode in ["cpu", "cpu-noavx"]:
# run_python_script_in_env("comfyui", "-m pip install --pre torch torchvision torchaudio")
# run_python_script_in_env("comfyui", f"-m pip install -r {comfyui_folder}/requirements.txt")
self.app.comfyui = LollmsComfyUI(self.app)
ASCIIColors.green("Comfyui installed successfully")
self.app.HideBlockingMessage()
def upgrade(self):
root_dir = self.app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
comfyui_folder = shared_folder / "comfyui"
if not comfyui_folder.exists():
self.app.InfoMessage("Comfyui is not installed, install it first")
return
subprocess.run(["git", "pull", str(comfyui_folder)])
subprocess.run(["git", "pull", str(comfyui_folder/"custom_nodes/ComfyUI-Manager")])
subprocess.run(["git", "pull", str(comfyui_folder/"custom_nodes/efficiency-nodes-comfyui")])
ASCIIColors.success("DONE")
def wait_for_service_in_another_thread(self, max_retries=150, show_warning=True):
thread = threading.Thread(target=self.wait_for_service, args=(max_retries, show_warning))

41
lollms/services/tti/dalle/lollms_dalle.py
View File

@ -36,17 +36,30 @@ import os
class LollmsDalle(LollmsTTI):
def __init__(
self,
app:LollmsApplication,
key="",
generation_engine="dall-e-3",# other possibility "dall-e-2"
output_path=None
):
super().__init__(generation_engine,app)
self.key = key or os.getenv('OPENAI_API_KEY')
self.generation_engine = generation_engine
self.output_path = output_path
def __init__(self, app, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
# Check for the OPENAI_KEY environment variable if no API key is provided
api_key = os.getenv("OPENAI_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{"name":"api_key", "type":"str", "value":api_key, "help":"A valid Open AI key to generate text using anthropic api"},
{"name":"generation_engine", "type":"str", "value":"dall-e-3", "options":["dall-e-2","dall-e-3"], "help":"The engine to be used"},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("dall-e", app, service_config, output_folder)
def settings_updated(self):
pass
def paint(
self,
@ -66,11 +79,11 @@ class LollmsDalle(LollmsTTI):
if output_path is None:
output_path = self.output_path
if generation_engine is None:
generation_engine = self.generation_engine
generation_engine = self.service_config.generation_engine
if not PackageManager.check_package_installed("openai"):
PackageManager.install_package("openai")
import openai
openai.api_key = self.key
openai.api_key = self.service_config.api_key
if generation_engine=="dall-e-2":
supported_resolutions = [
[512, 512],
@ -134,7 +147,7 @@ class LollmsDalle(LollmsTTI):
if not PackageManager.check_package_installed("openai"):
PackageManager.install_package("openai")
import openai
openai.api_key = self.key
openai.api_key = self.service_config.api_key
generation_engine="dall-e-2"
supported_resolutions = [
[512, 512],

98
lollms/services/tti/diffusers/lollms_diffusers.py
View File

@ -8,6 +8,7 @@ import sys
from lollms.app import LollmsApplication
from lollms.utilities import PackageManager, check_and_install_torch, find_next_available_filename
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
import sys
import requests
from typing import List, Dict, Any
@ -17,7 +18,6 @@ from lollms.paths import LollmsPaths
from lollms.tti import LollmsTTI
from lollms.utilities import git_pull
from tqdm import tqdm
import threading
import pipmaster as pm
if not pm.is_installed("torch"):
@ -28,6 +28,11 @@ import torch
if not torch.cuda.is_available():
ASCIIColors.yellow("Diffusers: Torch not using cuda. Reinstalling it")
pm.install_multiple(["torch","torchvision","torchaudio"], "https://download.pytorch.org/whl/cu121", force_reinstall=True)
if not pm.is_installed("transformers"):
pm.install("transformers")
if not pm.is_installed("diffusers"):
pm.install("diffusers")
@ -55,19 +60,8 @@ def install_model(lollms_app:LollmsApplication, model_url):
root_dir = lollms_app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
diffusers_folder = shared_folder / "diffusers"
if not PackageManager.check_package_installed("diffusers"):
PackageManager.install_or_update("diffusers")
if not PackageManager.check_package_installed("torch"):
check_and_install_torch(True)
if not PackageManager.check_package_installed("torch"):
PackageManager.install_package("torch torchvision torchaudio", "https://download.pytorch.org/whl/cu121")
if not PackageManager.check_package_installed("transformers"):
PackageManager.install_package("transformers")
if not PackageManager.check_package_installed("diffusers"):
PackageManager.install_package("diffusers")
import torch
from diffusers import PixArtSigmaPipeline
@ -78,16 +72,6 @@ def install_model(lollms_app:LollmsApplication, model_url):
)
def install_diffusers(lollms_app:LollmsApplication):
root_dir = lollms_app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
diffusers_folder = shared_folder / "diffusers"
diffusers_folder.mkdir(exist_ok=True, parents=True)
models_dir = diffusers_folder / "models"
models_dir.mkdir(parents=True, exist_ok=True)
PackageManager.reinstall("diffusers")
PackageManager.reinstall("xformers")
@ -99,19 +83,33 @@ def upgrade_diffusers(lollms_app:LollmsApplication):
class LollmsDiffusers(LollmsTTI):
has_controlnet = False
def __init__(
self,
app:LollmsApplication,
wm = "Artbot",
):
if not PackageManager.check_package_installed("torch"):
PackageManager.install_package("torch torchvision torchaudio", "https://download.pytorch.org/whl/cu121")
def __init__(self, app, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
if not PackageManager.check_package_installed("transformers"):
PackageManager.install_package("transformers")
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{"name":"model", "type":"str", "value":"v2ray/stable-diffusion-3-medium-diffusers", "help":"The model to be used"},
{"name":"wm", "type":"str", "value":"lollms", "help":"The water marking"},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
if not PackageManager.check_package_installed("diffusers"):
PackageManager.install_package("diffusers")
super().__init__("diffusers", app, service_config)
if not pm.is_installed("torch"):
pm.install("torch torchvision torchaudio", "https://download.pytorch.org/whl/cu121")
if not pm.is_installed("transformers"):
pm.install("transformers")
if not pm.is_installed("diffusers"):
pm.install("diffusers")
super().__init__("diffusers",app)
self.ready = False
@ -119,8 +117,6 @@ class LollmsDiffusers(LollmsTTI):
lollms_paths = app.lollms_paths
root_dir = lollms_paths.personal_path
self.wm = wm
shared_folder = root_dir/"shared"
self.diffusers_folder = shared_folder / "diffusers"
self.output_dir = root_dir / "outputs/diffusers"
@ -138,25 +134,21 @@ class LollmsDiffusers(LollmsTTI):
ASCIIColors.red(" ______ ")
ASCIIColors.red(" |______| ")
ASCIIColors.red("")
ASCIIColors.yellow(f"Using model: {app.config.diffusers_model}")
ASCIIColors.yellow(f"Using model: {self.service_config.model}")
import torch
if not PackageManager.check_package_installed("diffusers"):
check_and_install_torch("nvidia" in self.app.config.hardware_mode)
PackageManager.install_or_update("diffusers")
PackageManager.install_or_update("sentencepiece")
PackageManager.install_or_update("accelerate")
try:
if "stable-diffusion-3" in app.config.diffusers_model:
if "stable-diffusion-3" in self.service_config.model:
from diffusers import StableDiffusion3Pipeline # AutoPipelineForImage2Image#PixArtSigmaPipeline
self.tti_model = StableDiffusion3Pipeline.from_pretrained(
app.config.diffusers_model, torch_dtype=torch.float16, cache_dir=self.tti_models_dir,
self.service_config.model, torch_dtype=torch.float16, cache_dir=self.tti_models_dir,
use_safetensors=True,
)
self.iti_model = None
else:
from diffusers import AutoPipelineForText2Image # AutoPipelineForImage2Image#PixArtSigmaPipeline
self.tti_model = AutoPipelineForText2Image.from_pretrained(
app.config.diffusers_model, torch_dtype=torch.float16, cache_dir=self.tti_models_dir,
self.service_config.model, torch_dtype=torch.float16, cache_dir=self.tti_models_dir,
use_safetensors=True,
)
self.iti_model = None
@ -165,7 +157,7 @@ class LollmsDiffusers(LollmsTTI):
# self.tti_model = StableDiffusionPipeline.from_pretrained(
# "CompVis/stable-diffusion-v1-4", torch_dtype=torch.float16, cache_dir=self.tti_models_dir,
# use_safetensors=True,
# ) # app.config.diffusers_model
# ) # self.service_config.model
# Enable memory optimizations.
try:
if app.config.diffusers_offloading_mode=="sequential_cpu_offload":
@ -177,6 +169,20 @@ class LollmsDiffusers(LollmsTTI):
except Exception as ex:
self.tti_model= None
trace_exception(ex)
def install_diffusers(self):
root_dir = self.app.lollms_paths.personal_path
shared_folder = root_dir/"shared"
diffusers_folder = shared_folder / "diffusers"
diffusers_folder.mkdir(exist_ok=True, parents=True)
models_dir = diffusers_folder / "models"
models_dir.mkdir(parents=True, exist_ok=True)
pm.install("diffusers")
pm.install("xformers")
@staticmethod
def verify(app:LollmsApplication):
# Clone repository
@ -295,7 +301,7 @@ class LollmsDiffusers(LollmsTTI):
from diffusers import AutoPipelineForImage2Image
self.iti_model = AutoPipelineForImage2Image.from_pretrained(
self.app.config.diffusers_model, torch_dtype=torch.float16, variant="fp16", use_safetensors=True
self.self.service_config.model, torch_dtype=torch.float16, variant="fp16", use_safetensors=True
)
if sampler_name!="":
sc = self.get_scheduler_by_name(sampler_name)

44
lollms/services/tti/diffusers_client/lollms_diffusers_client.py
View File

@ -8,6 +8,7 @@ import sys
from lollms.app import LollmsApplication
from lollms.utilities import PackageManager, check_and_install_torch, find_next_available_filename, install_cuda, check_torch_version
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
import sys
import requests
from typing import List, Dict, Any
@ -67,21 +68,42 @@ def upgrade_diffusers(lollms_app:LollmsApplication):
class LollmsDiffusersClient(LollmsTTI):
has_controlnet = False
def __init__(
self,
app:LollmsApplication,
wm = "Artbot",
base_url:str="http://localhost:8593"
):
super().__init__("diffusers_client",app)
def __init__(self, app, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{
"name": "base_url",
"type": "str",
"value": "http://127.0.0.1:8188/",
"help": "The base URL for the service. This is the address where the service is hosted (e.g., http://127.0.0.1:8188/)."
},
{
"name": "wm",
"type": "str",
"value": "lollms",
"help": "Watermarking text or identifier to be used in the service."
},
{"name":"model", "type":"str", "value":"v2ray/stable-diffusion-3-medium-diffusers", "help":"The model to be used"},
{"name":"wm", "type":"str", "value":"lollms", "help":"The water marking"},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("diffusers_client", app, service_config)
self.ready = False
# Get the current directory
lollms_paths = app.lollms_paths
root_dir = lollms_paths.personal_path
self.base_url = base_url
self.wm = wm
shared_folder = root_dir/"shared"
self.diffusers_folder = shared_folder / "diffusers"
@ -137,7 +159,7 @@ class LollmsDiffusersClient(LollmsTTI):
restore_faces=True,
output_path=None
):
url = f"{self.base_url}/generate-image"
url = f"{self.service_config.base_url}/generate-image"
payload = {
"positive_prompt": positive_prompt,

42
lollms/services/tti/fooocus/lollms_fooocus.py
View File

@ -34,6 +34,7 @@ from tqdm import tqdm
import threading
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
def download_file(url, folder_path, local_filename):
# Make sure 'folder_path' exists
@ -67,21 +68,42 @@ def upgrade_fooocus(lollms_app:LollmsApplication):
class LollmsFooocus(LollmsTTI):
def __init__(
self,
app:LollmsApplication,
wm = "Artbot",
base_url="localhost:1024"
):
super().__init__("fooocus",app)
def __init__(self, app, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{
"name": "base_url",
"type": "str",
"value": "localhost:1024",
"help": "The base URL for the service. This is the address where the service is hosted (e.g., http://127.0.0.1:8188/)."
},
{
"name": "wm",
"type": "str",
"value": "lollms",
"help": "Watermarking text or identifier to be used in the service."
},
{"name":"model", "type":"str", "value":"v2ray/stable-diffusion-3-medium-diffusers", "help":"The model to be used"},
{"name":"wm", "type":"str", "value":"lollms", "help":"The water marking"},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("fooocus", app, service_config)
self.ready = False
self.base_url = base_url
# Get the current directory
lollms_paths = app.lollms_paths
root_dir = lollms_paths.personal_path
self.wm = wm
shared_folder = root_dir/"shared"
self.fooocus_folder = shared_folder / "fooocus"
self.output_dir = root_dir / "outputs/fooocus"

74
lollms/services/tti/midjourney/lollms_midjourney.py
View File

@ -65,24 +65,60 @@ def split_image(file_path, folder_path, i):
return split_paths
class LollmsMidjourney(LollmsTTI):
def __init__(
self,
app:LollmsApplication,
key="",
timeout=300,
retries=2,
interval=1,
output_path=None
):
super().__init__("midjourney",app)
self.key = key
self.output_path = output_path
self.timeout = timeout
self.retries = retries
self.interval = interval
def __init__(self, app, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
# Check for the MIDJOURNEY_KEY environment variable if no API key is provided
api_key = os.getenv("MIDJOURNEY_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{
"name": "api_key",
"type": "str",
"value": api_key,
"help": "A valid API key for Midjourney, used to access the text generation service via the Anthropic API."
},
{
"name": "timeout",
"type": "int",
"value": 300,
"help": "The maximum time (in seconds) to wait for a response from the API before timing out."
},
{
"name": "retries",
"type": "int",
"value": 2,
"help": "The number of times to retry the request if it fails or times out."
},
{
"name": "interval",
"type": "int",
"value": 1,
"help": "The time interval (in seconds) between retry attempts."
}
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("midjourney", app, service_config)
self.output_folder = output_folder
self.session = requests.Session()
self.headers = {
"Authorization": f"Bearer {key}",
"Authorization": f"Bearer {self.service_config.api_key}",
"Content-Type": "application/json"
}
def settings_updated(self):
self.session = requests.Session()
self.headers = {
"Authorization": f"Bearer {self.service_config.api_key}",
"Content-Type": "application/json"
}
@ -259,13 +295,13 @@ class LollmsMidjourney(LollmsTTI):
try:
# Send prompt and get initial response
positive_prompt += self.get_nearest_aspect_ratio(width, height)
initial_response = self.send_prompt_with_retry(positive_prompt, self.retries)
initial_response = self.send_prompt_with_retry(positive_prompt, self.service_config.retries)
message_id = initial_response.get("messageId")
if not message_id:
raise ValueError("No messageId returned from initial prompt")
# Poll progress until image generation is done
progress_response = self.poll_progress(message_id, self.timeout, self.interval)
progress_response = self.poll_progress(message_id, self.service_config.timeout, self.service_config.interval)
if "error" in progress_response:
raise ValueError(progress_response["error"])
@ -281,7 +317,7 @@ class LollmsMidjourney(LollmsTTI):
raise ValueError("No messageId returned from initial prompt")
# Poll progress until image generation is done
progress_response = self.poll_progress(message_id, self.timeout, self.interval)
progress_response = self.poll_progress(message_id, self.service_config.timeout, self.service_config.interval)
if "error" in progress_response:
raise ValueError(progress_response["error"])

139
lollms/services/tti/sd/lollms_sd.py
View File

@ -243,38 +243,97 @@ class ControlNetUnit:
class LollmsSD(LollmsTTI):
has_controlnet = False
def __init__(
self,
app:LollmsApplication,
wm = "Artbot",
max_retries=50,
sampler="Euler a",
steps=20,
use_https=False,
username=None,
password=None,
auto_sd_base_url=None,
share=False,
wait_for_service=True
):
super().__init__("stable_diffusion",app)
if auto_sd_base_url=="" or auto_sd_base_url=="http://127.0.0.1:7860":
auto_sd_base_url = None
def __init__(self, app:LollmsApplication, output_folder:str|Path=None):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{
"name": "base_url",
"type": "str",
"value": "http://127.0.0.1:7860",
"help": "Watermarking text or identifier to be used in the generated content."
},
{
"name": "wm",
"type": "str",
"value": "Artbot",
"help": "Watermarking text or identifier to be used in the generated content."
},
{
"name": "max_retries",
"type": "int",
"value": 50,
"help": "The maximum number of retries to attempt before determining that the service is unavailable."
},
{
"name": "sampler",
"type": "str",
"value": "Euler a",
"help": "The sampler algorithm to use for image generation. Example: 'Euler a'."
},
{
"name": "steps",
"type": "int",
"value": 20,
"help": "The number of steps to use in the generation process. Higher values may improve quality but increase processing time."
},
{
"name": "use_https",
"type": "bool",
"value": False,
"help": "If set to true, the service will use HTTPS for secure communication."
},
{
"name": "username",
"type": "str",
"value": None,
"help": "The username for authentication, if required by the service."
},
{
"name": "password",
"type": "str",
"value": None,
"help": "The password for authentication, if required by the service."
},
{
"name": "auto_sd_base_url",
"type": "str",
"value": None,
"help": "The base URL for the Auto SD service. If not provided, a default or local URL will be used."
},
{
"name": "share",
"type": "bool",
"value": False,
"help": "If set to true, the server will be accessible from outside your local machine (e.g., over the internet)."
}
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("stable_diffusion", app, service_config, output_folder)
self.ready = False
# Get the current directory
lollms_paths = app.lollms_paths
root_dir = lollms_paths.personal_path
self.wm = wm
# Store the path to the script
if auto_sd_base_url is None:
self.auto_sd_base_url = "http://127.0.0.1:7860"
if service_config.auto_sd_base_url is None:
self.service_config.base_url = "http://127.0.0.1:7860"
if not LollmsSD.verify(app):
install_sd(app.lollms_paths)
else:
self.auto_sd_base_url = auto_sd_base_url
self.auto_sd_url = self.auto_sd_base_url+"/sdapi/v1"
self.auto_sd_url = self.service_config.base_url+"/sdapi/v1"
shared_folder = root_dir/"shared"
self.sd_folder = shared_folder / "auto_sd"
self.output_dir = root_dir / "outputs/sd"
@ -291,43 +350,43 @@ class LollmsSD(LollmsTTI):
ASCIIColors.red(" Forked from Auto1111's Stable diffusion api")
ASCIIColors.red(" Integration in lollms by ParisNeo using mix1009's sdwebuiapi ")
if not self.wait_for_service(1,False) and auto_sd_base_url is None:
if not self.wait_for_service(1,False) and service_config.auto_sd_base_url is None:
ASCIIColors.info("Loading lollms_sd")
os.environ['SD_WEBUI_RESTARTING'] = '1' # To forbid sd webui from showing on the browser automatically
# Launch the Flask service using the appropriate script for the platform
if platform.system() == "Windows":
ASCIIColors.info("Running on windows")
script_path = self.sd_folder / "lollms_sd.bat"
if share:
run_script_in_env("autosd", str(script_path) +" --share", cwd=self.sd_folder)
if self.service_config.share:
pass # TODO : implement
# run_script_in_env("autosd", str(script_path) +" --share", cwd=self.sd_folder)
else:
run_script_in_env("autosd", str(script_path), cwd=self.sd_folder)
pass # TODO : implement
# run_script_in_env("autosd", str(script_path), cwd=self.sd_folder)
else:
ASCIIColors.info("Running on linux/MacOs")
script_path = str(self.sd_folder / "lollms_sd.sh")
ASCIIColors.info(f"launcher path: {script_path}")
ASCIIColors.info(f"sd path: {self.sd_folder}")
if share:
run_script_in_env("autosd","bash " + script_path +" --share", cwd=self.sd_folder)
if self.service_config.share:
pass # TODO: implement
# run_script_in_env("autosd","bash " + script_path +" --share", cwd=self.sd_folder)
else:
run_script_in_env("autosd","bash " + script_path, cwd=self.sd_folder)
pass # TODO: implement
# run_script_in_env("autosd","bash " + script_path, cwd=self.sd_folder)
ASCIIColors.info("Process done")
ASCIIColors.success("Launching Auto1111's SD succeeded")
# Wait until the service is available at http://127.0.0.1:7860/
if wait_for_service:
self.wait_for_service(max_retries=max_retries)
else:
self.wait_for_service_in_another_thread(max_retries=max_retries)
self.default_sampler = sampler
self.default_steps = steps
self.default_sampler = self.service_config.sampler
self.default_steps = self.service_config.steps
self.session = requests.Session()
if username and password:
self.set_auth(username, password)
if self.service_config.username and self.service_config.password:
self.set_auth(self.service_config.username, self.service_config.password)
else:
self.check_controlnet()
@ -1105,7 +1164,7 @@ class LollmsSD(LollmsTTI):
return thread
def wait_for_service(self, max_retries = 50, show_warning=True):
url = f"{self.auto_sd_base_url}/internal/ping"
url = f"{self.service_config.base_url}/internal/ping"
# Adjust this value as needed
retries = 0

116
lollms/services/tts/eleven_labs_tts/lollms_eleven_labs_tts.py
View File

@ -7,6 +7,7 @@
from pathlib import Path
from lollms.app import LollmsApplication
from lollms.paths import LollmsPaths
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
import sys
import requests
from typing import List, Dict, Any
@ -15,14 +16,15 @@ from ascii_colors import ASCIIColors, trace_exception
from lollms.paths import LollmsPaths
from lollms.utilities import PackageManager, find_next_available_filename
from lollms.tts import LollmsTTS
if not PackageManager.check_package_installed("sounddevice"):
PackageManager.install_package("sounddevice")
if not PackageManager.check_package_installed("soundfile"):
PackageManager.install_package("soundfile")
import pipmaster as pm
if not pm.is_installed("sounddevice"):
pm.install("sounddevice")
if not pm.is_installed("soundfile"):
pm.install("soundfile")
import sounddevice as sd
import soundfile as sf
import os
def get_Whisper(lollms_paths:LollmsPaths):
return LollmsElevenLabsTTS
@ -31,34 +33,82 @@ class LollmsElevenLabsTTS(LollmsTTS):
def __init__(
self,
app: LollmsApplication,
model_id: str = "eleven_turbo_v2_5",
voice_name: str = "Sarah",
api_key: str = "",
output_path: Path | str = None,
stability: float = 0.5,
similarity_boost: float = 0.5,
streaming: bool = False
output_folder: Path | str = None,
):
super().__init__("elevenlabs_tts", app, model_id, voice_name, api_key, output_path)
self.voice_name = voice_name
self.model_id = model_id
self.api_key = api_key
self.output_path = output_path
self.stability = stability
self.similarity_boost = similarity_boost
self.streaming = streaming
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
# Check for the ELEVENLABS_KEY environment variable if no API key is provided
api_key = os.getenv("ELEVENLABS_KEY","")
service_config = TypedConfig(
ConfigTemplate(
[
{
"name": "model_id",
"type": "str",
"value": "eleven_turbo_v2_5",
"options": ["eleven_turbo_v2_5","eleven_flash_v2","eleven_multilingual_v2","eleven_multilingual_v1","eleven_english_sts_v2","eleven_english_sts_v1"],
"help": "The ID of the model to use for text-to-speech generation. Example: 'eleven_turbo_v2_5'."
},
{
"name": "voice_name",
"type": "str",
"value": "Sarah",
"help": "The name of the voice to use for text-to-speech generation. Example: 'Sarah'."
},
{
"name": "language",
"type": "str",
"value": "en",
"options": ["en", "ja", "zh", "de", "hi", "fr", "ko", "pt", "it", "es", "id", "nl", "tr", "fil", "pl", "sv", "bg", "ro", "ar", "cs", "el", "fi", "hr", "ms", "sk", "da", "ta", "uk", "ru", "hu", "no", "vi"], # Dynamically populated based on the selected model_id
"help": "The language to use for text-to-speech generation. Supported languages depend on the selected model."
},
{
"name": "api_key",
"type": "str",
"value": api_key,
"help": "A valid API key for accessing the Eleven Labs service."
},
{
"name": "similarity_boost",
"type": "bool",
"value": False,
"help": "If enabled, increases the similarity of the generated speech to the selected voice."
},
{
"name": "streaming",
"type": "bool",
"value": False,
"help": "If enabled, the text-to-speech output will be streamed in real-time instead of being generated all at once."
}
]
),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("elevenlabs_tts", app, service_config, output_folder)
self.ready = True
self.voices = []
self.voice_id_map = {}
try:
self._fetch_voices()
self.voice_id = self._get_voice_id(voice_name)
self.voice_id = self._get_voice_id(service_config.voice_name)
except:
pass
def settings_updated(self):
pass
def _fetch_voices(self):
url = "https://api.elevenlabs.io/v1/voices"
headers = {"xi-api-key": self.api_key}
headers = {"xi-api-key": self.service_config.api_key}
try:
response = requests.get(url, headers=headers)
@ -82,7 +132,7 @@ class LollmsElevenLabsTTS(LollmsTTS):
def set_voice(self, voice_name: str):
if voice_name in self.voices:
self.voice_name = voice_name
self.service_config.voice_name = voice_name
self.voice_id = self._get_voice_id(voice_name)
else:
raise ValueError(f"Voice '{voice_name}' not found. Available voices: {', '.join(self.voices)}")
@ -94,18 +144,18 @@ class LollmsElevenLabsTTS(LollmsTTS):
payload = {
"text": text,
"language_code": language,
"model_id": self.model_id,
"model_id": self.service_config.model_id,
"voice_settings": {
"stability": self.stability,
"similarity_boost": self.similarity_boost
"stability": self.service_config.stability,
"similarity_boost": self.service_config.similarity_boost
}
}
headers = {
"xi-api-key": self.api_key,
"xi-api-key": self.service_config.api_key,
"Content-Type": "application/json"
}
if self.streaming:
if self.service_config.streaming:
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self.voice_id}/stream"
response = requests.post(url, json=payload, headers=headers)
# Handle streaming response if needed
@ -126,18 +176,18 @@ class LollmsElevenLabsTTS(LollmsTTS):
payload = {
"text": text,
"language_code": language,
"model_id": self.model_id,
"model_id": self.service_config.model_id,
"voice_settings": {
"stability": self.stability,
"similarity_boost": self.similarity_boost
"stability": self.service_config.stability,
"similarity_boost": self.service_config.similarity_boost
}
}
headers = {
"xi-api-key": self.api_key,
"xi-api-key": self.service_config.api_key,
"Content-Type": "application/json"
}
if self.streaming:
if self.service_config.streaming:
url = f"https://api.elevenlabs.io/v1/text-to-speech/{self.voice_id}/stream"
response = requests.post(url, json=payload, headers=headers)
# Handle streaming response if needed

55
lollms/services/tts/fish/lollms_fish_tts.py
View File

@ -6,14 +6,17 @@ from lollms.app import LollmsApplication
from lollms.paths import LollmsPaths
from lollms.tts import LollmsTTS
from lollms.utilities import PackageManager, find_next_available_filename
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
import pipmaster as pm
import os
if not PackageManager.check_package_installed("sounddevice"):
PackageManager.install_package("sounddevice")
if not PackageManager.check_package_installed("soundfile"):
PackageManager.install_package("soundfile")
if not pm.is_installed("sounddevice"):
pm.install("sounddevice")
if not pm.is_installed("soundfile"):
pm.install("soundfile")
if not PackageManager.check_package_installed("ormsgpack"):
PackageManager.install_package("ormsgpack")
if not pm.is_installed("ormsgpack"):
pm.install("ormsgpack")
import ormsgpack
@ -39,15 +42,35 @@ def get_FishAudioTTS(lollms_paths: LollmsPaths):
class LollmsFishAudioTTS(LollmsTTS):
def __init__(
self,
app: LollmsApplication,
voice_name: str = "default",
api_key: str = "",
output_path: Path | str = None,
reference_folder: Path | str = None
):
super().__init__("fishaudio_tts", app, "default", voice_name, api_key, output_path)
self.reference_folder = Path(reference_folder) if reference_folder else None
self,
app: LollmsApplication,
output_folder: Path | str = None,
):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
# Check for the FISHTTS_KEY environment variable if no API key is provided
api_key = os.getenv("FISHTTS_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{"name":"voice_name", "type":"str", "value":"default", "help":"A valid model id"},
{"name":"api_key", "type":"str", "value":api_key, "help":"A valid eleven labs key"},
{"name":"similarity_boost", "type":"bool", "value":False, "help":"A valid model id"},
{"name":"streaming", "type":"bool", "value":False, "help":"A valid model id"},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("fishaudio_tts", app, service_config, output_folder)
self.output_folder = output_folder
self.reference_folder = app.lollms_paths.custom_voices_path/"fish_tts"
self.reference_folder.mkdir(exist_ok=True, parents=True)
self.voices = self._load_voices()
self.ready = True
@ -97,7 +120,7 @@ class LollmsFishAudioTTS(LollmsTTS):
"https://api.fish.audio/v1/tts",
content=ormsgpack.packb(request, option=ormsgpack.OPT_SERIALIZE_PYDANTIC),
headers={
"authorization": f"Bearer {self.api_key}",
"authorization": f"Bearer {self.service_config.api_key}",
"content-type": "application/msgpack",
},
timeout=None,

80
lollms/services/tts/open_ai_tts/lollms_openai_tts.py
View File

@ -38,10 +38,10 @@ if not pm.is_installed("openai"):
pm.install("openai")
from openai import OpenAI
if not PackageManager.check_package_installed("sounddevice"):
PackageManager.install_package("sounddevice")
if not PackageManager.check_package_installed("soundfile"):
PackageManager.install_package("soundfile")
if not pm.is_installed("sounddevice"):
pm.install("sounddevice")
if not pm.is_installed("soundfile"):
pm.install("soundfile")
import sounddevice as sd
import soundfile as sf
@ -53,34 +53,58 @@ class LollmsOpenAITTS(LollmsTTS):
def __init__(
self,
app:LollmsApplication,
model ="tts-1",
voice="alloy",
api_key="",
output_path=None
output_folder=None
):
super().__init__("openai_tts", app, model, voice, api_key, output_path)
self.client = OpenAI(api_key=api_key)
self.voices = [
"alloy",
"echo",
"fable",
"nova",
"shimmer"
]
self.models = [
"tts-1"
]
"""
Initializes the LollmsDalle binding.
self.voice = voice
self.output_path = output_path
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
# Check for the OPENAI_KEY environment variable if no API key is provided
api_key = os.getenv("OPENAI_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{
"name": "model",
"type": "str",
"value": "tts-1",
"options": ["alloy", "echo", "fable", "nova", "shimmer"],
"help": "The model to use for text-to-speech. Options: 'alloy', 'echo', 'fable', 'nova', 'shimmer'."
},
{
"name": "voice",
"type": "str",
"value": "alloy",
"help": "The voice to use for text-to-speech. Options: 'alloy', 'echo', 'fable', 'nova', 'shimmer'."
},
{
"name": "api_key",
"type": "str",
"value": api_key,
"help": "A valid API key for accessing the text-to-speech service."
},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("openai_tts", app, service_config, output_folder)
self.client = OpenAI(api_key=self.service_config.api_key)
self.output_folder = output_folder
self.ready = True
def settings_updated(self):
self.client = OpenAI(api_key=self.service_config.api_key)
def tts_file(self, text, speaker, file_name_or_path, language="en"):
speech_file_path = file_name_or_path
response = self.client.audio.speech.create(
model=self.model,
voice=self.voice,
model=self.service_config.model,
voice=self.service_config.voice,
input=text,
response_format="wav"
@ -91,8 +115,8 @@ class LollmsOpenAITTS(LollmsTTS):
def tts_audio(self, text, speaker:str=None, file_name_or_path:Path|str=None, language="en", use_threading=False):
speech_file_path = file_name_or_path
response = self.client.audio.speech.create(
model=self.model,
voice=self.voice if speaker is None else speaker,
model=self.service_config.model,
voice=self.service_config.voice if speaker is None else speaker,
input=text,
response_format="wav"
@ -114,8 +138,8 @@ class LollmsOpenAITTS(LollmsTTS):
speech_file_path = file_name_or_path
text = self.clean_text(text)
response = self.client.audio.speech.create(
model=self.model,
voice=self.voice,
model=self.service_config.model,
voice=self.service_config.voice,
input=text,
response_format="wav"

58
lollms/services/tts/xtts/lollms_xtts.py
View File

@ -27,14 +27,14 @@ import numpy as np
# Ensure required packages are installed
if not PackageManager.check_package_installed("TTS"):
PackageManager.install_or_update("TTS")
if not pm.is_installed("TTS"):
pm.install("TTS")
if not PackageManager.check_package_installed("simpleaudio"):
PackageManager.install_or_update("simpleaudio")
if not pm.is_installed("simpleaudio"):
pm.install("simpleaudio")
if not PackageManager.check_package_installed("wave"):
PackageManager.install_or_update("wave")
if not pm.is_installed("wave"):
pm.install("wave")
import re
from pathlib import Path
@ -48,6 +48,7 @@ import time
from queue import Queue
import re
import pipmaster as pm
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
# List of common sampling rates
common_sampling_rates = [8000, 11025, 16000, 22050, 32000, 44100, 48000, 96000, 192000]
@ -60,11 +61,54 @@ def xtts_install():
pm.install_or_update("tts", force_reinstall=True)
class LollmsXTTS(LollmsTTS):
def __init__(
self,
app:LollmsApplication,
output_folder=None
):
"""
Initializes the LollmsDalle binding.
Args:
api_key (str): The API key for authentication.
output_folder (Path|str): The output folder where to put the generated data
"""
service_config = TypedConfig(
ConfigTemplate([
{
"name": "model",
"type": "str",
"value": "",
"options": [],
"help": "The model to use for text-to-speech. Options: 'alloy', 'echo', 'fable', 'nova', 'shimmer'."
},
{
"name": "voice",
"type": "str",
"value": "alloy",
"help": "The voice to use for text-to-speech. Options: 'alloy', 'echo', 'fable', 'nova', 'shimmer'."
},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("lollms_xtts", app, service_config, output_folder)
voices_folder = app.lollms_paths.custom_voices_path/"xtts"
voices_folder.mkdir(exist_ok=True, parents=True)
self.voices_folders = [voices_folder] + [Path(__file__).parent/"voices"]
voices = self.get_voices()
service_config.config_template["model"]["options"]=voices
def settings_updated(self):
voices = self.get_voices()
self.service_config.config_template["model"]["options"]=voices
def __init__(self, app: LollmsApplication, voices_folders: List[str|Path], freq = 22050):
super().__init__("lollms_xtts", app)
self.freq = freq
self.generation_threads = {}
self.voices_folders = [Path(v) for v in voices_folders] + [Path(__file__).parent/"voices"]
self.stop_event = threading.Event()
# Show a cool LOGO using ASCIIColors

39
lollms/services/ttv/lumalabs/lollms_lumalabs.py
View File

@ -1,14 +1,44 @@
import requests
from typing import Optional, Dict
from lollms.ttv import LollmsTTV
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
import os
from pathlib import Path
class LollmsLumaLabs(LollmsTTV):
def __init__(self, api_key: str):
self.api_key = api_key
def __init__(self, app, output_folder:str|Path=None):
"""
Initializes the NovitaAITextToVideo binding.
Args:
api_key (str): The API key for authentication.
base_url (str): The base URL for the Novita.ai API. Defaults to "https://api.novita.ai/v3/async".
"""
# Check for the LUMALABS_KEY environment variable if no API key is provided
api_key = os.getenv("LUMALABS_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{"name":"api_key", "type":"str", "value":api_key, "help":"A valid Lumalabs AI key to generate text using anthropic api"},
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("lumalabs", app, service_config, output_folder)
self.base_url = "https://api.lumalabs.ai/dream-machine/v1/generations"
self.headers = {
"accept": "application/json",
"authorization": f"Bearer {self.api_key}",
"authorization": f"Bearer {self.service_config.api_key}",
"content-type": "application/json"
}
def settings_updated(self):
self.base_url = "https://api.lumalabs.ai/dream-machine/v1/generations"
self.headers = {
"accept": "application/json",
"authorization": f"Bearer {self.service_config.api_key}",
"content-type": "application/json"
}
@ -40,6 +70,9 @@ class LollmsLumaLabs(LollmsTTV):
return output_path
def generate_video_by_frames(self, prompts, frames, negative_prompt, fps = 8, num_inference_steps = 50, guidance_scale = 6, seed = None):
pass # TODO : implement
def extend_video(self, prompt: str, generation_id: str, reverse: bool = False) -> str:
keyframes = {
"frame0" if not reverse else "frame1": {

125
lollms/services/ttv/novita_ai/lollms_novita_ai.py
View File

@ -1,16 +1,24 @@
from pathlib import Path
from typing import List, Optional, Dict, Any
from lollms.ttv import LollmsTTV
from lollms.app import LollmsApplication
from lollms.config import TypedConfig, ConfigTemplate, BaseConfig
from lollms.utilities import find_next_available_filename
import requests
import json
import os
import time
class LollmsNovitaAITextToVideo(LollmsTTV):
"""
A binding for the Novita.ai Text-to-Video API.
This class allows generating videos from text prompts using the Novita.ai service.
"""
def __init__(self, api_key: str, base_url: str = "https://api.novita.ai/v3/async"):
def __init__(
self,
app:LollmsApplication,
output_folder:str|Path
):
"""
Initializes the NovitaAITextToVideo binding.
@ -18,29 +26,51 @@ class LollmsNovitaAITextToVideo(LollmsTTV):
api_key (str): The API key for authentication.
base_url (str): The base URL for the Novita.ai API. Defaults to "https://api.novita.ai/v3/async".
"""
super().__init__("novita_ai")
if api_key is None:
# Check for the NOVITA_AI_KEY environment variable if no API key is provided
api_key = os.getenv("NOVITA_AI_KEY","")
if api_key is None:
raise ValueError("No API key provided and NOVITA_AI_KEY environment variable is not set.")
self.api_key = api_key
self.base_url = base_url
# Check for the NOVITA_AI_KEY environment variable if no API key is provided
api_key = os.getenv("NOVITA_AI_KEY","")
service_config = TypedConfig(
ConfigTemplate([
{"name":"api_key", "type":"str", "value":api_key, "help":"A valid Novita AI key to generate text using anthropic api"},
{"name":"model_name","type":"str","value":"darkSushiMixMix_225D_64380.safetensors", "options": ["darkSushiMixMix_225D_64380.safetensors"], "help":"The model name"}
]),
BaseConfig(config={
"api_key": "", # use avx2
})
)
super().__init__("novita_ai", app, service_config,output_folder)
self.model_name = self.service_config.model_name
self.base_url = "https://api.novita.ai/v3/async"
def getModels(self):
"""
Gets the list of models
"""
url = "https://api.novita.ai/v3/model"
headers = {
"Content-Type": "<content-type>",
"Authorization": "<authorization>"
}
response = requests.request("GET", url, headers=headers)
return response.json()["models"]
def generate_video(
self,
prompt: str,
negative_prompt: Optional[str] = None,
model_name: str = "darkSushiMixMix_225D_64380.safetensors",
model_name: str = "",
height: int = 512,
width: int = 512,
steps: int = 20,
seed: int = -1,
nb_frames: int = 64,
guidance_scale: Optional[float] = None,
loras: Optional[List[Dict[str, Any]]] = None,
embeddings: Optional[List[Dict[str, Any]]] = None,
closed_loop: Optional[bool] = None,
clip_skip: Optional[int] = None,
output_dir:str | Path =None,
) -> str:
"""
Generates a video from text prompts using the Novita.ai API.
@ -62,32 +92,91 @@ class LollmsNovitaAITextToVideo(LollmsTTV):
Returns:
str: The task_id for retrieving the generated video.
"""
if model_name=="":
model_name = self.model_name
url = f"{self.base_url}/txt2video"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Authorization": f"Bearer {self.service_config.api_key}",
"Content-Type": "application/json",
}
payload = {
"extra": {
"response_video_type": "mp4", # gif
"enterprise_plan": {"enabled": False}
},
"model_name": model_name,
"height": height,
"width": width,
"steps": steps,
"prompts": [prompt],
"prompts": [
{
"frames": nb_frames,
"prompt": prompt
}
],
"negative_prompt": negative_prompt,
"seed": seed,
"guidance_scale": guidance_scale,
"seed": seed,
"loras": loras,
"embeddings": embeddings,
"closed_loop": closed_loop,
"clip_skip": clip_skip,
}
"clip_skip": clip_skip
}
# Remove None values from the payload to avoid sending null fields
payload = {k: v for k, v in payload.items() if v is not None}
response = requests.post(url, headers=headers, data=json.dumps(payload))
response.raise_for_status() # Raise an exception for HTTP errors
task_id = response.json().get("task_id")
return response.json().get("task_id")
url = f"https://api.novita.ai/v3/async/task-result?task_id={task_id}"
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {self.service_config.api_key}",
}
done = False
while not done:
response = requests.request("GET", url, headers=headers)
infos = response.json()
if infos["task"]["status"]=="TASK_STATUS_SUCCEED" or infos["task"]["status"]=="TASK_STATUS_FAILED":
done = True
time.sleep(1)
if infos["task"]["status"]=="TASK_STATUS_SUCCEED":
if output_dir:
output_dir = Path(output_dir)
file_name = output_dir/find_next_available_filename(output_dir, "vid_novita_") # You can change the filename if needed
self.download_video(infos["videos"][0]["video_url"], file_name )
else:
file_name = self.app.personality.personality_output_folder/find_next_available_filename(output_dir, "vid_novita_") # You can change the filename if needed
self.download_video(infos["videos"][0]["video_url"], self.app.personality.personality_output_folder/"video_{}.mp4" )
return infos
def settings_updated(self):
pass
def generate_video_by_frames(self, prompts: List[str], frames: List[int], negative_prompt: str, fps: int = 8,
num_inference_steps: int = 50, guidance_scale: float = 6.0,
seed: Optional[int] = None) -> str:
"""
Generates a video from a list of prompts and corresponding frames.
Args:
prompts (List[str]): List of text prompts for each frame.
frames (List[int]): List of frame indices corresponding to each prompt.
negative_prompt (str): Text describing elements to avoid in the video.
fps (int): Frames per second. Default is 8.
num_inference_steps (int): Number of steps for the model to infer. Default is 50.
guidance_scale (float): Controls how closely the model adheres to the prompt. Default is 6.0.
seed (Optional[int]): Random seed for reproducibility. Default is None.
Returns:
str: The path to the generated video.
"""
pass
def get_task_result(self, task_id: str) -> Dict[str, Any]:
"""
@ -101,13 +190,13 @@ class LollmsNovitaAITextToVideo(LollmsTTV):
"""
url = f"{self.base_url}/task-result"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Authorization": f"Bearer {self.service_config.api_key}",
}
params = {
"task_id": task_id,
}
response = requests.get(url, headers=headers, params=params)
response = requests.get(url, headers=headers, params=params)
response.raise_for_status() # Raise an exception for HTTP errors
return response.json()

38
lollms/stt.py
View File

@ -11,6 +11,9 @@ from lollms.app import LollmsApplication
from lollms.utilities import PackageManager
from pathlib import Path
from ascii_colors import ASCIIColors
from lollms.main_config import LOLLMSConfig
from lollms.config import TypedConfig
from lollms.service import LollmsSERVICE
try:
if not PackageManager.check_package_installed("sounddevice"):
@ -27,38 +30,33 @@ try:
except:
ASCIIColors.error("Couldn't load sound tools")
class LollmsSTT:
class LollmsSTT(LollmsSERVICE):
"""
LollmsSTT is a base class for implementing Speech-to-Text (STT) functionalities within the LollmsApplication.
Attributes:
app (LollmsApplication): The instance of the main Lollms application.
model (str): The STT model to be used for transcription.
output_path (Path or str): Path where the output transcription files will be saved.
LollmsTTI is a base class for implementing Text-to-Image (TTI) functionalities within the LollmsApplication.
"""
def __init__(
self,
name:str,
app: LollmsApplication,
model="",
output_path=None,
models=[]
app: LollmsApplication,
service_config: TypedConfig,
output_folder: str|Path=None
):
"""
Initializes the LollmsSTT class with the given parameters.
Initializes the LollmsTTI class with the given parameters.
Args:
app (LollmsApplication): The instance of the main Lollms application.
model (str, optional): The STT model to be used for transcription. Defaults to an empty string.
output_path (Path or str, optional): Path where the output transcription files will be saved. Defaults to None.
model (str, optional): The TTI model to be used for image generation. Defaults to an empty string.
api_key (str, optional): API key for accessing external TTI services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output image files will be saved. Defaults to None.
"""
self.name = name
self.ready = False
self.app = app
self.output_path = output_path
self.model = model
self.models = models
super().__init__(name, app, service_config)
if output_folder is not None:
self.output_folder = Path(output_folder)
else:
self.output_folder = app.lollms_paths.personal_outputs_path/name
self.output_folder.mkdir(exist_ok=True, parents=True)
def transcribe(
self,

34
lollms/tti.py
View File

@ -10,27 +10,21 @@ Author: ParisNeo, a computer geek passionate about AI
from lollms.app import LollmsApplication
from pathlib import Path
from typing import List, Dict
from lollms.main_config import LOLLMSConfig
from lollms.config import TypedConfig
from lollms.service import LollmsSERVICE
class LollmsTTI:
class LollmsTTI(LollmsSERVICE):
"""
LollmsTTI is a base class for implementing Text-to-Image (TTI) functionalities within the LollmsApplication.
Attributes:
app (LollmsApplication): The instance of the main Lollms application.
model (str): The TTI model to be used for image generation.
api_key (str): API key for accessing external TTI services (if needed).
output_path (Path or str): Path where the output image files will be saved.
voices (List[str]): List of available voices for TTI (to be filled by the child class).
models (List[str]): List of available models for TTI (to be filled by the child class).
"""
def __init__(
self,
name:str,
app: LollmsApplication,
model="",
api_key="",
output_path=None
app: LollmsApplication,
service_config: TypedConfig,
output_folder: str|Path=None
):
"""
Initializes the LollmsTTI class with the given parameters.
@ -41,14 +35,12 @@ class LollmsTTI:
api_key (str, optional): API key for accessing external TTI services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output image files will be saved. Defaults to None.
"""
self.ready = False
self.name = name
self.app = app
self.model = model
self.api_key = api_key
self.output_path = output_path
self.models = [] # To be filled by the child class
super().__init__(name, app, service_config)
if output_folder is not None:
self.output_folder = Path(output_folder)
else:
self.output_folder = app.lollms_paths.personal_outputs_path/name
self.output_folder.mkdir(exist_ok=True, parents=True)
def paint(self,
positive_prompt: str,
negative_prompt: str = "",

41
lollms/ttm.py
View File

@ -10,44 +10,39 @@ Author: ParisNeo, a computer geek passionate about AI
from lollms.app import LollmsApplication
from pathlib import Path
from typing import List, Dict
from lollms.main_config import LOLLMSConfig
from lollms.config import TypedConfig
from lollms.service import LollmsSERVICE
class LollmsTTM:
class LollmsTTM(LollmsSERVICE):
"""
LollmsTTM is a base class for implementing Text-to-Music (TTM) functionalities within the LollmsApplication.
Attributes:
app (LollmsApplication): The instance of the main Lollms application.
model (str): The TTM model to be used for image generation.
api_key (str): API key for accessing external TTM services (if needed).
output_path (Path or str): Path where the output image files will be saved.
voices (List[str]): List of available voices for TTM (to be filled by the child class).
models (List[str]): List of available models for TTM (to be filled by the child class).
"""
def __init__(
self,
name:str,
app: LollmsApplication,
model="",
api_key="",
output_path=None
app: LollmsApplication,
service_config: TypedConfig,
output_folder: str|Path=None
):
"""
Initializes the LollmsTTM class with the given parameters.
Initializes the LollmsTTI class with the given parameters.
Args:
app (LollmsApplication): The instance of the main Lollms application.
model (str, optional): The TTM model to be used for image generation. Defaults to an empty string.
api_key (str, optional): API key for accessing external TTM services. Defaults to an empty string.
model (str, optional): The TTI model to be used for image generation. Defaults to an empty string.
api_key (str, optional): API key for accessing external TTI services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output image files will be saved. Defaults to None.
"""
self.ready = False
self.name = name
self.app = app
self.model = model
self.api_key = api_key
self.output_path = output_path
self.models = [] # To be filled by the child class
super().__init__(name, app, service_config)
if output_folder is not None:
self.output_folder = Path(output_folder)
else:
self.output_folder = app.lollms_paths.personal_outputs_path/name
self.output_folder.mkdir(exist_ok=True, parents=True)
def generate(self,
positive_prompt: str,

65
lollms/tts.py
View File

@ -10,60 +10,51 @@ from lollms.app import LollmsApplication
from lollms.utilities import PackageManager
from pathlib import Path
from ascii_colors import ASCIIColors
from lollms.main_config import LOLLMSConfig
from lollms.config import TypedConfig
import re
try:
if not PackageManager.check_package_installed("sounddevice"):
# os.system("sudo apt-get install portaudio19-dev")
PackageManager.install_package("sounddevice")
PackageManager.install_package("wave")
except:
# os.system("sudo apt-get install portaudio19-dev -y")
PackageManager.install_package("sounddevice")
PackageManager.install_package("wave")
from lollms.service import LollmsSERVICE
import pipmaster as pm
if not pm.is_installed("sounddevice"):
# os.system("sudo apt-get install portaudio19-dev")
pm.install("sounddevice")
pm.install("wave")
try:
import sounddevice as sd
import wave
except:
ASCIIColors.error("Couldn't load sound tools")
class LollmsTTS:
class LollmsTTS(LollmsSERVICE):
"""
LollmsTTS is a base class for implementing Text-to-Speech (TTS) functionalities within the LollmsApplication.
Attributes:
app (LollmsApplication): The instance of the main Lollms application.
voice (str): The voice model to be used for TTS.
api_key (str): API key for accessing external TTS services (if needed).
output_path (Path or str): Path where the output audio files will be saved.
LollmsTTI is a base class for implementing Text-to-Image (TTI) functionalities within the LollmsApplication.
"""
def __init__(
self,
self,
name:str,
app: LollmsApplication,
model="",
voice="",
api_key="",
output_path=None
app: LollmsApplication,
service_config: TypedConfig,
output_folder: str|Path=None
):
"""
Initializes the LollmsTTS class with the given parameters.
Initializes the LollmsTTI class with the given parameters.
Args:
app (LollmsApplication): The instance of the main Lollms application.
model (str, optional): The speach generation model to be used for TTS. Defaults to "".
voice (str, optional): The voice model to be used for TTS. Defaults to "alloy".
api_key (str, optional): API key for accessing external TTS services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output audio files will be saved. Defaults to None.
model (str, optional): The TTI model to be used for image generation. Defaults to an empty string.
api_key (str, optional): API key for accessing external TTI services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output image files will be saved. Defaults to None.
"""
self.name = name
self.ready = False
self.app = app
self.model = model
self.voice = voice
self.api_key = api_key
self.output_path = output_path
self.voices = [] # To be filled by the child class
self.models = [] # To be filled by the child class
super().__init__(name, app, service_config)
if output_folder is not None:
self.output_folder = Path(output_folder)
else:
self.output_folder = app.lollms_paths.personal_outputs_path/name
self.output_folder.mkdir(exist_ok=True, parents=True)
def tts_file(self, text, file_name_or_path, speaker=None, language="en")->str:
"""

49
lollms/ttv.py
View File

@ -1,18 +1,47 @@
from abc import ABC, abstractmethod
from abc import abstractmethod
from typing import List, Optional
from lollms.app import LollmsApplication
from lollms.main_config import LOLLMSConfig
from lollms.config import TypedConfig
from lollms.utilities import find_next_available_filename
from lollms.service import LollmsSERVICE
from pathlib import Path
class LollmsTTV(ABC):
class LollmsTTV(LollmsSERVICE):
"""
Abstract base class for text-to-video generation services.
Subclasses must implement the methods to generate videos from text prompts.
LollmsTTI is a base class for implementing Text-to-Image (TTI) functionalities within the LollmsApplication.
"""
def __init__(self, service_name):
self.name = service_name
def __init__(
self,
name:str,
app: LollmsApplication,
service_config: TypedConfig,
output_folder: str|Path=None
):
"""
Initializes the LollmsTTI class with the given parameters.
Args:
app (LollmsApplication): The instance of the main Lollms application.
model (str, optional): The TTI model to be used for image generation. Defaults to an empty string.
api_key (str, optional): API key for accessing external TTI services. Defaults to an empty string.
output_path (Path or str, optional): Path where the output image files will be saved. Defaults to None.
"""
super().__init__(name, app, service_config)
if output_folder is not None:
self.output_folder = Path(output_folder)
else:
self.output_folder = app.lollms_paths.personal_outputs_path/name
self.output_folder.mkdir(exist_ok=True, parents=True)
@abstractmethod
def generate_video(self, prompt: str, negative_prompt: str, num_frames: int = 49, fps: int = 8,
num_inference_steps: int = 50, guidance_scale: float = 6.0,
seed: Optional[int] = None) -> str:
seed: Optional[int] = None,
output_dir:str | Path =None,) -> str:
"""
Generates a video from a single text prompt.
@ -50,3 +79,9 @@ class LollmsTTV(ABC):
str: The path to the generated video.
"""
pass
def getModels(self):
"""
Gets the list of models
"""
return []

56
lollms/utilities.py
View File

@ -463,7 +463,29 @@ def discussion_path_to_url(file_path:str|Path)->str:
url = "/"+file_path[file_path.index("discussion_databases"):].replace("\\","/").replace("discussion_databases","discussions")
return "/".join([urllib.parse.quote(p, safe="") for p in url.split("/")])
def always_get_an_event_loop() -> asyncio.AbstractEventLoop:
"""
Ensure that there is always an event loop available.
This function tries to get the current event loop. If the current event loop is closed or does not exist,
it creates a new event loop and sets it as the current event loop.
Returns:
asyncio.AbstractEventLoop: The current or newly created event loop.
"""
try:
# Try to get the current event loop
current_loop = asyncio.get_event_loop()
if current_loop.is_closed():
raise RuntimeError("Event loop is closed.")
return current_loop
except RuntimeError:
# If no event loop exists or it is closed, create a new one
ASCIIColors.info("Creating a new event loop in main thread.")
new_loop = asyncio.new_event_loop()
asyncio.set_event_loop(new_loop)
return new_loop
def yes_or_no_input(prompt):
while True:
@ -649,31 +671,33 @@ def is_asyncio_loop_running():
except RuntimeError: # This gets raised if there's no running event loop
return False
def run_async(func):
import asyncio
from typing import Callable, Coroutine, Any
def run_async(func: Callable[[], Coroutine[Any, Any, None]]) -> None:
"""
run_async(func) -> None
Utility function to run async functions in sync environment. Takes an async function as input and runs it within an async context.
Utility function to run async functions in a synchronous environment.
Takes an async function as input and runs it within an async context.
Parameters:
func (function): The async function to run.
func (Callable[[], Coroutine[Any, Any, None]]): The async function to run.
Returns:
None: Nothing is returned since the function is meant to perform side effects.
"""
if is_asyncio_loop_running():
# We're in a running event loop, so we can call the function with asyncio.create_task
#task = asyncio.run_coroutine_threadsafe(func(), asyncio.get_event_loop())
#task.result()
loop = asyncio.get_running_loop()
task = loop.create_task(func())
else:
# We're not in a running event loop, so we need to create one and run the function in it
try:
asyncio.run(func())
except:
func()
asyncio.sleep(0)
try:
# Check if an event loop is already running
loop = asyncio.get_event_loop()
except RuntimeError:
# If no event loop is running, create a new one
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
# If the loop is not running, run the coroutine until it completes
loop.run_until_complete(func())
def terminate_thread(thread):
"""