|
|
from aiohttp import web |
|
|
from typing import Optional |
|
|
from folder_paths import folder_names_and_paths, get_directory_by_type |
|
|
from api_server.services.terminal_service import TerminalService |
|
|
import app.logger |
|
|
import os |
|
|
|
|
|
class InternalRoutes: |
|
|
''' |
|
|
The top level web router for internal routes: /internal/* |
|
|
The endpoints here should NOT be depended upon. It is for ComfyUI frontend use only. |
|
|
Check README.md for more information. |
|
|
''' |
|
|
|
|
|
def __init__(self, prompt_server): |
|
|
self.routes: web.RouteTableDef = web.RouteTableDef() |
|
|
self._app: Optional[web.Application] = None |
|
|
self.prompt_server = prompt_server |
|
|
self.terminal_service = TerminalService(prompt_server) |
|
|
|
|
|
def setup_routes(self): |
|
|
@self.routes.get('/logs') |
|
|
async def get_logs(request): |
|
|
return web.json_response("".join([(l["t"] + " - " + l["m"]) for l in app.logger.get_logs()])) |
|
|
|
|
|
@self.routes.get('/logs/raw') |
|
|
async def get_raw_logs(request): |
|
|
self.terminal_service.update_size() |
|
|
return web.json_response({ |
|
|
"entries": list(app.logger.get_logs()), |
|
|
"size": {"cols": self.terminal_service.cols, "rows": self.terminal_service.rows} |
|
|
}) |
|
|
|
|
|
@self.routes.patch('/logs/subscribe') |
|
|
async def subscribe_logs(request): |
|
|
json_data = await request.json() |
|
|
client_id = json_data["clientId"] |
|
|
enabled = json_data["enabled"] |
|
|
if enabled: |
|
|
self.terminal_service.subscribe(client_id) |
|
|
else: |
|
|
self.terminal_service.unsubscribe(client_id) |
|
|
|
|
|
return web.Response(status=200) |
|
|
|
|
|
|
|
|
@self.routes.get('/folder_paths') |
|
|
async def get_folder_paths(request): |
|
|
response = {} |
|
|
for key in folder_names_and_paths: |
|
|
response[key] = folder_names_and_paths[key][0] |
|
|
return web.json_response(response) |
|
|
|
|
|
@self.routes.get('/files/{directory_type}') |
|
|
async def get_files(request: web.Request) -> web.Response: |
|
|
directory_type = request.match_info['directory_type'] |
|
|
if directory_type not in ("output", "input", "temp"): |
|
|
return web.json_response({"error": "Invalid directory type"}, status=400) |
|
|
|
|
|
directory = get_directory_by_type(directory_type) |
|
|
sorted_files = sorted( |
|
|
(entry for entry in os.scandir(directory) if entry.is_file()), |
|
|
key=lambda entry: -entry.stat().st_mtime |
|
|
) |
|
|
return web.json_response([entry.name for entry in sorted_files], status=200) |
|
|
|
|
|
|
|
|
def get_app(self): |
|
|
if self._app is None: |
|
|
self._app = web.Application() |
|
|
self.setup_routes() |
|
|
self._app.add_routes(self.routes) |
|
|
return self._app |
|
|
|
