Сергей Горностаев

За минуту нагуглил пример из документации и несколько статей - раз, два, три. Ещё минута понадобилась на то, чтобы проверить работу кода:

import cv2
import numpy as np
from sklearn.cluster import KMeans
from collections import Counter


def get_dominant_color(image, k=4):
    image = image.reshape((image.shape[0] * image.shape[1], 3))

    clt = KMeans(n_clusters = k)
    labels = clt.fit_predict(image)

    label_counts = Counter(labels)

    dominant_color = clt.cluster_centers_[label_counts.most_common(1)[0][0]]

    return list(dominant_color)


bgr_image = cv2.imread('image.png')
hsv_image = cv2.cvtColor(bgr_image, cv2.COLOR_BGR2HSV)
dom_color = get_dominant_color(hsv_image)
dom_color_hsv = np.full(bgr_image.shape, dom_color, dtype='uint8')
dom_color_bgr = cv2.cvtColor(dom_color_hsv, cv2.COLOR_HSV2BGR)
output_image = np.hstack((bgr_image, dom_color_bgr))
cv2.imshow('Dominant Color', output_image)
cv2.waitKey(0)

При запуске JVM передать параметр -Djava.security.policy=server.policy, а в файле server.policy написать

grant {
    permission java.net.SocketPermission "127.0.0.1:25565", "connect,accept,resolve";
};

JVM не позволит программам занимать иные порты, кроме указанных в политике. Правда, она ещё много чего запретит и скорее всего вам придётся долго и кропотливо выяснять, какие ещё права нужны вашему серверу для нормальной работы.

Асинхронный код сложнее. Выбирать Торнадо стоит тогда, когда ваш сервер должен обрабатывать тысячи одновременно открытых соединений. Для вашей задачи вполне подойдёт связка из Django и Celery или просто из Django и запускаемых кроном management commands.

При создании БД на сервере ей надо было задать кодировку UTF-8.

При открытии окна со списком сообщений ставить таймут на запуск через 15-30 секунд javascript-функции, которая получит из этого окна список всех сообщений с меткой "непрочитано", отправит ajax'ом список их идентификаторов специальной вьюхе и в случае положительного ответа поставит им отметку "прочитано". Вьюха на серваке получается предельно быстрая - всего-то надо сделать пакетное обновление по списку первичных ключей. И нужно написать management command запускаемую кроном раз в 10 минут, выгребающую из базы все сообщения старше 10 минут с меткой "непрочитано" и отправляющую письмо "У вас есть непрочитанные сообщения". Причём после отправки надо где-нибудь поставить метку, что пользователю такому-то тогда-то было отправлено уведомление, и не отправлять новых до тех пор, пока пользователь не прочитает хоть одно сообщение.

Прежде всего хочу заметить, что настолько больших возможностей по расширению языка, как в Lisp'ах, никакой другой язык не предоставляет. Но в некоторых языках, в том числе в Python, можно немного поколдовать с кодом на этапе его разбора.

dumb_translator.py

import ast
import sys

class DumbLisp(ast.NodeTransformer):
    def fix(self, source, result):
        ast.copy_location(result, source)
        ast.fix_missing_locations(result)
        return result

    def visit_Expr(self, node):
        if isinstance(node.value, ast.Tuple):
            t = node.value
            if isinstance(t.elts[0], ast.Name):
                if t.elts[0].id == 'newif':
                    _else = [ast.Expr(value=t.elts[3])] if len(t.elts) == 4 else []
                    _if = ast.If(test=t.elts[1],
                                body=[ast.Expr(value=t.elts[2])],
                                orelse=_else)
                    self.generic_visit(_if)
                    return self.fix(node, _if)
                elif t.elts[0].id == 'define':
                    assign = ast.Assign(targets=[ast.Name(id=t.elts[1].id, ctx=ast.Store())],
                                        value=t.elts[2])
                    return self.fix(node, assign)
                else:
                    call = ast.Expr(value=ast.Call(func=t.elts[0], args=t.elts[1:], keywords=[]))
                    return self.fix(node, call)
        return node


with open(sys.argv[1]) as fh:
    tree = ast.parse(fh.read())
    DumbLisp().visit(tree)
    code = compile(tree, filename=sys.argv[1], mode="exec")
    exec(code)

test.dl

(define, a, 1)

(newif, a == 1, (print, 'Yes'), (print, 'No'))

Запускаем
python dumb_translator.py test.dl

Естественно, пример предельно простой, а потому в качестве s-форм приходится использовать кортежи. Но есть полноценный фронтенд компилятора Python, транслирующий в python-байткод диалект Lisp близкий к Clojure - Hy.

