Console

pamiq_core.console.WebApiServer

WebApiServer(
    system_status: SystemStatusProvider,
    host: str = "localhost",
    port: int = 8391,
    max_queue_size: int = DEFAULT_QUEUE_SIZE,
)

Web API Server for controlling the system.

This class provides a simple Web API for controlling the thread controller, allowing external applications to pause, resume, and shutdown the system.

API Endpoints

• GET /api/status

  • Returns the current system status.
  • Response: {"status": status_name}
  • Status codes: 200 OK, 500 Internal Server Error

• POST /api/pause

  • Pauses the system.
  • Response: {"result": "ok"} or {"error": "Command queue is full, try again later"}
  • Status codes: 200 OK, 503 Service Unavailable

• POST /api/resume

  • Resumes the system.
  • Response: {"result": "ok"} or {"error": "Command queue is full, try again later"}
  • Status codes: 200 OK, 503 Service Unavailable

• POST /api/shutdown

  • Shuts down the system.
  • Response: {"result": "ok"} or {"error": "Command queue is full, try again later"}
  • Status codes: 200 OK, 503 Service Unavailable

• POST /api/save-state:

  • Saves a state of the current system state.
  • Response: {"result": "ok"} or {"error": "Command queue is full, try again later"}
  • Status codes: 200 OK, 503 Service Unavailable

Error Responses

• 404 Not Found: {"error": "Invalid API endpoint"}
• 405 Method Not Allowed: {"error": "Invalid API method"}
• 500 Internal Server Error: {"error": "Internal server error"}

Initialize the WebApiHandler.

PARAMETER	DESCRIPTION
system_status	Provider for system status information. TYPE: SystemStatusProvider
host	Hostname to run the API server on. TYPE: str DEFAULT: 'localhost'
port	Port to run the API server on. TYPE: int DEFAULT: 8391
max_queue_size	Maximum size of the command queue. TYPE: int DEFAULT: DEFAULT_QUEUE_SIZE

Source code in src/pamiq_core/console/web_api.py

def __init__(
    self,
    system_status: SystemStatusProvider,
    host: str = "localhost",
    port: int = 8391,
    max_queue_size: int = DEFAULT_QUEUE_SIZE,
) -> None:
    """Initialize the WebApiHandler.

    Args:
        system_status: Provider for system status information.
        host: Hostname to run the API server on.
        port: Port to run the API server on.
        max_queue_size: Maximum size of the command queue.
    """
    self._logger = logging.getLogger(get_class_module_path(self.__class__))
    self._system_status = system_status
    self._host = host
    self._port = port

    self._received_commands_queue: Queue[ControlCommands] = Queue(
        maxsize=max_queue_size
    )

    # Define routes for the Starlette app
    routes = [
        Route("/api/status", endpoint=self._get_status, methods=["GET"]),
        Route("/api/pause", endpoint=self._post_pause, methods=["POST"]),
        Route("/api/resume", endpoint=self._post_resume, methods=["POST"]),
        Route("/api/shutdown", endpoint=self._post_shutdown, methods=["POST"]),
        Route(
            "/api/save-state",
            endpoint=self._post_save_state,
            methods=["POST"],
        ),
    ]

    # Create Starlette app with routes and exception handlers
    self._app = Starlette(
        routes=routes,
        exception_handlers={
            404: self._error_404,
            405: self._error_405,
            500: self._error_500,
        },
    )

    # Setup background thread
    self._handler_thread = threading.Thread(
        target=self._run_server, daemon=True, name="webapi"
    )

run_in_background

run_in_background() -> None

Run the API server in a background thread.

Source code in src/pamiq_core/console/web_api.py

def run_in_background(self) -> None:
    """Run the API server in a background thread."""
    self._handler_thread.start()

has_commands

has_commands() -> bool

Check if there are commands in the queue.

RETURNS	DESCRIPTION
bool	True if there are commands, False otherwise.

Source code in src/pamiq_core/console/web_api.py

def has_commands(self) -> bool:
    """Check if there are commands in the queue.

    Returns:
        True if there are commands, False otherwise.
    """
    return not self._received_commands_queue.empty()

receive_command

receive_command() -> ControlCommands

Receive a command from the queue without blocking.

RETURNS	DESCRIPTION
ControlCommands	The command from the queue.

RAISES	DESCRIPTION
Empty	If there are no commands in the queue.

