107 lines
2.9 KiB
Python
107 lines
2.9 KiB
Python
import logging
|
|
from enum import Enum
|
|
|
|
import boto3
|
|
from mypy_boto3_ec2 import EC2ServiceResource
|
|
from nicegui import ui
|
|
from nicegui.element import Element
|
|
from rich.console import Console
|
|
from rich.logging import RichHandler
|
|
from rich.traceback import install
|
|
|
|
|
|
class InstanceMode(Enum):
|
|
PENDING = 0
|
|
RUNNING = 16
|
|
SHUTTING_DOWN = 32
|
|
TERMINATED = 48
|
|
STOPPING = 64
|
|
STOPPED = 80
|
|
START = 1
|
|
STOP = 2
|
|
|
|
|
|
class InstanceStateHandler:
|
|
state: int
|
|
|
|
|
|
install(show_locals=True)
|
|
console = Console()
|
|
logging.basicConfig(
|
|
level="INFO",
|
|
format="%(message)s",
|
|
datefmt="[%X]",
|
|
handlers=[RichHandler(rich_tracebacks=True)],
|
|
)
|
|
|
|
log = logging.getLogger("rich")
|
|
|
|
|
|
def aws_launch_ec2(
|
|
ec2_resource: EC2ServiceResource,
|
|
name: str = "foundry",
|
|
mode: InstanceMode = InstanceMode.START,
|
|
) -> None:
|
|
for instance in ec2_resource.instances.all():
|
|
for tag_dict in instance.tags:
|
|
if tag_dict["Key"] == "Name" and tag_dict["Value"] == name:
|
|
if (
|
|
instance.state["Code"] == 80 and mode == InstanceMode.START
|
|
): # stopped
|
|
log.info(f"Starting instance {instance.id}")
|
|
log.info(instance.start())
|
|
elif instance.state["Code"] == 16 and mode == InstanceMode.STOP:
|
|
log.info(f"Stopping instance {instance.id}")
|
|
log.info(instance.stop())
|
|
|
|
|
|
def update_instance_table(
|
|
ec2_resource: EC2ServiceResource, container: Element, columns: list
|
|
) -> None:
|
|
container.clear()
|
|
with container:
|
|
rows = []
|
|
table = ui.table(columns=columns, rows=rows, pagination=10)
|
|
for instance in ec2_resource.instances.all():
|
|
name = instance.id
|
|
for tag in instance.tags:
|
|
if tag["Key"] == "Name":
|
|
name = tag["Value"]
|
|
table.add_rows({"name": name, "status": instance.state["Name"]})
|
|
log.info("Table updated")
|
|
|
|
|
|
def nicegui_ui() -> None:
|
|
ec2_resource = boto3.resource("ec2")
|
|
with ui.row():
|
|
ui.button(
|
|
"Start Foundry AWS Server", on_click=lambda: aws_launch_ec2(ec2_resource)
|
|
)
|
|
ui.button(
|
|
"Stop Foundry AWS Server",
|
|
on_click=lambda: aws_launch_ec2(ec2_resource, mode=InstanceMode.STOP),
|
|
)
|
|
columns = [
|
|
{"name": "name", "label": "Name", "field": "name", "required": True},
|
|
{"name": "status", "label": "Status", "field": "status"},
|
|
]
|
|
rows = []
|
|
with ui.element() as container:
|
|
ui.table(columns=columns, rows=rows, pagination=10)
|
|
timer = ui.timer(
|
|
15, lambda: update_instance_table(ec2_resource, container, columns)
|
|
)
|
|
textarea = ui.textarea()
|
|
ui.slider(min=5, max=120, step=5, value=15).bind_value_to(
|
|
timer, "interval"
|
|
).bind_value_to(textarea)
|
|
ui.run()
|
|
|
|
|
|
def main() -> None:
|
|
nicegui_ui()
|
|
|
|
|
|
if __name__ in {"__main__", "__mp_main__"}:
|
|
main()
|