Skip to main content

A lightweight tool to mirror and dynamically update a Python list in your terminal, with built-in support for concurrent output (asyncio / threading / multiprocessing).

Project description

ci Coverage PyPI version

list2term

A lightweight tool to mirror and dynamically update a Python list in your terminal, with built-in support for concurrent output (asyncio / threading / multiprocessing).

Why use list2term?

  • Live list reflection: keep a list’s contents in sync with your terminal display — updates, additions, or removals are reflected in place.
  • Minimal dependencies: not a full TUI framework—just what you need to display and update lists.
  • Concurrency-aware: includes helpers for safely displaying progress or status messages from asyncio tasks, multiprocessing.Pool workers or threads.
  • TTY-aware fallback: detects when output isn’t a terminal (e.g. piped logs) and disables interactive behavior gracefully.
  • Thread safety: all public mutating operations are serialized with a re-entrant lock, ensuring atomic updates to internal state and terminal output when called from multiple threads.

Installation

pip install list2term

Key Concepts & API

Lines — main class

list2term revolves around the Lines class, a subclass of collections.UserList, which you use to represent and display a list in the terminal.

Constructor signature (default values shown):

Lines(
    data=None,
    size=None,
    lookup=None,
    show_index=True,
    show_x_axis=True,
    max_chars=None,
    use_color=True,
    y_axis_labels=None,
    x_axis=None)

Parameters

Parameter Description
data The initial list or iterable containing the items to display and sync with the terminal.
size Integer specifying the initial length of the list. When provided, the list is pre-populated with empty strings. Use this when you know the desired list size but not the initial values.
lookup A list of unique string identifiers used to route messages from concurrent workers to specific lines. Each identifier in the lookup list corresponds to one line in the display (default: None).
show_index Boolean flag to display line indices or labels on the left side of each line (default: True).
show_x_axis Boolean flag to display an X-axis ruler above the data for reference (default: True).
max_chars Maximum character width allowed per line; text exceeding this limit is truncated and suffixed with ... (default: 150).
use_color Boolean flag to apply terminal color styling to line indices and labels (default: True).
y_axis_labels A list of custom labels to display on the Y-axis (left side), replacing default numeric indices. Must match the length of data. Labels are right-justified before each line (default: None, uses numeric indices).
x_axis A string or list of strings to display as X-axis ruler(s) above the data. Accepts a single string for one line or a list for multiple lines. If not provided, a default numbered ruler is auto-generated (default: None).

Internally, Lines is backed by its .data attribute (like any UserList). You can mutate it:

lines[index] = "new value"
lines.append("another")
lines.pop(2)

These updates automatically refresh the terminal.

Concurrent Workers & Message Routing

When running tasks concurrently (via asyncio or multiprocessing.Pool), you often want each worker to report status lines. list2term supports that via:

Lines.write(...) — accepts strings in the form "{identifier}->{message}". The identifier is looked up in lookup to decide which line to update.

Multiprocessing helpers — the package offers pool_map and other abstractions in list2term.multiprocessing to simplify running functions in parallel and routing their messages.

Your worker functions must accept a logging object (e.g. LinesQueue) and use logger.write(...) to send messages back.

Examples

Display list - example1

Start with a list of 15 items containing random sentences, then update sentences at random indexes. As items in the list are updated the respective line in the terminal is updated to show the current contents of the list.

Code
import time
import random
from faker import Faker
from list2term import Lines

def main():
    print('Generating random sentences...')
    docgen = Faker()
    with Lines(size=15, show_x_axis=True, max_chars=100) as lines:
        for _ in range(200):
            index = random.randint(0, len(lines) - 1)
            lines[index] = docgen.sentence()
            time.sleep(.05)

if __name__ == '__main__':
    main()

example1

Display list of dynamic size - example2

Start with a list of 10 items containing random sentences, then add sentences to the list, update existing sentences or remove items from the list at random indexes. As items in the list are added, updated, and removed the respective line in the terminal is updated to show the current contents of the list.

Code
import time
import random
from faker import Faker
from list2term import Lines

def main():
    print('Generating random sentences...')
    docgen = Faker()
    with Lines(data=[''] * 10, max_chars=100) as lines:
        for _ in range(100):
            index = random.randint(0, len(lines) - 1)
            lines[index] = docgen.sentence()
        for _ in range(100):
            update = ['update'] * 18
            append = ['append'] * 18
            pop = ['pop'] * 14
            clear = ['clear']
            choice = random.choice(append + pop + clear + update)
            if choice == 'pop':
                if len(lines) > 0:
                    index = random.randint(0, len(lines) - 1)
                    lines.pop(index)
            elif choice == 'append':
                lines.append(docgen.sentence())
            elif choice == 'update':
                if len(lines) > 0:
                    index = random.randint(0, len(lines) - 1)
                    lines[index] = docgen.sentence()
            else:
                if len(lines) > 0:
                    lines.pop()
                if len(lines) > 0:
                    lines.pop()
            time.sleep(.1)

if __name__ == '__main__':
    main()

example2

Display messages from asyncio processes - example3

This example demonstrates how list2term can be used to display messages from asyncio processes to the terminal. Each item of the list represents a asnycio process.

Code
import asyncio
import random
from faker import Faker
from list2term import Lines

async def do_work(worker, lines):
    total = random.randint(10, 65)
    for _ in range(total):
        # mimic an IO-bound process
        await asyncio.sleep(random.choice([.05, .1, .025]))
        lines[worker] = f'processed {Faker().name()}'
    return total

async def run(workers):
    y_axis_labels = [f'Worker {str(i + 1).zfill(len(str(workers)))}' for i in range(workers)]
    with Lines(size=workers, y_axis_labels=y_axis_labels) as lines:
        return await asyncio.gather(*(do_work(worker, lines) for worker in range(workers)))

