Skip to content

worker_formatter

Worker formatter base module for displaying worker query results.

This module defines the abstract base class WorkerFormatter, which provides a standard interface for formatting and displaying information about Merlin workers. Worker formatters are responsible for presenting logical and physical worker information in a structured, user-friendly manner (e.g., through text, tables, or rich console output).

Intended Usage:

Subclasses of `WorkerFormatter` (e.g., those using Rich for terminal
visualization) should implement `format_and_display` to render
worker information, while reusing `get_worker_statistics` for
consistent metrics across implementations.

WorkerFormatter

Bases: ABC

Abstract base class for formatting and displaying worker query results.

Provides a consistent interface for formatting logical and physical worker information, including a utility method to calculate worker statistics.

Attributes:

Name Type Description
console Console

A Rich Console object used for displaying output to the terminal.

Methods:

Name Description
format_and_display

Abstract method that formats and outputs worker information. Must be implemented by subclasses.
get_worker_statistics

Compute counts and statuses of logical and physical workers, including totals and breakdown by status.
Source code in merlin/workers/formatters/worker_formatter.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
class WorkerFormatter(ABC):
    """
    Abstract base class for formatting and displaying worker query results.

    Provides a consistent interface for formatting logical and physical worker
    information, including a utility method to calculate worker statistics.

    Attributes:
        console (rich.console.Console): A Rich Console object used for displaying
            output to the terminal.

    Methods:
        format_and_display: Abstract method that formats and outputs worker
            information. Must be implemented by subclasses.
        get_worker_statistics: Compute counts and statuses of logical and
            physical workers, including totals and breakdown by status.
    """

    def __init__(self):
        """
        Initializer for the WorkerFormatter class.

        Sets up a rich console object so that subclasses can easily print formatted
        output to the console.
        """
        self.console = Console()

    @abstractmethod
    def format_and_display(self, logical_workers: List, filters: Dict, merlin_db: MerlinDatabase):
        """
        Format and display information about logical and physical workers.

        This method must be implemented by subclasses to define the output
        format (e.g., JSON, Rich tables, text). Implementations should make
        use of `get_worker_statistics` if worker summary metrics are required.

        Args:
            logical_workers (List[LogicalWorkerEntity]): List of logical worker
                entities to be displayed.
            filters (Dict): Optional filters applied to the worker query.
            merlin_db (MerlinDatabase): Database interface for retrieving
                physical worker details.
        """
        raise NotImplementedError("Subclasses of `WorkerFormatter` must implement a `format_and_display` method.")

    def get_worker_statistics(self, logical_workers: List[LogicalWorkerEntity], merlin_db: MerlinDatabase) -> Dict[str, int]:
        """
        Calculate comprehensive statistics for logical and physical workers.

        Iterates through all logical workers and their associated physical
        instances to compute counts of running, stopped, stalled, and rebooting
        workers, as well as counts of logical workers with or without instances.

        Args:
            logical_workers (List[db_scripts.entities.logical_worker_entity.LogicalWorkerEntity]):
                List of logical worker entities.
            merlin_db (db_scripts.merlin_db.MerlinDatabase): Database interface to fetch physical
                worker details.

        Returns:
            Dictionary containing worker statistics:\n
                - total_logical: Total number of logical workers.
                - logical_with_instances: Number of logical workers with physical instances.
                - logical_without_instances: Number of logical workers without physical instances.
                - total_physical: Total number of physical workers.
                - physical_running: Count of running physical workers.
                - physical_stopped: Count of stopped physical workers.
                - physical_stalled: Count of stalled physical workers.
                - physical_rebooting: Count of rebooting physical workers.
        """
        stats = {
            "total_logical": len(logical_workers),
            "logical_with_instances": 0,
            "logical_without_instances": 0,
            "total_physical": 0,
            "physical_running": 0,
            "physical_stopped": 0,
            "physical_stalled": 0,
            "physical_rebooting": 0,
        }

        for logical_worker in logical_workers:
            physical_worker_ids = logical_worker.get_physical_workers()

            if physical_worker_ids:
                stats["logical_with_instances"] += 1
                physical_workers = [merlin_db.get("physical_worker", pid) for pid in physical_worker_ids]

                for physical_worker in physical_workers:
                    stats["total_physical"] += 1
                    status = physical_worker.get_worker_status()

                    if status == WorkerStatus.RUNNING:
                        stats["physical_running"] += 1
                    elif status == WorkerStatus.STOPPED:
                        stats["physical_stopped"] += 1
                    elif status == WorkerStatus.STALLED:
                        stats["physical_stalled"] += 1
                    elif status == WorkerStatus.REBOOTING:
                        stats["physical_rebooting"] += 1
            else:
                stats["logical_without_instances"] += 1

        return stats

__init__()

Initializer for the WorkerFormatter class.

Sets up a rich console object so that subclasses can easily print formatted output to the console.

