"""
title: DOOM
author: justinh-rahb
author_url: https://github.com/justinh-rahb/webui-doom
funding_url: https://github.com/justinh-rahb/webui-doom
version: 6.6.6r9
license: MIT
"""
from pydantic import BaseModel, Field
from typing import Union, Generator, Iterator, Dict
from utils.misc import get_last_user_message
from apps.webui.models.files import Files
from apps.webui.models.users import Users
from main import generate_chat_completions
import requests
import time
import uuid
import os
import json
from config import UPLOAD_DIR
class Pipe:
class Valves(BaseModel):
MODEL_ID: str = Field(
default="DOOM:latest", description="Identifier for the DOOM model."
)
MODEL_NAME: str = Field(default="DOOM", description="Name for the DOOM model.")
FALLBACK_MODEL_ID: str = Field(
default="llama3:latest", description="Title-generation model."
)
GITHUB_REPO_URL: str = Field(
default="https://raw.githubusercontent.com/justinh-rahb/webui-doom/main/src/",
description="Base URL to load assets from.",
)
WAD_FILE_URL: str = Field(
default="https://raw.githubusercontent.com/justinh-rahb/webui-doom/main/src/doom1.wad",
description="Bring your own .WAD file!",
)
def __init__(self):
self.type = "manifold"
self.valves = self.Valves()
self.token = None
pass
def pipes(self):
return [{"name": self.valves.MODEL_NAME, "id": self.valves.MODEL_ID}]
def create_file(
self, file_name: str, title: str, content: Union[str, bytes], content_type: str
):
base_path = UPLOAD_DIR + "/"
file_id = str(uuid.uuid4())
file_path = base_path + file_id + "_" + file_name
# Create a file
mode = "w" if isinstance(content, str) else "wb"
with open(file_path, mode) as f:
f.write(content)
meta = {
"source": file_path,
"title": title,
"content_type": content_type,
"size": os.path.getsize(file_path),
"path": file_path,
}
class FileForm(BaseModel):
id: str
filename: str
meta: dict = {}
formData = FileForm(id=file_id, filename=file_name, meta=meta)
file = Files.insert_new_file(self.user_id, formData)
return file.id
def get_file_url(self, file_id: str) -> str:
return f"/api/v1/files/{file_id}/content"
def responses(
self, command: str, messages: list
) -> Union[str, Generator, Iterator]:
print(f"responses:{__name__}")
command_body = get_last_user_message(messages)[len(command) + 2 :]
print("Command Body:", command_body)
list_of_responses = [
"\n\n",
]
# Check if the files already exist
files = Files.get_files()
files = [file for file in files if file.user_id == self.user_id]
existing_html = next(
(file for file in files if "index.html" in file.filename), None
)
if existing_html:
list_of_responses.append(
"Looks like you already have the game files... 🤘\n\n"
)
list_of_responses.append(
"{{HTML_FILE_ID_{FILE_ID}}}".replace("{FILE_ID}", existing_html.id)
)
else:
wad_url = self.valves.WAD_FILE_URL
cfg_url = f"{self.valves.GITHUB_REPO_URL}default.cfg"
html_url = f"{self.valves.GITHUB_REPO_URL}index.html"
js_url = f"{self.valves.GITHUB_REPO_URL}websockets-doom.js"
wasm_url = f"{self.valves.GITHUB_REPO_URL}websockets-doom.wasm"
# Step 1: Download and upload the WAD file as doom1.wad
list_of_responses.append("```console\nDownloading WAD.......... ")
wad_id = self.download_and_create_file(
"doom1.wad", wad_url, "application/x-doom"
)
list_of_responses.append(f"ID: {wad_id} DONE\n")
print(f"doom1.wad ID: {wad_id}")
# Step 2: Download and upload the default.cfg file
list_of_responses.append("Downloading CFG.......... ")
cfg_id = self.download_and_create_file(
"default.cfg", cfg_url, "application/octet-stream"
)
list_of_responses.append(f"ID: {cfg_id} DONE\n")
print(f"default.cfg ID: {cfg_id}")
# Step 3: Download and upload the WASM file
list_of_responses.append("Downloading WASM......... ")
wasm_id = self.download_and_create_file(
"websockets-doom.wasm", wasm_url, "application/wasm"
)
list_of_responses.append(f"ID: {wasm_id} DONE\n")
print(f"websockets-doom.wasm ID: {wasm_id}")
# Step 4: Download, modify, and upload the JavaScript file
list_of_responses.append("Downloading JS........... ")
js_content = requests.get(js_url).text
js_content = js_content.replace(
"websockets-doom.wasm", self.get_file_url(wasm_id)
)
js_content = js_content.replace(
"doom1.wad", self.get_file_url(wad_id)
).replace("default.cfg", self.get_file_url(cfg_id))
js_id = self.create_file(
"websockets-doom.js",
"websockets-doom.js",
js_content,
"application/javascript",
)
list_of_responses.append(f"ID: {js_id} DONE\n")
print(f"websockets-doom.js ID: {js_id}")
# Step 5: Download, modify, and upload the HTML file
list_of_responses.append("Downloading HTML......... ")
html_content = requests.get(html_url).text
html_content = html_content.replace(
'Module.FS.createPreloadedFile("", "doom1.wad", "doom1.wad", true, true);',
f'Module.FS.createPreloadedFile("", "doom1.wad", "{self.get_file_url(wad_id)}/doom1.wad", true, true);',
)
html_content = html_content.replace(
'Module.FS.createPreloadedFile("", "default.cfg", "default.cfg", true, true);',
f'Module.FS.createPreloadedFile("", "default.cfg", "{self.get_file_url(cfg_id)}/default.cfg", true, true);',
)
html_content = html_content.replace(
"websockets-doom.js", self.get_file_url(js_id)
)
html_id = self.create_file(
"index.html", "index.html", html_content, "text/html"
)
list_of_responses.append(f"ID: {html_id} DONE\n```\n\n")
print(f"index.html ID: {html_id}")
# Step 6: Provide the final HTML ID to display in iframe
list_of_responses.append("Time to play... 😈\n\n")
list_of_responses.append(
"{{HTML_FILE_ID_{FILE_ID}}}".replace("{FILE_ID}", html_id)
)
for response in list_of_responses:
time.sleep(0.5)
yield response
return "Done"
def download_and_create_file(self, file_name: str, url: str, content_type: str):
try:
response = requests.get(url)
response.raise_for_status()
content = response.content
# Create the file with the correct content type
return self.create_file(file_name, file_name, content, content_type)
except Exception as e:
raise Exception(f"Error handling {file_name}: {str(e)}")
async def pipe(self, body: dict, __user__: dict) -> Union[str, Generator, Iterator]:
print(f"pipe:{__name__}")
self.user_id = __user__["id"]
messages = body["messages"]
user_message = get_last_user_message(messages)
if user_message.startswith("/"):
command = user_message.split(" ")[0][1:]
print(f"Command: {command}")
return self.responses(command, messages)
else:
print("No command found - calling API")
# Call the API based on the API_TYPE
payload = {
"model": self.valves.FALLBACK_MODEL_ID,
"messages": body["messages"],
"stream": body.get("stream", False),
}
print(f"call_api:{__name__}")
print(f"call_api:{payload}")
user = Users.get_user_by_id(__user__["id"])
return await generate_chat_completions(form_data=payload, user=user)