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']
BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8)
OP_SEQ = '\033[%dm'
RESET_SEQ = '\033[0m'
COLOR_SEQ = '\033[1;%dm'
IS_WINDOWS = platform.system() == 'Windows'
ITERM_PROFILE = os.environ.get('ITERM_PROFILE')
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
@python_2_unicode_compatible
[docs]class colored(object):
"""Terminal colored text.
Example:
>>> c = colored(enabled=True)
>>> print(str(c.red('the quick '), c.blue('brown ', c.bold('fox ')),
... c.magenta(c.underline('jumps over')),
... c.yellow(' the lazy '),
... c.green('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': self.black,
'red': self.red,
'green': self.green,
'yellow': self.yellow,
'blue': self.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 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))))
def __repr__(self):
return repr(self.no_color())
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 codecs.open(path, mode='rb') as fh:
encoded = base64.b64encode(fh.read())
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)