Source code for Karana.KUtils.multirun._display

# Copyright (c) 2024-2025 Karana Dynamics Pty Ltd. All rights reserved.
#
# NOTICE TO USER:
#
# This source code and/or documentation (the "Licensed Materials") is
# the confidential and proprietary information of Karana Dynamics Inc.
# Use of these Licensed Materials is governed by the terms and conditions
# of a separate software license agreement between Karana Dynamics and the
# Licensee ("License Agreement"). Unless expressly permitted under that
# agreement, any reproduction, modification, distribution, or disclosure
# of the Licensed Materials, in whole or in part, to any third party
# without the prior written consent of Karana Dynamics is strictly prohibited.
#
# THE LICENSED MATERIALS ARE PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND.
# KARANA DYNAMICS DISCLAIMS ALL WARRANTIES, EXPRESS OR IMPLIED, INCLUDING
# BUT NOT LIMITED TO WARRANTIES OF MERCHANTABILITY, NON-INFRINGEMENT, AND
# FITNESS FOR A PARTICULAR PURPOSE.
#
# IN NO EVENT SHALL KARANA DYNAMICS BE LIABLE FOR ANY DAMAGES WHATSOEVER,
# INCLUDING BUT NOT LIMITED TO LOSS OF PROFITS, DATA, OR USE, EVEN IF
# ADVISED OF THE POSSIBILITY OF SUCH DAMAGES, WHETHER IN CONTRACT, TORT,
# OR OTHERWISE ARISING OUT OF OR IN CONNECTION WITH THE LICENSED MATERIALS.
#
# U.S. Government End Users: The Licensed Materials are a "commercial item"
# as defined at 48 C.F.R. 2.101, and are provided to the U.S. Government
# only as a commercial end item under the terms of this license.
#
# Any use of the Licensed Materials in individual or commercial software must
# include, in the user documentation and internal source code comments,
# this Notice, Disclaimer, and U.S. Government Use Provision.

"""This module provides the capability to display the overall status of runs for an runner."""

import asyncio
import shutil
import sys

from ._runner import AbstractRunner

__all__ = ["displayStatus"]




async def displayStatus(runner: AbstractRunner):
    """Display and asynchronously update a run status bar.

    Parameters
    ----------
    runner: AbstractRunner
        The runner to monitor
    """
    if not sys.stdout.isatty():
        print("Omitting status bar: stdout not a tty")
        return

    def _buildStatus() -> list[str]:
        try:
            term_width = shutil.get_terminal_size().columns - 2
        except OSError:
            term_width = 38

        stats = runner.computeStatistics()
        percent = stats.complete_count / stats.total_count
        percent_str = f" {percent:.1%}"
        count_str = f"{stats.complete_count}/{stats.total_count}"
        text_len = len(count_str) + len(percent_str) + 3
        bar_width = max(1, term_width - text_len)
        filled_chars = int(bar_width * percent)
        bar = "#" * filled_chars + "-" * (bar_width - filled_chars)
        status_bar = f"{count_str} [{bar}]{percent_str}"

        tokens = [
            f"Skip:{stats.skipped_count}",
            f"Pend:{stats.pending_count}",
            f"Run:{stats.running_count}",
            f"Suc:{stats.success_count}",
            f"Fail:{stats.failure_count}",
            f"Tot:{stats.total_count}",
        ]
        return [
            status_bar,
            " ".join(tokens),
        ]

    for line in _buildStatus():
        print(line)

    try:
        while True:
            await asyncio.sleep(0.1)
            # Move the cursor up two lines
            print(f"\033[3A")
            for line in _buildStatus():
                print("\033[2K" + line)
    finally:
        # Move the cursor up two lines
        print(f"\033[3A")
        for line in _buildStatus():
            print("\033[2K" + line)