一个基于约束传播的微型计算语言的设计和实现

byronhe

发布于 2021-06-25 10:49:07

3090

发布于 2021-06-25 10:49:07

文章被收录于专栏：Tech Explorer

一个基于约束传播的，玩具级微型计算语言的设计和简单实现。

这个程序就是做来玩和练习的，代码是玩具级别的，用的python，基本可以正常工作了。

先介绍应用背景：

在流体机械设计中，通常根据性能参数进行设计，算出其它变量，但问题是，在设计过程中，需要进行变量的手工调整，例如圆整，修正到某一范围，校核等等。

其计算模式举例如下：

1.定义变量，如输入压力Pin=0.98，输入温度Tin=27,输入流量Qvin=400,kv2，φ2r，b2，D2，u2，qin等等。。。

2.根据某些物理公式，算出几个新的量，如转速 n=33.9*sqrt(kv2*φ2r*b2/D2*(u2^3)/qin)

3.把n从8296.93圆整为整数8300，

4.重新计算b2/D2=0.06455，校核可知0.02<0.06455<0.065，符合要求

5.根据n计算出其它新的变量，修正，校核。。。

。。。

观察可以发现，这种计算模式，和《计算机程序的构造与解释》中提到的约束传播系统很像，如果把一个变量看作一个对象，那么，当它位于一个公式的左侧，例如n，也就意味着，右侧变量例如kv2更新时，应该给它发送一个消息，让它重新计算自己的值，当n更新后，如果它发现有别的变量依赖于自己，它应该发消息通知它们更新自己的值。

还可以看出，这种依赖关系形成了一个图，例如应该有一条从kv2到n的边，把n称为kv2的订阅者。

所以这种计算模式可以用约束传播系统建模，但是此处和书里的约束传播系统有差异：此处的约束传播系统是有向图，而书里是无向图，设计成有向图主要是为了简单，无向图的消息发送顺序是难以控制的，而且构造的时候公式中的每个变量都要持有其它对象的引用，太麻烦，有向图只需要在公式左侧的那个变量哪里保存公式右侧的每个变量的引用。

形成有向图后，每个变量的状态设为invaild，这种状态下，不会向它的会订阅者发送更新消息，收到获取值的消息时报错。

有向图中，还有一些源点，是最初给定值的变量。

当某个变量被赋值时，它把自己设为新值，同时向它的订阅者发送更新消息。订阅者计算自己的新值，如果和旧值相同，就沉默；否则，更新自己，通知订阅者更新。

so，想象一下，在你的面前，虚空之中漂浮着一张有向图，由kv2—>n这样的一条条边练成，当一个点被赋予值，从这点荡出一圈圈涟漪，传到它的下一级，再从更新过的点荡出新的波纹，传开，传开。。。直到所有的点都收敛，水面恢复宁静。

好了，说代码，每一个变量都要保存它的订阅者，它的表达式，注意到，数字1.1是一种变量，变量a是一种表达式，所以代码如下：

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109

#!/usr/bin/env python # -*- coding: utf-8 -*- """ 变量is-a表达式数值is-a表达式故有如下继承关系通过env的符号表可以查到Expr的实例 """ __all__ = ['Expr','Var','Number'] class Expr(object): op="" #a function parameters=[] #Expr list desc="" def __init__(self,op,paras,desc=""): self.op=op self.parameters=paras self.desc=desc def get_value(self): pl=[p.get_value() for p in self.parameters] return self.op(*pl) def set_desc(self,d): self.desc=d def dump(self): pas=[] if len(self.parameters): pas=[s.dump() for s in self.parameters] pas.insert(0, '('+self.op.__name__) return ' '.join(pas) + ')' class Number(Expr): value=0.0 def __init__(self,v): self.value=v def get_value(self): return self.value def dump(self): return str(self.value) def update(self): pass class Var(Expr): name="" desc="" expr=None value=0.0 subscribers=[] state="invaild" def __init__(self,name,desc=""): self.name=name self.desc=desc self.state="invaild" def broadcast(self): for s in self.subscribers: s.update() def update(self): self.state="normal" new_value=self.expr.get_value() if new_value == self.value: return self.value=new_value self.broadcast() def set_expr(self,exp): self.expr=exp if isinstance(exp,Number): self.update() def set_value(self,v): self.value=v self.state="normal" self.broadcast() def get_value(self): if self.state=="invaild": self.update() assert self.state=="normal" return self.value def subscribe(self,subs): for sub in subs: self.subscribers.append(sub) def dump(self): expr_d="" if self.expr: expr_d=' '+self.expr.dump() return str(self.name) +"="+str(self.value)+expr_d#+" "+self.desc def test(): a=Var("a","变量a") b=Var("b","变量b") if __name__=="__main__": test()