Source code in src/pamiq_core/console/web_api.py

def receive_command(self) -> ControlCommands:
    """Receive a command from the queue without blocking.

    Returns:
        The command from the queue.

    Raises:
        Empty: If there are no commands in the queue.
    """
    return self._received_commands_queue.get_nowait()

pamiq_core.console.WebApiClient

WebApiClient(host: str, port: int)

Client for PAMIQ Web API communication.

Provides methods to interact with PAMIQ system via HTTP API.

Initialize Web API client.

PARAMETER	DESCRIPTION
host	API server host TYPE: str
port	API server port TYPE: int

Source code in src/pamiq_core/console/web_api.py

def __init__(self, host: str, port: int) -> None:
    """Initialize Web API client.

    Args:
        host: API server host
        port: API server port
    """
    self.host = host
    self.port = port
    self._client = httpx.Client()

get_status

get_status() -> SystemStatus

Get system status.

RETURNS	DESCRIPTION
SystemStatus	Status enum. If error is occurred, return offline status

Source code in src/pamiq_core/console/web_api.py

def get_status(self) -> SystemStatus:
    """Get system status.

    Returns:
        Status enum. If error is occurred, return offline status
    """
    try:
        response = self._client.get(f"{self._base_url}/status")
        response.raise_for_status()
        return SystemStatus(json.loads(response.text)["status"])
    except (httpx.RequestError, httpx.HTTPStatusError):
        return SystemStatus.OFFLINE

pause

pause() -> str | None

Pause the system.

RETURNS	DESCRIPTION
str | None	Result message or None if request failed

Source code in src/pamiq_core/console/web_api.py

def pause(self) -> str | None:
    """Pause the system.

    Returns:
        Result message or None if request failed
    """
    try:
        response = self._client.post(f"{self._base_url}/pause")
        response.raise_for_status()
        return json.loads(response.text)["result"]
    except (httpx.RequestError, httpx.HTTPStatusError):
        return None

resume

resume() -> str | None

Resume the system.

RETURNS	DESCRIPTION
str | None	Result message or None if request failed

Source code in src/pamiq_core/console/web_api.py

def resume(self) -> str | None:
    """Resume the system.

    Returns:
        Result message or None if request failed
    """
    try:
        response = self._client.post(f"{self._base_url}/resume")
        response.raise_for_status()
        return json.loads(response.text)["result"]
    except (httpx.RequestError, httpx.HTTPStatusError):
        return None

save_state

save_state() -> str | None

Save system state.

RETURNS	DESCRIPTION
str | None	Result message or None if request failed

Source code in src/pamiq_core/console/web_api.py

def save_state(self) -> str | None:
    """Save system state.

    Returns:
        Result message or None if request failed
    """
    try:
        response = self._client.post(f"{self._base_url}/save-state")
        response.raise_for_status()
        return json.loads(response.text)["result"]
    except (httpx.RequestError, httpx.HTTPStatusError):
        return None

shutdown

shutdown() -> str | None

Shutdown the system.

RETURNS	DESCRIPTION
str | None	Result message or None if request failed

Source code in src/pamiq_core/console/web_api.py

def shutdown(self) -> str | None:
    """Shutdown the system.

    Returns:
        Result message or None if request failed
    """
    try:
        response = self._client.post(f"{self._base_url}/shutdown")
        response.raise_for_status()
        return json.loads(response.text)["result"]
    except (httpx.RequestError, httpx.HTTPStatusError):
        return None

close

close() -> None

Close the HTTP client.

Source code in src/pamiq_core/console/web_api.py

def close(self) -> None:
    """Close the HTTP client."""
    self._client.close()

pamiq_core.console.SystemStatus

Bases: Enum

Enum representing the system's operational status.

status_name property

status_name: str

Returns the lowercase string representation of the status.

RETURNS	DESCRIPTION
str	The lowercase string name of the status.

pamiq_core.console.Console

Console(host: str, port: int)

pamiq-console.

Users can Control pamiq with CUI interface interactively.

Initialize CUI interface.

Source code in src/pamiq_core/console/cui.py

def __init__(self, host: str, port: int) -> None:
    """Initialize CUI interface."""
    super().__init__()
    self._client = WebApiClient(host, port)
    self.all_commands: list[str] = [
        attr[len("command_") :] for attr in dir(self) if attr.startswith("command_")
    ]
    self._completer = WordCompleter(self.all_commands)
    self.status = SystemStatus.OFFLINE

