Обнаружение вложенности контекстного менеджера

Недавно мне было интересно, есть ли способ определить, вложен ли менеджер контекста.

Я создал классы Timer и TimerGroup:

class Timer:
    def __init__(self, name="Timer"):
        self.name = name
        self.start_time = clock()

    @staticmethod
    def seconds_to_str(t):
        return str(timedelta(seconds=t))

    def end(self):
        return clock() - self.start_time

    def print(self, t):
        print(("{0:<" + str(line_width - 18) + "} >> {1}").format(self.name, self.seconds_to_str(t)))

    def __enter__(self):
        return self

    def __exit__(self, exc_type, value, traceback):
        self.print(self.end())


class TimerGroup(Timer):
    def __enter__(self):
        print(('= ' + self.name + ' ').ljust(line_width, '='))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        total_time = self.seconds_to_str(self.end())
        print(" Total: {0}".format(total_time).rjust(line_width, '='))
        print()

Этот код выводит тайминги в удобочитаемом формате:

with TimerGroup("Collecting child documents for %s context" % context_name):
    with Timer("Collecting context features"):
        # some code...
    with Timer("Collecting child documents"):
        # some code...


= Collecting child documents for Global context ============
Collecting context features                >> 0:00:00.001063
Collecting child documents                 >> 0:00:10.611130
====================================== Total: 0:00:10.612292

Однако, когда я вложил TimerGroups, все испортилось:

with TimerGroup("Choosing the best classifier for %s context" % context_name):
    with Timer("Splitting datasets"):
        # some code...
    for cname, cparams in classifiers.items():
        with TimerGroup("%s classifier" % cname):
            with Timer("Training"):
                # some code...
            with Timer("Calculating accuracy on testing set"):
                # some code


= Choosing the best classifier for Global context ==========
Splitting datasets                         >> 0:00:00.002054
= Naive Bayes classifier ===================================
Training                                   >> 0:00:34.184903
Calculating accuracy on testing set        >> 0:05:08.481904
====================================== Total: 0:05:42.666949

====================================== Total: 0:05:42.669078

Все, что мне нужно сделать, это каким-то образом сделать отступ для вложенных Timer и TimerGroups. Должен ли я передавать какие-либо параметры их конструкторам? Или я могу обнаружить это внутри класса?


person QooBooS    schedule 28.06.2017    source источник
comment
Вы можете изменить TimerGroup, чтобы принять другую группу таймеров в качестве родительской группы таймеров, и в каждом экземпляре TimeGroup сохранить отступ (0, если нет родителя, и parent.indentation + 1, если он есть)   -  person Artyer    schedule 28.06.2017


Ответы (3)


Если все, что вам нужно сделать, это отрегулировать уровень отступа в зависимости от того, сколько вложенных менеджеров контекста вы выполняете, тогда используйте атрибут класса с именем indent_level и настраивайте его каждый раз, когда вы входите и выходите из диспетчера контекста. Что-то вроде следующего:

class Context:
    indent_level = 0

    def __init__(self, name):
        self.name = name

    def __enter__(self):
        print(' '*4*self.indent_level + 'Entering ' + self.name)
        self.adjust_indent_level(1)
        return self

    def __exit__(self, *a, **k):
        self.adjust_indent_level(-1)
        print(' '*4*self.indent_level + 'Exiting ' + self.name)

    @classmethod
    def adjust_indent_level(cls, val):
        cls.indent_level += val

И используйте его как:

>>> with Context('Outer') as outer_context:
        with Context('Inner') as inner_context:
            print(' '*inner_context.indent_level*4 + 'In the inner context')


Entering Outer
    Entering Inner
        In the inner context
    Exiting Inner
Exiting Outer
person Billy    schedule 28.06.2017
comment
Ваш код работает только для одного класса таймера. Однако у меня здесь есть Timer и TimerGroup(Timer), и счетчик отступов, кажется, создается для каждого независимо. Есть ли способ создать поле, которое будет общим для всех классов в иерархии? Моя идея состояла в том, чтобы изменить Timer.indent_level в методе adjust_indent_level() и вызывать его только внутри методов __enter__ и __exit__ TimerGroup. - person QooBooS; 28.06.2017
comment
Если вы всегда используете однопоточность, это сработает. Для многопоточного подхода см. ответ Мартейн-Питерс об использовании threading.local. - person 9000; 28.06.2017

Никаких специальных средств для обнаружения вложенных менеджеров контекста нет. Вам придется справиться с этим самостоятельно. Вы можете сделать это в своем собственном менеджере контекста:

import threading


class TimerGroup(Timer):
    _active_group = threading.local()

    def __enter__(self):
        if getattr(TimerGroup._active_group, 'current', False):
            raise RuntimeError("Can't nest TimerGroup context managers")
        TimerGroup._active_group.current = self
        print(('= ' + self.name + ' ').ljust(line_width, '='))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        TimerGroup._active_group.current = None
        total_time = self.seconds_to_str(self.end())
        print(" Total: {0}".format(total_time).rjust(line_width, '='))
        print()

Затем вы можете использовать атрибут TimerGroup._active_group в другом месте, чтобы захватить текущую активную группу. Я использовал локальный объект потока, чтобы убедиться, что это можно использовать в нескольких потоках выполнения.

В качестве альтернативы вы можете сделать это счетчиком стека и просто увеличивать и уменьшать вложенные вызовы __enter__ или стек список и помещать self в этот стек, снова извлекая его, когда вы __exit__:

import threading


class TimerGroup(Timer):
    _active_group = threading.local()

    def __enter__(self):
        if not hasattr(TimerGroup._active_group, 'current'):
            TimerGroup._active_group.current = []
        stack = TimerGroup._active_group.current
        if stack:
            # nested context manager.
            # do something with stack[-1] or stack[0]
        TimerGroup._active_group.current.append(self)

        print(('= ' + self.name + ' ').ljust(line_width, '='))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        last = TimerGroup._active_group.current.pop()
        assert last == self, "Context managers being exited out of order"
        total_time = self.seconds_to_str(self.end())
        print(" Total: {0}".format(total_time).rjust(line_width, '='))
        print()
person Martijn Pieters    schedule 28.06.2017
comment
Естественно также сделать группы вложенными и сделать _active_group list, то есть стек. - person 9000; 28.06.2017

import this:

Явное лучше, чем неявное

Более чистый дизайн позволил бы явно указать группу:

with TimerGroup('Doing big task') as big_task_tg:
    with Timer('Foo', big_task_tg):
      foo_result = foo()
    with Timer('Bar', big_task_tg):
      bar(baz(foo_result))

С другой стороны, вы всегда можете использовать traceback.extract_stack и посмотреть для вызовов конкретной функции восходящего потока. Это очень полезно для ведения журналов и отчетов об ошибках и может быть умеренно полезным для обеспечения того, чтобы определенные функции вызывались только в определенном контексте. Но это имеет тенденцию создавать зависимости, которые очень трудно отследить.

Я бы избегал этого для группировки таймеров, хотя вы можете попробовать. Если вам очень нужна автоматическая группировка, подход @Martijn-Pieters намного лучше.

person 9000    schedule 28.06.2017
comment
Ваша идея довольно хороша для базового использования, однако мне иногда нужно вызывать таймеры из разных методов (один метод вызывает другой и использует таймеры, которые должны быть вложены). Так что ответ @Billy, я думаю, больше подходит для моей проблемы. - person QooBooS; 28.06.2017