Source code in merlin/workers/formatters/worker_formatter.py 
51
52
53
54
55
56
57
58
def __init__(self):
    """
    Initializer for the WorkerFormatter class.

    Sets up a rich console object so that subclasses can easily print formatted
    output to the console.
    """
    self.console = Console()

format_and_display(logical_workers, filters, merlin_db) abstractmethod

Format and display information about logical and physical workers.

This method must be implemented by subclasses to define the output format (e.g., JSON, Rich tables, text). Implementations should make use of get_worker_statistics if worker summary metrics are required.

Parameters:

Name Type Description Default
logical_workers List[LogicalWorkerEntity]

List of logical worker entities to be displayed.

 required
filters Dict

Optional filters applied to the worker query.

 required
merlin_db MerlinDatabase

Database interface for retrieving physical worker details.

 required
Source code in merlin/workers/formatters/worker_formatter.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
@abstractmethod
def format_and_display(self, logical_workers: List, filters: Dict, merlin_db: MerlinDatabase):
    """
    Format and display information about logical and physical workers.

    This method must be implemented by subclasses to define the output
    format (e.g., JSON, Rich tables, text). Implementations should make
    use of `get_worker_statistics` if worker summary metrics are required.

    Args:
        logical_workers (List[LogicalWorkerEntity]): List of logical worker
            entities to be displayed.
        filters (Dict): Optional filters applied to the worker query.
        merlin_db (MerlinDatabase): Database interface for retrieving
            physical worker details.
    """
    raise NotImplementedError("Subclasses of `WorkerFormatter` must implement a `format_and_display` method.")

get_worker_statistics(logical_workers, merlin_db)

Calculate comprehensive statistics for logical and physical workers.

Iterates through all logical workers and their associated physical instances to compute counts of running, stopped, stalled, and rebooting workers, as well as counts of logical workers with or without instances.

Parameters:

Name Type Description Default
logical_workers List[LogicalWorkerEntity]

List of logical worker entities.

 required
merlin_db MerlinDatabase

Database interface to fetch physical worker details.

 required

Returns:

Type Description
Dict[str, int]

Dictionary containing worker statistics:

  • total_logical: Total number of logical workers.
  • logical_with_instances: Number of logical workers with physical instances.
  • logical_without_instances: Number of logical workers without physical instances.
  • total_physical: Total number of physical workers.
  • physical_running: Count of running physical workers.
  • physical_stopped: Count of stopped physical workers.
  • physical_stalled: Count of stalled physical workers.
  • physical_rebooting: Count of rebooting physical workers.
Source code in merlin/workers/formatters/worker_formatter.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
def get_worker_statistics(self, logical_workers: List[LogicalWorkerEntity], merlin_db: MerlinDatabase) -> Dict[str, int]:
    """
    Calculate comprehensive statistics for logical and physical workers.

    Iterates through all logical workers and their associated physical
    instances to compute counts of running, stopped, stalled, and rebooting
    workers, as well as counts of logical workers with or without instances.

    Args:
        logical_workers (List[db_scripts.entities.logical_worker_entity.LogicalWorkerEntity]):
            List of logical worker entities.
        merlin_db (db_scripts.merlin_db.MerlinDatabase): Database interface to fetch physical
            worker details.

    Returns:
        Dictionary containing worker statistics:\n
            - total_logical: Total number of logical workers.
            - logical_with_instances: Number of logical workers with physical instances.
            - logical_without_instances: Number of logical workers without physical instances.
            - total_physical: Total number of physical workers.
            - physical_running: Count of running physical workers.
            - physical_stopped: Count of stopped physical workers.
            - physical_stalled: Count of stalled physical workers.
            - physical_rebooting: Count of rebooting physical workers.
    """
    stats = {
        "total_logical": len(logical_workers),
        "logical_with_instances": 0,
        "logical_without_instances": 0,
        "total_physical": 0,
        "physical_running": 0,
        "physical_stopped": 0,
        "physical_stalled": 0,
        "physical_rebooting": 0,
    }

    for logical_worker in logical_workers:
        physical_worker_ids = logical_worker.get_physical_workers()

        if physical_worker_ids:
            stats["logical_with_instances"] += 1
            physical_workers = [merlin_db.get("physical_worker", pid) for pid in physical_worker_ids]

            for physical_worker in physical_workers:
                stats["total_physical"] += 1
                status = physical_worker.get_worker_status()

                if status == WorkerStatus.RUNNING:
                    stats["physical_running"] += 1
                elif status == WorkerStatus.STOPPED:
                    stats["physical_stopped"] += 1
                elif status == WorkerStatus.STALLED:
                    stats["physical_stalled"] += 1
                elif status == WorkerStatus.REBOOTING:
                    stats["physical_rebooting"] += 1
        else:
            stats["logical_without_instances"] += 1

    return stats