controller

dev_tool.launcher.ipc.controller

log = logging.getLogger(__name__) module-attribute

AntiAustinMechanism

A class for ensuring only one instance of the application runs at a time.

This class uses a network socket to lock a port and prevent multiple instances.

The constructor for the AntiAustinMechanism class.

Parameters:

• base_port (int, default: 6666 ) –

  The base port number for socket binding.

• host (str, default: '127.0.0.1' ) –

  The host address for socket binding.

Source code in dev_tool/launcher/ipc/controller.py

def __init__(self, base_port: int = 6666, host: str = '127.0.0.1') -> None:
    """
    The constructor for the AntiAustinMechanism class.

    :param base_port: The base port number for socket binding.
    :param host: The host address for socket binding.
    """

    self.base_port = base_port
    self.host = host
    self.connection: socket.socket | None = None
    self.lock_port: int = 0
    self.ipc_port: int = 0
    self.ipc_server: IPCServer | None = None
    self._calculate_port()

base_port = base_port instance-attribute

host = host instance-attribute

connection = None instance-attribute

lock_port = 0 instance-attribute

ipc_port = 0 instance-attribute

ipc_server = None instance-attribute

__enter__

The context manager entry method that acquires the lock.

Returns:

• Self –

  The AntiAustinMechanism instance.

Source code in dev_tool/launcher/ipc/controller.py

def __enter__(self) -> Self:
    """
    The context manager entry method that acquires the lock.

    :return: The AntiAustinMechanism instance.
    """

    self.acquire_lock()
    return self

__exit__

The context manager exit method that releases the lock.

Parameters:

• exc_type (type[BaseException] | None) –

  The exception type, if an exception was raised.

• exc_val (BaseException | None) –

  The exception value, if an exception was raised.

• traceback (object) –

  The traceback, if an exception was raised.

Source code in dev_tool/launcher/ipc/controller.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    traceback: object
) -> None:
    """
    The context manager exit method that releases the lock.

    :param exc_type: The exception type, if an exception was raised.
    :param exc_val: The exception value, if an exception was raised.
    :param traceback: The traceback, if an exception was raised.
    """

    self.release_lock()

acquire_lock

A method that acquires the lock by binding to a socket.

If the lock cannot be acquired, this method will prompt the user to send a shutdown command to the running instance.

Source code in dev_tool/launcher/ipc/controller.py

def acquire_lock(self) -> None:
    """
    A method that acquires the lock by binding to a socket.

    If the lock cannot be acquired, this method will prompt the user to send
    a shutdown command to the running instance.
    """

    self.connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    try:
        socket_address = (self.host, self.lock_port)
        self.connection.bind(socket_address)
    except OSError as exception:
        message = f'An instance is already running for "{CONTEXT.configuration.get_project_name()}".'
        CONTEXT.notification.warning_text(message)

        log.exception(message)

        prompt = (
            f'Another instance is running on {self.host}:{self.lock_port}. \n'
            'Do you want to send a shutdown command to the running instance? (y/n): '
        )

        if get_input_and_clear_terminal(prompt) == 'y':
            client = IPCClient(self.ipc_port, self.host)

            try:
                client.send_shutdown()

                message = 'Shutdown command sent. Waiting for the instance to shut down...'
                CONTEXT.notification.normal_text(message)

                log.debug(message)

                time.sleep(2)

                socket_address = (self.host, self.lock_port)
                self.connection.bind(socket_address)
            except Exception as exception:
                message = f'Failed to send shutdown command via IPC: {exception}'
                CONTEXT.notification.error_banner(message)

                log.exception(message)
                raise LauncherLockError(message) from exception
        else:
            message = (
                'Please exit the running instance before attempting '
                'to launch another instance or set "single-instance" to '
                '"false" in the pyproject.toml. Exiting...'
            )

            CONTEXT.notification.error_banner(message)

            log.exception(message)
            raise SingleInstanceError(message) from exception

start_ipc_server

A method that starts the IPC server for handling shutdown commands.

Parameters:

• shutdown_callback (Callable[[], None]) –

  The callback function to call when a shutdown command is received.

Source code in dev_tool/launcher/ipc/controller.py

def start_ipc_server(self, shutdown_callback: Callable[[], None]) -> None:
    """
    A method that starts the IPC server for handling shutdown commands.

    :param shutdown_callback: The callback function to call when a shutdown command is received.
    """

    self.ipc_server = IPCServer(
        self.ipc_port,
        shutdown_callback,
        self.host
    )

    self.ipc_server.start()

release_lock

A method that releases the lock by closing the socket and stopping the IPC server.

Source code in dev_tool/launcher/ipc/controller.py

def release_lock(self) -> None:
    """A method that releases the lock by closing the socket and stopping the IPC server."""

    if self.connection:
        self.connection.close()
        self.connection = None

    if self.ipc_server:
        self.ipc_server.stop()