MesopUI

@contextmanager
def create(self, app: Runnable, import_string: str) -> Iterator[None]:
    logger.info(f"Creating MesopUI with import string: {import_string}")
    MesopUI._app = app
    MesopUI._import_string = import_string

    start_script = """import fastagency.ui.mesop.main"""

    with TemporaryDirectory() as temp_dir:
        main_path = Path(temp_dir) / "main.py"
        with main_path.open("w") as f:
            f.write(start_script)

        MESOP_FLAGS.mark_as_parsed()
        MesopUI._main_path = str(main_path)
        MesopUI._created_instance = self

        yield

def create_subconversation(self) -> "MesopUI":
    sub_conversation = MesopUI(self)
    self.sub_conversations.append(sub_conversation)

    return sub_conversation

@classmethod
def get_conversation(cls, id: str) -> "MesopUI":
    return cls._registry[id]

@classmethod
def get_created_instance(cls) -> "MesopUI":
    created_instance = cls._created_instance
    if created_instance is None:
        raise RuntimeError("MesopUI has not been created yet.")

    return created_instance

def get_message_stream(self) -> Generator[MesopMessage, None, None]:
    while True:
        message = self.out_queue.get()
        if self._is_stream_braker(message.io_message):
            yield message
            break
        yield message

def handle_wsgi(
    self,
    app: "Runnable",
    environ: dict[str, Any],
    start_response: Callable[..., Any],
) -> list[bytes]:
    logger.info(f"Starting MesopUI using WSGI interface with app: {app}")
    MesopUI._created_instance = self
    MesopUI._app = app

    if MesopUI._me is None:
        logger.error("MesopUI._me is None")
        raise RuntimeError("MesopUI._me is None")

    return MesopUI._me(environ, start_response)  # type: ignore[no-any-return]

def process_message(self, message: IOMessage) -> Optional[str]:
    return self.visit(message)

@classmethod
def register(cls, conversation: "MesopUI") -> None:
    cls._registry[conversation.id] = conversation

def respond(self, message: str) -> None:
    self.in_queue.put(message)

@classmethod
def respond_to(
    cls, conversation_id: str, message: str
) -> Generator[MesopMessage, None, None]:
    conversation = cls.get_conversation(conversation_id)
    conversation.respond(message)
    return conversation.get_message_stream()

def start(
    self,
    *,
    app: Runnable,
    import_string: str,
    name: Optional[str] = None,
    initial_message: Optional[str] = None,
    single_run: bool = False,
) -> None:
    logger.info(
        f"Starting MesopUI: import_string={self._import_string}, main_path={self._main_path}"
    )
    if single_run:
        logger.warning("single_run parameter is currently not supported in MesopUI")

    MesopUI._app = app

    mesop_main(["mesop", self._main_path])

@classmethod
def unregister(cls, conversation: "MesopUI") -> None:
    del cls._registry[conversation.id]

def visit(self, message: IOMessage) -> Optional[str]:
    method_name = f"visit_{message.type}"
    method = getattr(self, method_name, self.visit_default)
    return method(message)

def visit_default(self, message: IOMessage) -> None:
    mesop_msg = self._mesop_message(message)
    self._publish(mesop_msg)

def visit_error(self, message: Error) -> Optional[str]:
    return self.visit_default(message)

def visit_function_call_execution(
    self, message: FunctionCallExecution
) -> Optional[str]:
    return self.visit_default(message)

def visit_multiple_choice(self, message: MultipleChoice) -> str:
    mesop_msg = self._mesop_message(message)
    self._publish(mesop_msg)
    return self.in_queue.get()

def visit_suggested_function_call(
    self, message: SuggestedFunctionCall
) -> Optional[str]:
    return self.visit_default(message)

def visit_system_message(self, message: SystemMessage) -> Optional[str]:
    return self.visit_default(message)

def visit_text_input(self, message: TextInput) -> str:
    mesop_msg = self._mesop_message(message)
    self._publish(mesop_msg)
    return self.in_queue.get()

def visit_text_message(self, message: TextMessage) -> None:
    mesop_msg = self._mesop_message(message)
    self._publish(mesop_msg)

def visit_workflow_completed(self, message: WorkflowCompleted) -> Optional[str]:
    return self.visit_default(message)