装饰器对代码的特定行而不是整个方法进行计时？答案

【问题标题】：Decorator to time specific lines of the code instead of whole method?装饰器对代码的特定行而不是整个方法进行计时？
【发布时间】：2015-05-25 08:17:01
【问题描述】：

让我们假设一个简单的方法：

def test_method():
    a = 1
    b = 10000
    c = 20000
    sum1 = sum(range(a,b))
    sum2 = sum(range(b,c))
    return (sum1,sum2)

要使用装饰器对这个方法计时，一个简单的装饰器应该是：

from functools import wraps
def timed_decorator(f):
    @wraps(f)
    def wrapper(*args, **kwds):
        start = time.time()
        result = f(*args, **kwds)
        elapsed = (time.time() - start)*1000
        logger.debug("f::{0} t::{1:0.2f} ms".format(f.__name__, elapsed))
        return result
    return wrapper

现在，如果我想为 test_method 的特定行计时，比如第 4 行 sum1 = sum(range(a,b)) ，当前的实现涉及内联编码，例如：

 def test_method():
        a = 1
        b = 10000
        c = 20000
        start = time.time()
        sum1 = sum(range(a,b)) # timing specific line or lines
        elapsed = (time.time() - start)*1000
        logger.debug("This part took::{1:0.2f} ms".format(elapsed))
        sum2 = sum(range(b,c))
        return (sum1,sum2)

目的是在不修改方法中的代码的情况下，使用装饰器对特定方法的 M 到 N 行进行时间处理。是否可以使用装饰器注入这样的逻辑？

【问题讨论】：

标签： python

【解决方案1】：

您可以使用上下文管理器。

import contextlib

@contextlib.contextmanager
def time_measure(ident):
    tstart = time.time()
    yield
    elapsed = time.time() - tstart
    logger.debug("{0}: {1} ms".format(ident, elapsed))

在您的代码中，您可以像使用它一样

with time_measure('test_method:sum1'):
    sum1 = sum(range(a, b))

顺便说一句，如果你想改进你的代码，你可以使用高斯和公式（解释为here）而不是sum(range(a, b))。

def sum_range(a, b):
    r_a = (a ** 2 + a) / 2 - a
    r_b = (b ** 2 + b) / 2 - b
    return r_b - r_a

【讨论】：

我正在寻找在不修改原始方法的情况下对这些行进行计时的方法。
如果不修改原始代码，您将无法做到这一点。从逻辑上判断，只有当它是一个完整的单元时，您才能在外部对代码进行计时，例如module、function 等。至于只有几行代码，@Niklas R 提供了一种非常不错的 Pythonic 方式来实现您的目标.

【解决方案2】：

我能想到的一种方法是在跟踪函数中处理“line”事件时使用 sys.settrace() 并记录时间。但需要注意的是，设置跟踪器的做法可能会导致记录的时间不准确。

总体思路是：

在包装目标方法的装饰器中设置跟踪函数。
获取该方法第一行的行号，用FLN = inspect.currentframe().f_lineno.
在跟踪函数中，处理“调用”事件并返回本地跟踪函数以跟踪范围内的“线路”事件。 Read this 如果您感到困惑。
在本地跟踪函数中，获取当前行号LN，如果LN-FLN == M，记录开始时间；如果LN-FLN == N，记录结束时间，执行M到N行的时间为endtime - starttime。

代码：

import sys
from functools import wraps
import time
import linecache

_func_name_ = None
_func_ln_ = 0
_start_ = 0
_end_ = 0
_timestamp_ = 0

def trace_calls(frame, event, arg): 
    global _func_name_, _func_ln_
    def trace_lines(frame, event, arg): 
        global _timestamp_
            if event != 'line':
                return
        line_no = frame.f_lineno
        filename = frame.f_code.co_filename
        if line_no-_func_ln_ == _start_:                        
            _timestamp_ = time.time()
            print "%d %s TS:%d"%(line_no, linecache.getline(filename, line_no)[:-1], _timestamp_)

        elif line_no-_func_ln_ == _end_:
            _timestamp_ = time.time() - _timestamp_
            print "%d %s"%(line_no, linecache.getline(filename, line_no)[:-1])
            print "Lines %d to %d of %s takes %d seconds."%(_start_, _end_, _func_name_,  _timestamp_)      

    if event != 'call':    
        return   

    co = frame.f_code      
    _func_ln_ = frame.f_lineno  # record the line number at function entry point
    func_name = co.co_name 

    if func_name != _func_name_:
        return             
    return trace_lines

def time_lines(start, end):
    global _start_, _end_
    _start_, _end_ = start+1, end+2     # function name takes a line, end is inclusive
    def inner(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            global _func_name_
            _func_name_ = f.__name__    
            sys.settrace(trace_calls)
            f(*args, **kwargs)
            sys.settrace(None)
        return wrapper
    return inner

@time_lines(2,4)
def tested_func():
    print "Enter target function"
    time.sleep(2)
    time.sleep(1)
    time.sleep(3)
    print "Exit target function"

if __name__=="__main__":
    tested_func()

【讨论】：

【解决方案3】：

它很丑陋，而且不是很稳定的代码。但我发现执行此任务的唯一方法是在注入代码后再次执行该函数的代码。
像这样的：

import inspect
import re
import time

def inject_timer(f,n,m):
    codelines = inspect.getsourcelines(f)[0]
    ident_lvl = re.search("^[ \t]*",codelines[n]).group(0)
    codelines.insert(n,ident_lvl + "start_longJibrishTo_preventCollision = time.time()\n")
    codelines.insert(m+2,ident_lvl + "elapsed_longJibrishTo_preventCollision = (time.time() - start_longJibrishTo_preventCollision)*1000\n")
    codelines.insert(m+3,ident_lvl + """print("f::{0} t::{1:0.2f} ms".format("""+f.__name__+""", elapsed_longJibrishTo_preventCollision))\n""")
    #print "".join(codelines)
    exec "".join(codelines) in globals()

def test_method():
    a = 1
    b = 10000
    time.sleep(2)
    c = 20000    
    sum1 = sum(range(a,b))
    sum2 = sum(range(b,c))    
    return (sum1,sum2)

inject_timer(test_method,3,5)

【讨论】：

【解决方案4】：

装饰器只能装饰可调用对象（例如函数、方法、类）。只要您不将它们包装在自己的可调用对象中，单行或一组行就不可调用。

要对代码单元进行计时，您应该选择适当的重复次数。目标是确保执行时间不只是几微秒或几毫秒，否则测量误差会太大。

您看过timeit 模块吗？

【讨论】：

同意callables，但是有没有办法访问装饰器内部的可调用行M到N？
不，没有指定单元 U 的“行 N 到 M”的机制。Python 解释器不跟踪单个代码单元内的行数。这将是巨大的开销。

【解决方案5】：

带有自定义上下文管理器的非常简单的解决方案：

class elapsed:
    def __enter__(self): self.start = time.time()
    def __exit__(self, *args): print("%.1f ms" % ((time.time() - self.start)*1000))

示例用法：

with elapsed():
    sum1 = sum(x ** 2 for x in range(1, 1000000))
# 547.0 ms

另一种解决方案：这里是@NiklasR 的答案略有不同，没有logger 但print，以及一个准备运行的示例：

import contextlib, time

@contextlib.contextmanager
def time_measure(ident):
    tstart = time.time()
    yield
    elapsed = time.time() - tstart
    print("{0}: {1} ms".format(ident, elapsed))

with time_measure('hello'):
    sum1 = sum(x ** 2 for x in range(1, 1000000))

# hello: 0.577033281326294 ms

【讨论】：