This document describes the current stable version of Celery (4.0). For development docs, go here.

Source code for celery.utils.term

# -*- coding: utf-8 -*-
"""Terminals and colors."""
from __future__ import absolute_import, unicode_literals
import base64
import codecs
import os
import sys
import platform
from functools import reduce
from celery.five import python_2_unicode_compatible, string
from celery.platforms import isatty

__all__ = ['colored']

OP_SEQ = '\033[%dm'
RESET_SEQ = '\033[0m'
COLOR_SEQ = '\033[1;%dm'

IS_WINDOWS = platform.system() == 'Windows'

TERM = os.environ.get('TERM')
TERM_IS_SCREEN = TERM and TERM.startswith('screen')

# tmux requires unrecognized OSC sequences to be wrapped with DCS tmux;
# <sequence> ST, and for all ESCs in <sequence> to be replaced with ESC ESC.
# It only accepts ESC backslash for ST.
_IMG_PRE = '\033Ptmux;\033\033]' if TERM_IS_SCREEN else '\033]'
_IMG_POST = '\a\033\\' if TERM_IS_SCREEN else '\a'

def fg(s):
    return COLOR_SEQ % s

[docs]class colored(object): """Terminal colored text. Example: >>> c = colored(enabled=True) >>> print(str('the quick '),'brown ', c.bold('fox ')), ... c.magenta(c.underline('jumps over')), ... c.yellow(' the lazy '), ...'dog '))) """ def __init__(self, *s, **kwargs): self.s = s self.enabled = not IS_WINDOWS and kwargs.get('enabled', True) self.op = kwargs.get('op', '') self.names = { 'black':, 'red':, 'green':, 'yellow': self.yellow, 'blue':, 'magenta': self.magenta, 'cyan': self.cyan, 'white': self.white, } def _add(self, a, b): return string(a) + string(b) def _fold_no_color(self, a, b): try: A = a.no_color() except AttributeError: A = string(a) try: B = b.no_color() except AttributeError: B = string(b) return ''.join((string(A), string(B)))
[docs] def no_color(self): if self.s: return string(reduce(self._fold_no_color, self.s)) return ''
[docs] def embed(self): prefix = '' if self.enabled: prefix = self.op return ''.join((string(prefix), string(reduce(self._add, self.s))))
def __str__(self): suffix = '' if self.enabled: suffix = RESET_SEQ return string(''.join((self.embed(), string(suffix))))
[docs] def node(self, s, op): return self.__class__(enabled=self.enabled, op=op, *s)
[docs] def black(self, *s): return self.node(s, fg(30 + BLACK))
[docs] def red(self, *s): return self.node(s, fg(30 + RED))
[docs] def green(self, *s): return self.node(s, fg(30 + GREEN))
[docs] def yellow(self, *s): return self.node(s, fg(30 + YELLOW))
[docs] def blue(self, *s): return self.node(s, fg(30 + BLUE))
[docs] def magenta(self, *s): return self.node(s, fg(30 + MAGENTA))
[docs] def cyan(self, *s): return self.node(s, fg(30 + CYAN))
[docs] def white(self, *s): return self.node(s, fg(30 + WHITE))
def __repr__(self): return repr(self.no_color())
[docs] def bold(self, *s): return self.node(s, OP_SEQ % 1)
[docs] def underline(self, *s): return self.node(s, OP_SEQ % 4)
[docs] def reverse(self, *s): return self.node(s, OP_SEQ % 7)
[docs] def bright(self, *s): return self.node(s, OP_SEQ % 8)
[docs] def ired(self, *s): return self.node(s, fg(40 + RED))
[docs] def igreen(self, *s): return self.node(s, fg(40 + GREEN))
[docs] def iyellow(self, *s): return self.node(s, fg(40 + YELLOW))
[docs] def iblue(self, *s): return self.node(s, fg(40 + BLUE))
[docs] def imagenta(self, *s): return self.node(s, fg(40 + MAGENTA))
[docs] def icyan(self, *s): return self.node(s, fg(40 + CYAN))
[docs] def iwhite(self, *s): return self.node(s, fg(40 + WHITE))
[docs] def reset(self, *s): return self.node(s or [''], RESET_SEQ)
def __add__(self, other): return string(self) + string(other)
def supports_images(): return isatty(sys.stdin) and ITERM_PROFILE def _read_as_base64(path): with, mode='rb') as fh: encoded = base64.b64encode( return encoded if type(encoded) == 'str' else encoded.decode('ascii') def imgcat(path, inline=1, preserve_aspect_ratio=0, **kwargs): return '\n%s1337;File=inline=%d;preserveAspectRatio=%d:%s%s' % ( _IMG_PRE, inline, preserve_aspect_ratio, _read_as_base64(path), _IMG_POST)