所有的变量当然是要保存到一个符号表（或称环境）里的，同时，这个环境里还要有加减乘除，sin，sqrt这样的基本运算的定义，pi，e这样的常数的定义，python的operator和math模块就够用了。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121

#!/usr/bin/env python # -*- coding: utf-8 -*- import math import operator from expr import Var,Number,Expr from parser import Parser class CmdFilter(object): def __init__(self,env,input_file): self.env=env self.input_file=input_file def readline(self): while True: s=self.input_file.readline() if not s: return s if self.env.filter_cmd(s): return s class Env(object): """ 求值环境，提供变量符号表和函数符号表 """ symbol_table={} #存放变量 expr_table=[] #存放自由表达式 function_table={}#存放函数，对外只读 cmds=['dump','run'] #env先于parser处理掉一些命令,如dump parser=None def __init__(self): self.fill_function_table() self.fill_symbol_table() self.parser=Parser(self) def dump(self): print "-"*70,"\ndump all variables and expressions:" for k,v in self.symbol_table.items(): print k+":",v.get_value() print "\nall checkings:" for e in self.expr_table: print e.get_value(),"=",e.dump() print "-"*70 def run(self): for k,v in self.symbol_table.items(): v.update() def fill_function_table(self): #算术运算符 #1.+,-,*,/,^,=,(,) 算术运算符 self.function_table['+']=operator.add self.function_table['-']=operator.sub self.function_table['*']=operator.mul self.function_table['/']=operator.div self.function_table['^']=operator.pow #逻辑运算符 #2.==,>=,>,<=,<,!= 逻辑运算符 self.function_table['==']=operator.eq self.function_table['>=']=operator.ge self.function_table['>']=operator.gt self.function_table['<=']=operator.le self.function_table['<']=operator.lt self.function_table['!=']=operator.ne self.function_table['sqrt']=math.sqrt self.function_table['sin']=math.sin self.function_table['cos']=math.cos self.function_table['tan']=math.tan self.function_table['asin']=math.asin self.function_table['acos']=math.acos self.function_table['atan']=math.atan self.function_table['exp']=math.exp self.function_table['pow']=math.pow self.function_table['factorial']=math.factorial self.function_table['fabs']=math.fabs self.function_table['ln']=math.log self.function_table['log10']=math.log10 def fill_symbol_table(self): self.symbol_table['pi']=Number(math.pi) self.symbol_table['e'] =Number(math.e) def add_expr(self,e): if e: self.expr_table.append(e) def get_function(self,name): if self.function_table.has_key(name): return self.function_table[name] else: return None def get_variable(self,name): if self.symbol_table.has_key(name): return self.symbol_table[name] else: return None def set_variable(self,name,var): self.symbol_table[name]=var def filter_cmd(self,s): s=s.strip() if s in self.cmds: fun=getattr(self,s) fun() return None return s def exec_stream(self,in_file): input_file=CmdFilter(self,in_file) self.parser.parse(input_file) import sys def test(): env=Env() env.exec_stream(sys.stdin) if __name__=="__main__": test()

接下来，词法分析和语法分析，词法分析没有手写，也没有用flex那样的工具，直接用了一排正则表达式，挨个往下匹配，匹配上了就返回。

严格来说，这个是不太好的，此处的词法分析原理上是不能和flex比的，flex里的多个正则表达式是合并到一个NFA里，再转化成一个DFA的，所以它的规则首先是最长优先，其次是长度相同时写在前面的优先，此处只有写在前面的优先，不太好。