fetch_status

fetch_status() -> None

Check WebAPI status.

Source code in src/pamiq_core/console/cui.py

def fetch_status(self) -> None:
    """Check WebAPI status."""
    self.status = self._client.get_status()

run_command

run_command(command: str) -> bool | None

Check connection status before command execution.

Source code in src/pamiq_core/console/cui.py

def run_command(self, command: str) -> bool | None:
    """Check connection status before command execution."""
    # Update self.status before command execution.
    self.fetch_status()
    # Check command depend on WebAPI
    if command in ["pause", "p", "resume", "r", "save", "shutdown"]:
        # Check if WebAPI available.
        if self.status is SystemStatus.OFFLINE:
            print(f'Command "{command}" not executed. Can\'t connect AMI system.')
            return False
    # Execute command
    loop_end = getattr(self, f"command_{command}")()
    # Update self.status after command execution.
    self.fetch_status()
    # If True, main_loop ends.
    return loop_end

main_loop

main_loop() -> None

Running CUI interface.

Source code in src/pamiq_core/console/cui.py

def main_loop(self) -> None:
    """Running CUI interface."""
    print('Welcome to the PAMIQ console. "help" lists commands.\n')
    while True:
        self.fetch_status()
        command = prompt(
            f"pamiq-console ({self.status.status_name}) > ",
            completer=self._completer,
        )
        if command == "":
            continue
        if command in self.all_commands:
            if self.run_command(command):
                break
        else:
            print(f"*** Unknown syntax: {command}")

command_help

command_help() -> None

Show all commands and details.

Source code in src/pamiq_core/console/cui.py

def command_help(self) -> None:
    """Show all commands and details."""
    print(
        "\n".join(
            [
                "h/help    Show all commands and details.",
                "p/pause   Pause the AMI system.",
                "r/resume  Resume the AMI system.",
                "save      Save a checkpoint.",
                "shutdown  Shutdown the AMI system.",
                "q/quit    Exit the console.",
            ]
        )
    )

command_h

command_h() -> None

Show all commands and details.

Source code in src/pamiq_core/console/cui.py

def command_h(self) -> None:
    """Show all commands and details."""
    self.command_help()

command_pause

command_pause() -> None

Pause the AMI system.

Source code in src/pamiq_core/console/cui.py

def command_pause(self) -> None:
    """Pause the AMI system."""
    response = self._client.pause()
    if response:
        print(response)
    else:
        print("Failed to pause...")

command_p

command_p() -> None

Pause the AMI system.

Source code in src/pamiq_core/console/cui.py

def command_p(self) -> None:
    """Pause the AMI system."""
    self.command_pause()

command_resume

command_resume() -> None

Resume the AMI system.

Source code in src/pamiq_core/console/cui.py

def command_resume(self) -> None:
    """Resume the AMI system."""
    response = self._client.resume()
    if response:
        print(response)
    else:
        print("Failed to resume...")

command_r

command_r() -> None

Resume the AMI system.

Source code in src/pamiq_core/console/cui.py

def command_r(self) -> None:
    """Resume the AMI system."""
    self.command_resume()

command_shutdown

command_shutdown() -> bool

Shutdown the AMI system.

Source code in src/pamiq_core/console/cui.py

def command_shutdown(self) -> bool:
    """Shutdown the AMI system."""
    confirm = input("Confirm AMI system shutdown? (y/[N]): ")
    if confirm.lower() in ["y", "yes"]:
        response = self._client.shutdown()
        if response:
            return True
        else:
            print("Failed to shutdown...")
    print("Shutdown cancelled.")
    return False

command_quit

command_quit() -> bool

Exit the console.

Source code in src/pamiq_core/console/cui.py

def command_quit(self) -> bool:
    """Exit the console."""
    return True

command_q

command_q() -> bool

Exit the console.

Source code in src/pamiq_core/console/cui.py

def command_q(self) -> bool:
    """Exit the console."""
    return self.command_quit()

command_save

command_save() -> None

Save a checkpoint.

Source code in src/pamiq_core/console/cui.py

def command_save(self) -> None:
    """Save a checkpoint."""
    response = self._client.save_state()
    if response:
        print(response)
    else:
        print("Failed to save state...")