Есть у вас, предположим, модель новости и её можно комментировать, для чего используется модель комментария. А через месяц эксплуатации сайта заказчик захотел ещё и статьи, и чтобы их тоже можно было комментировать. А потом ещё события всякие - театральные премьеры, киносеансы, концерты. И обзоры ресторанов маячат впереди. Вот только ForeignKey может указывать только на конкретную модель. Придётся либо все модели наследовать от общего корня, либо по новой писать модель комментариев и весь сопроводительный код для каждой комментируемой модели. И то, и другое - решение плохое. Выход - обобщённые отношения, позволяющие одну модель комментариев использовать для неопределённо большого круга заранее неизвестных моделей.

Многопоточный вариант

import time
from threading import Thread

run = True

def ticker():
    while run:
        time.sleep(1)
        print('Тик')
        time.sleep(1)
        print('Так')


# Запуск функции ticker в отдельном потоке.
# Параметр daemon=True нужен чтобы
# дочерний поток умирал вместе с основным
# в случае внештатного выхода.
Thread(target=ticker, daemon=True).start()

while run:
    command = input('Для выхода введите "exit"\n')
    if command.lower() == 'exit':
        run = False

Асинхронный вариант

import asyncio
import signal

def shutdown():
    # Отменяем все задачи, кроме вызвавшей
    for task in asyncio.Task.all_tasks():
        if task is not asyncio.tasks.Task.current_task():
            task.cancel()


async def user_io():
    loop = asyncio.get_event_loop()
    # Ждём действия от пользователя
    while True:
        # Запускаем input() в отдельном потоке и ждём его завершения
        command = await loop.run_in_executor(None, input, 'Для выхода введите "exit"\n')
        if command.lower() == 'exit':
            shutdown() # Отменяем все задачи
            break      # и выходим из цикла


# Сопрограмма, выполняемая параллельно с ожиданием пользовательского ввода
async def task_manager():
    counter = 0
    while True:
        try:
            await asyncio.sleep(1)
        except asyncio.CancelledError:
            break # Выходим из цикла, если задачу отменили
        counter += 1
        print("I'm a task manager {}!".format(counter))


if __name__ == '__main__':
    # Устанавливаем обработчик Ctrl+C
    signal.signal(signal.SIGINT, lambda n, f: shutdown())

    # Запускаем цикл событий
    loop = asyncio.get_event_loop()
    # Задача ждущая завершения сопрограм user_io и task_manager
    main_task = asyncio.wait([user_io(), task_manager()])
    try:
        loop.run_until_complete(main_task)
    except asyncio.CancelledError:
        # Позволяем main_task завершиться
        loop.run_until_complete(main_task)
    loop.close()

Такого не должно возникать и лучше всё-таки выяснить первопричину, но возможен и костыль:

from os.path import splitext
from transliterate import slugify

def slugify_upload(instance, filename):
    name, ext = splitext(filename)
    return slugify(name) + ext


class Dancer(models.Model):
    photo = models.ImageFiled(upload_to=slugify_upload, ...)
    ...

Библиотека.

Библиотеки brukva и toredis уже не обновлялись по нескольку лет. Зато Tornado в последнем обновлении окончательно перешёл на использование очереди событий asyncio, что даёт возможность использовать свежие и хорошие библиотеки. Например aioredis:

import asyncio
import aioredis
from tornado import web, websocket
from tornado.ioloop import IOLoop

connections = []

class WSHandler(websocket.WebSocketHandler):
    def open(self):
        connections.append(self)

    def on_message(self, message):
        ...

    def on_close(self):
        connections.remove(self)


class GetHandler(web.RequestHandler):
    def get(self):
        self.render("chat.html")


async def consumer(channel):
    while await channel.wait_message():
        msg = await channel.get(encoding='utf-8')
        for connection in connections:
            await connection.write_message(msg)


async def setup():
    connection = await aioredis.create_redis('redis://localhost')
    channel = await connection.subscribe('notifications')
    asyncio.ensure_future(consumer(channel))


application = web.Application([
    (r'/', GetHandler),
    (r'/chat/', WSHandler),
])    


if __name__ == '__main__':
    application.listen(8000)
    loop = IOLoop.current()
    loop.add_callback(setup)
    loop.start()

Естественно, это максимально упрощённый пример, в реальном коде соединения не стоит держать в глобальной переменной, а при завершении работы сервера стоит отписаться от канала и закрыть соединение с redis.