def main():
    workers = 15
    print(f'Total of {workers} workers working concurrently')
    results = asyncio.run(run(workers))
    print(f'The {workers} workers processed a total of {sum(results)} items')

if __name__ == '__main__':
    main()

example3

Display messages from multiprocessing pool processes - example4

This example demonstrates how list2term can be used to display messages from processes executing in a multiprocessing Pool. Each item of the list represents a background process. The list2term.multiprocessing module contains a pool_map method that fully abstracts the required multiprocessing constructs, you simply pass it the function to execute, an iterable of arguments to pass each process, and an optional instance of Lines. The method will execute the functions asynchronously, update the terminal lines accordingly and return a multiprocessing.pool.AsyncResult object. Each line in the terminal represents a background worker process.

If you do not wish to use the abstraction, the list2term.multiprocessing module contains helper classes that facilitates communication between the worker processes and the main process; the QueueManager provide a way to create a LinesQueue queue which can be shared between different processes. Refer to example4b for how the helper methods can be used.

Note the function being executed must accept a LinesQueue object that is used to write messages via its write method, this is the mechanism for how messages are sent from the worker processes to the main process, it is the main process that is displaying the messages to the terminal. The messages must be written using the format {identifier}->{message}, where {identifier} is a string that uniquely identifies a process, defined via the lookup argument to Lines.

Code
import time
from list2term import Lines
from list2term.multiprocessing import pool_map
from list2term.multiprocessing import CONCURRENCY

def is_prime(num):
    if num == 1:
        return False
    for i in range(2, num):
        if (num % i) == 0:
            return False
    else:
        return True

def count_primes(start, stop, logger):
    worker_id = f'{start}:{stop}'
    primes = 0
    for number in range(start, stop):
        if is_prime(number):
            primes += 1
            logger.write(f'{worker_id}->{worker_id} {number} is prime')
    logger.write(f'{worker_id}->{worker_id} processing complete')
    return primes

def main(number):
    step = int(number / CONCURRENCY)
    print(f"Distributing {int(number / step)} ranges across {CONCURRENCY} workers running concurrently")
    iterable = [(index, index + step) for index in range(0, number, step)]
    lookup = [':'.join(map(str, item)) for item in iterable]
    # print to screen with lines context
    results = pool_map(count_primes, iterable, context=Lines(lookup=lookup))
    return sum(results.get())

if __name__ == '__main__':
    start = time.perf_counter()
    number = 100_000
    result = main(number)
    stop = time.perf_counter()
    print(f"Finished in {round(stop - start, 2)} seconds\nTotal number of primes between 0-{number}: {result}")

example4

Displaying messages from threads - example5

Code
import time
import random
import threading
from faker import Faker
from concurrent.futures import ThreadPoolExecutor
from list2term import Lines

def process_item(item, lines):
    thread_name = threading.current_thread().name
    lines.write(f'{Faker().name()} processed item {item}', line_id=thread_name)
    seconds = random.uniform(.04, .3)
    time.sleep(seconds)
    return seconds

def main():
    items = 500
    num_threads = 10
    with ThreadPoolExecutor(max_workers=num_threads, thread_name_prefix='thread') as executor:
        lookup = [f'thread_{index}' for index in range(num_threads)]
        with Lines(lookup=lookup) as lines:
            futures = [executor.submit(process_item, item, lines) for item in range(items)]
            return [future.result() for future in futures]

if __name__ == "__main__":
    main()

example5

Other examples

A Conway Game-Of-Life implementation that uses list2term to display game to the terminal.

Caveats & Notes

  • Best for small to medium lists — list2term is optimized for relatively compact lists (e.g. dozens to low hundreds of lines). Very large lists (> thousands) may overwhelm the terminal.

  • Printable elements — items must be convertible to str.

  • Non-TTY fallback — if the terminal output is not a TTY (e.g. piped to a file), interactive updates are disabled automatically.

  • Worker message format — when using concurrency, messages must either follow the pattern "{identifier}->{message}" so that Lines.write() can route updates to the correct line. Or pass in lines_id argument to Lines.write().

Development

Clone the repository and ensure the latest version of Docker is installed on your development server.

Build the Docker image:

docker image build \
-t list2term:latest .

Run the Docker container:

docker container run \
--rm \
-it \
-v $PWD:/code \
list2term:latest \
bash

Execute the dev pipeline:

make dev

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

list2term-1.3.1.tar.gz (20.2 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

list2term-1.3.1-py3-none-any.whl (14.8 kB view details)

Uploaded Python 3

File details

Details for the file list2term-1.3.1.tar.gz.

File metadata

  • Download URL: list2term-1.3.1.tar.gz
  • Upload date:
  • Size: 20.2 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for list2term-1.3.1.tar.gz
Algorithm Hash digest
SHA256 f91ec38dc45812ddcb9200682fb58678624864de87d9eadba9bf082afde1e0bf
MD5 9969d91e21c426583df6eea5cf63e08a
BLAKE2b-256 c5567f0ee286fb1e6fb562aad5ecf51b135e2fe3ee940328f99dce54f65cd59b

See more details on using hashes here.

File details

Details for the file list2term-1.3.1-py3-none-any.whl.

File metadata

  • Download URL: list2term-1.3.1-py3-none-any.whl
  • Upload date:
  • Size: 14.8 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.12.12

File hashes

Hashes for list2term-1.3.1-py3-none-any.whl
Algorithm Hash digest
SHA256 863bca95729369274bd3276c7389a45ffe5fc2472fe53e8f1ada80093db7bcad
MD5 d80816b81d22297848f1539ebb06b555
BLAKE2b-256 5d4682b9ba2b4e1100d0821b671a12e920cd46a5f38220a5a3d3f9370f212a10

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page