语法分析是递归下降文法分析。一行一行地分析，一行是一个token的list。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308

#!/usr/bin/env python # -*- coding: utf-8 -*- from math import * from operator import * import re from expr import Var,Number,Expr """ 把 a=1+b+c^2 c=2 b=c+2 这样的一行一行，分成 ID ASSIGN ADD ID ADD EQ POW 2 EOL ID ASSIGN 2 EOL ID ASSIGN ID ADD 2 EOL 这样的一行一行token, 输出是一行一个token list的形式 token分为如下几类: 1.+,-,*,/,^,=,(,),, 算术运算符 2.==,>=,>,<=,<,!= 逻辑运算符 3.[0-9]+\.[0-9]*[Ee][+-]?[0-9]+ [0-9]+ 字面浮点数和整数 4.[a-zA-Z_]+ 变量或函数名称标识符 5.[ \\t\\n] 忽略，或结束由于使用正则表达式直接匹配，所以和flex不同的是: 无法确保当有多个匹配项时，最长优先,因此，只能利用先后顺序解决冲突，因而必须把==放在=前面。 """ logic_op_re=re.compile(r'==|>=|>|<=|<|!=') number_re =re.compile(r'[+-]?[0-9]+\.?[0-9]*[Ee]?[+-]?[0-9]?') arith_op_re=re.compile(r'\+|-|\*|/|\^|=|\(|\)|,') int_re =re.compile(r'[0-9]+') id_re =re.compile(r'[a-zA-Z_]+') blank_re =re.compile(r'[\ |\t|\r]+') comment_re =re.compile(r'"([^"]*)"') other_re =re.compile(r'.+') def scanner(line): result=[] while True: line=line.strip() if not line: return result m=re.match(logic_op_re,line) if m: result.append(('logic_op',m.group())) line=line[m.end():] continue m=re.match(number_re ,line) if m: result.append(('number',float(m.group()))) line=line[m.end():] continue m=re.match(arith_op_re,line) if m: result.append(('arith_op',m.group())) line=line[m.end():] continue m=re.match(int_re ,line) if m: result.append(('number',float(m.group()))) line=line[m.end():] continue m=re.match(id_re ,line) if m: result.append(('id',m.group())) line=line[m.end():] continue m=re.match(comment_re ,line) if m: result.append(('comment',m.group())) line=line[m.end():] continue m=re.match(blank_re ,line) if m: line=line[m.end():] continue m=re.match(other_re,line) if m: print "亲，看看你都塞了一堆什么进来呀？\""+m.group()+"\" 人家好伤心的呀！" line=line[m.end():] return result class Parser(object): """ 文法分析： """ input_file=None env=None def __init__(self,env): self.env=env def parse(self,input_file): """ 如入可以是sys.stdin,a file,a string 要求实现readline()方法 """ self.input_file=input_file self.run() def run(self): while True: line=self.input_file.readline() if not line: return tokens=scanner(line) #把字母名称的函数标示出来 r=[] for t in tokens: if t[0]=='id' and self.env.get_function(t[1]): r.append(('function',t[1])) else: r.append(t) tokens=r #把comment提取出来 comments=map(lambda x:x[1], filter(lambda x:x[0]=="comment",tokens)) comments=' '.join(comments) tokens=filter(lambda x:x[0]!="comment",tokens) #含有=的表达式是约束方程，其它的都是expr c=tokens.count( ('arith_op', '=')) if c==0: #没有约束到变量的表达式 e=self.parse_expr(tokens) #e.set_desc(comments) self.env.add_expr(e) continue if c>1: print "亲，赋值一次就够了，你肿么赋值了"+str(c)+"次涅？" continue #c=1 if len(tokens)<3 or tokens[0][0]!='id' or \ tokens[1]!=('arith_op','='): print "亲，你给我的表达式格式有问题偶～:"+line continue var_name=tokens[0][1] var=self.env.get_variable(var_name) if var is None: var=Var(var_name,comments) self.env.set_variable(var_name,var) e=self.parse_expr(tokens[2:]) var.set_expr(e) def parse_expr(self,tokens): """ token分为如下几类: 1.+,-,*,/,^,=,(,),, 算术运算符 2.==,>=,>,<=,<,!= 逻辑运算符 3.[0-9]+\.[0-9]*[Ee][+-]?[0-9]+ [0-9]+ 字面浮点数和整数 4.[a-zA-Z_]+ 变量或函数名称标识符 5.[ \\t\\n] 忽略，或结束 BNF: expr=expr[==|>=|>|<=|<|!=]expr|expr expr=expr+expr | expr-expr expr=expr*expr | expr/expr expr=expr^expr expr=function(expr[,expr]) expr=(expr) expr=<float>|<var> """ if len(tokens): expr,rest=self.parse_logic_op(tokens) return expr #能处理就处理，不能处理原样返回。 def parse_logic_op(self,tokens): left,rest=self.parse_add_sub_op(tokens) if not len(rest): return left,rest logic_op_list=["==",">=",">","<=","<","!="] if rest[0][1] not in logic_op_list: return left,rest op=self.env.get_function(rest[0][1]) right,rest=self.parse_add_sub_op(rest[1:]) return Expr(op,[left,right]),rest def parse_add_sub_op(self,tokens): left,rest=self.parse_mul_div_op(tokens) add_sub_op_list=["+","-"] while len(rest) and rest[0][1] in add_sub_op_list: op=self.env.get_function(rest[0][1]) right,rest=self.parse_mul_div_op(rest[1:]) left=Expr(op,[left,right]) return left,rest def parse_mul_div_op(self,tokens): left,rest=self.parse_pow_op(tokens) mul_div_op_list=["*","/"] while len(rest) and rest[0][1] in mul_div_op_list: op=self.env.get_function(rest[0][1]) right,rest=self.parse_pow_op(rest[1:]) left=Expr(op,[left,right]) return left,rest def parse_pow_op(self,tokens): left,rest=self.parse_function_op(tokens) pow_op_list=["^"] while len(rest) and (rest[0][1] in pow_op_list): op=self.env.get_function(rest[0][1]) right,rest=self.parse_pow_op(rest[1:]) left=Expr(op,[left,right]) return left,rest def parse_function_op(self,tokens): if tokens[0][0] in ['number','id']: return self.parse_float_var_op(tokens) if tokens[0][1]=='(': return self.parse_parentheses_op(tokens) if tokens[0][0]!='function': return None,tokens op=self.env.get_function(tokens[0][1]) if op and tokens[1][1]=='(': paras=[] tokens=tokens[2:] left,tokens=self.parse_add_sub_op(tokens) paras.append(left) while tokens[0][1]==',': left,tokens=self.parse_add_sub_op(tokens[1:]) paras.append(left) if tokens[0][1]==')': tokens=tokens[1:] else: print "bad syntax. tokens found ->",tokens expr=Expr(op,paras) return expr,tokens else: print "error bad syntax ->",tokens return None,tokens def parse_parentheses_op(self,tokens): if tokens[0][1]=='(': left,tokens=self.parse_logic_op(tokens[1:]) if tokens[0][1]==')': return left,tokens[1:] return left,tokens return None,tokens def parse_float_var_op(self,tokens): if tokens[0][0] == 'number': n=Number(tokens[0][1]) return n,tokens[1:] if tokens[0][0] == 'id': var=self.env.get_variable(tokens[0][1]) if not var: var_name=tokens[0][1] var=Var(var_name,'') self.env.set_variable(var_name,var) var=self.env.get_variable(tokens[0][1]) return var,tokens[1:] return None,tokens import StringIO from env import * def test(): s=""" a=1+(b+c)^2/23.1e-10 "变量a" c=2 "变量c" "c是个好变量" b=c+2 "变量b" "b也是个好变量" "这是为它写的第三条注释" a>c "检查a是否大于c" a>=c "检查a是否大于等于c" run dump c=3 "change c again." "注释也可以在前面" c^2>=sin(pow(a,b)+b) run dump """ #for l in s.split('\n'): # scanner(l) print "+"*80 print '*'*70 print s print '*'*70 e=Env() i=StringIO.StringIO(s) e.exec_stream(i) print "+"*80 if __name__=="__main__": test()

最后是个main.py