-2

我想定义一个函数集群，可以使用给定的初始参数链接和调用，然后得到最终结果，我希望它可以充当 Linux 命令管道链：

                     test01() | test02() | test03() | test04()
[init_parameter]  ---------------------------------------------------->  [final result]

我正在考虑可以添加/减少/重新混合的功能序列，例如：

                     test02() | test03() | test01() | test04()
[init_parameter]  ---------------------------------------------------->  [final result]
                      test03()| test01() | test04() | test01() 
[init_parameter]  ---------------------------------------------------->  [final result]

我还希望这些函数可能带有嵌入的能力指示器，可用于预先检测的智能参数，例如，如果输入不是此函数可以接受的类型或输入超过其最大处理能力，则可以忽略链在这个计算流上而不是使用“try...except...”来捕捉那些“逻辑”错误。

请参阅下面的代码，未经测试，仅用于描述我的想法：

def test01(f_inspect=false, a, **b):
    my_capability = {
        "function_cluster":          "A01", 

        "input_acceptable":          type([]),
        "input_element_acceptable":  type(1),
        "input_length_max":          100,

        "return_type":               type([]),
        "return_element":            type(1),
        "return_length_max":         100,
        }

    if f_inspect:
        return my_capability

    return [x+1 for x in a]           # just sample and may raise error by python such as div 0 


def test02(f_inspect=false, a, **b):
    # similiar as test01

def test03(f_inspect=false, a, **b):
    # similiar as test01

def test04(f_inspect=false, a, **b):
    # similar as test01
#==========================================



#test if function chain is compatible
def f_capability_compatible(current_,next_):
    return True if 
        next_["function_cluster"] == current_["function_cluster"] and
        next_["input_acceptable"] is current_["return_type"]  and
        next_["input_length_max"] >= current_["return_element"]  and
        next_["input_length_max"] >= current_["return_length_max"] 
    return False

foos = [test01,test02,test03,test04]

from itertools import permutations
mypermutations = permutations(foos, 3)    # get permutations of testXX

init_parameter = [x for x in range(1,100)]
dummy_function_parameter = {
        "function_cluster":          "A01", 
        "input_acceptable":          type([]),
        "input_element_acceptable":  type(1),
        "input_length_max":          100,
        "return_type":               type([]),
        "return_element":            type(1)
        "return_length_max":         100,
                          }
chain_flag = [True for x in range(len(mypermutations))]
#[True, True, True, ..... True, True, True, True, True, True, True]

for x in len(mypermutations):
    tmp_para = dummy_function_parameter
    for y in mypermutations[x]:
        test_result = f_capability_compatible(tmp_para,y(f_inspect=true))
        chain_flag[x] = test_result
        tmp_para = y(f_inspect=true)
        if test_result == False :
            print "this can not be chained due to parameter incompatible at position %s" %x 
            break

#==========================================
result_of_chain = []
# to invoke:
for x in len(mypermutations):
    if chain_flag[x] == True:
        try :
            # invoking my_capability[x] chain in a go
            tmp_return = init_parameter
            for f in mypermutations[x]:
                tmp_return = f(tmp_return)  #parameter of testXX(a)
        except :
            result_of_chain[x] = "Error"
    else:
        result_of_chain[x] = "Incomp"

这是我的问题，有没有可能让这个功能链和组合思路更简单？

==================================================== ======================================

更新为什么我需要谓词参数和返回类型：

在 linux 命令行中，我们可以使用如下命令：

$ cat sometfile | grep something | awk '{print $0}' | grep something > file

这是有效的，因为这些命令之间的数据流可以被认为是“文本”类型。

然而，在python中，对于那些未知的函数，输入参数和返回结果基本上有多种类型的可能性。如果我想调用这些函数，我必须知道它的定义。例如

>>> def str_in_asciiout_out(str):
    return ord(str)
>>> 
>>> str_in_asciiout_out("a")
97
>>> str_in_asciiout_out(100)                 
# oops, I have to, try… expect…

Traceback (most recent call last):
  File "<pyshell#3>", line 1, in <module>
    str_in_asciiout_out(100)
  File "<pyshell#0>", line 2, in str_in_asciiout_out
    return ord(str)
TypeError: ord() expected string of length 1, but int found

尝试...除了...是正确和正确的编码方式。

但是如果我想组合数百个 str_in_asciiout_out() 类似的函数并将它们放入一个未知的序列中，我关注的是序列可以在短时间内交付的最佳最终结果。

例如，只是示例假设我定义了 1000 个函数，每个函数可能需要运行一天才能通过给定的输入获得输出，我随机挑选了 200 个函数成一个链，而 str_in_asciiout_out(100) 正好在最后一个位置 by bad幸运的是，我可能会得到一个糟糕的结果，直到浪费了 199 个小时。

这就是为什么我想知道该函数是否可以在浪费时间的调用之前显示它的能力。

上面的代码是我知道的一个丑陋的解决方案，所以我粘贴这个想法，看看是否有更好的解决方案来解决我的问题。

4

1 回答 1

1

我最近看到了一个关于Python 生成器的幻灯片演示，其中介绍了您可以使用生成器函数做的许多巧妙的事情，这些函数允许您像管道和过滤器系统一样使用它们。它们也被“懒惰地”评估，所以当你处理一个很长的列表时，它只会处理第一个列表中它需要的部分，以便为你提供生成器的第一个输出。

看起来您好像是在尝试将静态类型硬塞进 Python 中。虽然存在静态类型的情况，但我不确定以这种方式尝试将其强制转换为应用程序级别的动态语言是否是一个好主意。可以通过在少量输入上测试代码来改进您试图阻止的那种事情

最后，如果您尝试使用有关其返回类型的元数据来注释函数，则最好为该函数使用装饰器。例如，如果您热衷于使用类型，则可以使用来自 Decorators PEP 的示例装饰器：

def attrs(**kwds):
    def decorate(f):
        for k in kwds:
            setattr(f, k, kwds[k])
        return f
    return decorate

@attrs(argument_types=(int, int,),
       returns=int)
def add(a, b):
    return a + b

然后，无需使用f_inspect参数调用函数，您只需访问 的add.argument_types和add.returns成员变量即可add。

于 2011-01-30T06:12:28.473 回答