calculate-utils-3-lib/pym/calculate/lib/utils/tools.py

#-*- coding: utf-8 -*-

# Copyright 2014 Calculate Ltd. http://www.calculate-linux.org
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
from functools import wraps

from itertools import tee
from contextlib import contextmanager

class SavableIterator:
    """
    Итератор с возможность сохранять и восстанавливать состояние
    """
    def __init__(self,seq):
        self.seq = iter(seq)
        self.back = None

    def __iter__(self):
        return self

    def save(self):
        self.seq, self.back = tee(self.seq)
        return self

    def restore(self):
        if self.back:
            self.seq, self.back = tee(self.back)
        return self

    def next(self):
        return self.seq.next()

@contextmanager
def ignore(exception):
    try:
        yield
    except exception:
        pass

class AddonError(Exception):
    """
    Исключение с добавочным сообщением
    """

    def __init__(self, msg, addon=None):
        self.message = msg
        self.addon = addon
        Exception.__init__(self, msg)


class FalseObject(object):
    """
    Объект-заглушка при любом сравнении возвращает False
    """
    def __lt__(self,other):
        return False

    def __nonzero__(self):
        return False

    __gt__ = __ge__ = __le__ = __eq__ = __ne__ = __lt__


class Sizes(object):
    K = KiB = kibibyte = 1024
    kB = kilobyte = 1000
    M = MiB = mibibyte = K * 1024
    Mb = megabyte = kB * 1000
    G = GiB = gibibyte = M * 1024
    Gb = gigabyte = Mb * 1000
    T = TiB = tibibyte = G * 1024
    Tb = terabyte = Gb * 1000

    def __getattr__(self,name):
        if name.startswith('from_'):
            return lambda x:x*getattr(Sizes,name[5:])
        elif name.startswith('to_'):
            return lambda x:x/getattr(Sizes,name[3:])
        else:
            raise AttributeError

def imap_regexp(re_compiled, l, whole=False):
    """
    Обработать список регулярным выражением и вернуть полученные группы
    """
    if whole:
        retfunc = lambda x: x.group()
    else:
        retfunc = lambda x: x.groups()
    return (retfunc(x) for x in (re_compiled.search(x) for x in l) if x)

def cached(each_instance=False):
    """
    Кэширование результата
    """
    def cache_wrapper(func):
        value = {}

        def wrapper(*args, **kwargs):
            if each_instance:
                if args[0] not in value:
                    value[args[0]] = func(*args, **kwargs)
                return value[args[0]]
            else:
                if not None in value:
                    value[None] = func(*args, **kwargs)
                return value[None]
        return wrapper
    if each_instance in (True,False):
        return cache_wrapper
    else:
        return cache_wrapper(each_instance)


class Singleton(type):
    _instances = {}

    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super(Singleton, cls).__call__(*args,
                                                                 **kwargs)
        return cls._instances[cls]


class ClassifyingIterator(object):
    """
    Классифицирующий итератор
    """

def classificate(iterator):
    """
    Классифицировать элементы по признаку (first первый last последний)
    """
    class Mark:
        def __init__(self, first=False, last=False):
            self.first = first
            self.last = last

        def __repr__(self):
            return "Mark(first=%s,last=%s)"%(self.first,self.last)

    iterator = iter(iterator)
    obj = iterator.next()
    try:
        obj_next = iterator.next()
        yield Mark(first=True,last=False), obj
        try:
            while True:
                obj = obj_next
                obj_next = iterator.next()
                yield Mark(), obj
        except StopIteration:
            yield Mark(first=False,last=True), obj
    except StopIteration:
        yield Mark(first=True,last=True), obj


def debug(prefix="DEBUG {name}({args},{kw})={ret}"):
    def debug_decor(f):
        @wraps(f)
        def wrapper(*args, **kw):
            ret = f(*args, **kw)
            print prefix.format(name=f.func_name, args=args, kw=kw, ret=ret)
            return ret
        return wrapper
    return debug_decor