model.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533
  1. import sys
  2. sys.path.append("../kernel/")
  3. sys.path.append("../state/")
  4. from modelverse_state.main import ModelverseState as MvS
  5. from modelverse_kernel.main import ModelverseKernel as MvK
  6. from pypdevs.DEVS import AtomicDEVS, CoupledDEVS
  7. from pypdevs.simulator import Simulator
  8. import json
  9. def get_object_constructor(code):
  10. with open(".code.alc", "w") as f:
  11. f.write(code)
  12. f.flush()
  13. constructors = do_compile(".code.alc", "interface/HUTN/grammars/actionlanguage.g", "CD")
  14. return constructors
  15. def translate(operation):
  16. return {
  17. "CN": "create_node",
  18. "CE": "create_edge",
  19. "CNV": "create_nodevalue",
  20. "CD": "create_dict",
  21. "RV": "read_value",
  22. "RO": "read_outgoing",
  23. "RI": "read_incoming",
  24. "RE": "read_edge",
  25. "RD": "read_dict",
  26. "RDN": "read_dict_node",
  27. "RDNE": "read_dict_node_edge",
  28. "RDE": "read_dict_edge",
  29. "RRD": "read_reverse_dict",
  30. "RR": "read_root",
  31. "RDK": "read_dict_keys",
  32. "DE": "delete_edge",
  33. "DN": "delete_node",
  34. }[operation]
  35. class MvSState(object):
  36. def __init__(self):
  37. self.queue = []
  38. self.output = None
  39. self.mvs = MvS("../bootstrap/bootstrap.m")
  40. self.timer = float("inf")
  41. class ModelverseState(AtomicDEVS):
  42. def __init__(self,
  43. read_root,
  44. create_node,
  45. create_edge,
  46. create_nodevalue,
  47. create_dict,
  48. read_value,
  49. read_outgoing,
  50. read_incoming,
  51. read_edge,
  52. read_dict,
  53. read_dict_keys,
  54. read_dict_edge,
  55. read_dict_node,
  56. read_dict_node_edge,
  57. read_reverse_dict,
  58. delete_node,
  59. delete_edge):
  60. AtomicDEVS.__init__(self, "MvS")
  61. self.timings = {
  62. "read_root": read_root,
  63. "create_node": create_node,
  64. "create_edge": create_edge,
  65. "create_nodevalue": create_nodevalue,
  66. "create_dict": create_dict,
  67. "read_value": read_value,
  68. "read_outgoing": read_outgoing,
  69. "read_incoming": read_incoming,
  70. "read_edge": read_edge,
  71. "read_dict": read_dict,
  72. "read_dict_keys": read_dict_keys,
  73. "read_dict_edge": read_dict_edge,
  74. "read_dict_node": read_dict_node,
  75. "read_dict_node_edge": read_dict_node_edge,
  76. "read_reverse_dict": read_reverse_dict,
  77. "delete_node": delete_node,
  78. "delete_edge": delete_edge,
  79. }
  80. self.state = MvSState()
  81. self.from_mvk = self.addInPort("from_MvK")
  82. self.to_mvk = self.addOutPort("to_MvK")
  83. def extTransition(self, inputs):
  84. self.state.timer -= self.elapsed
  85. empty = len(self.state.queue) == 0
  86. self.state.queue.extend(inputs[self.from_mvk])
  87. if empty:
  88. # First message, so set the timer
  89. # And already compute the result so it is ready to output
  90. self.state.output = []
  91. self.state.timer = 0.0
  92. for v in self.state.queue[0]:
  93. self.state.output.append(getattr(self.state.mvs, translate(v[0]))(*v[1]))
  94. self.state.timer += self.timings[translate(v[0])]()
  95. else:
  96. # Just append the message to process
  97. pass
  98. return self.state
  99. def outputFnc(self):
  100. return {self.to_mvk: [self.state.output]}
  101. def intTransition(self):
  102. self.state.queue.pop(0)
  103. self.state.output = []
  104. if len(self.state.queue) > 0:
  105. self.state.timer = 0.0
  106. # Value contains a list of operations to do
  107. # So do them and calculate how long it takes
  108. for v in self.state.queue[0]:
  109. self.state.output.append(getattr(self.state.mvs, translate(v[0]))(*v[1])[0])
  110. self.state.timer += self.timings[translate(v[0])]()
  111. else:
  112. self.state.timer = float("inf")
  113. return self.state
  114. def timeAdvance(self):
  115. return self.state.timer
  116. class MvKState(object):
  117. def __init__(self):
  118. self.mvk = None
  119. self.waiting = False
  120. self.inputs = {}
  121. self.outputs = {}
  122. self.users = []
  123. self.reply = None
  124. self.phase = None
  125. self.commands = None
  126. self.root = None
  127. self.current_user = None
  128. self.loaded_primitives = False
  129. self.execution_counter = 0
  130. self.all_failed = True
  131. def __str__(self):
  132. return "\nMvK: %s\n" % self.mvk + \
  133. "waiting: %s\n" % self.waiting + \
  134. "inputs: %s\n" % self.inputs + \
  135. "outputs: %s\n" % self.outputs + \
  136. "users: %s\n" % self.users + \
  137. "reply: %s\n" % self.reply + \
  138. "phase: %s\n" % self.phase + \
  139. "commands: %s\n" % self.commands + \
  140. "root: %s\n" % self.root + \
  141. "current user: %s\n" % self.current_user + \
  142. "execution counter: %s\n"
  143. class ModelverseKernel(AtomicDEVS):
  144. def __init__(self, rules_per_phase):
  145. AtomicDEVS.__init__(self, "MvK")
  146. self.state = MvKState()
  147. self.from_mvi = self.addInPort("from_MvI")
  148. self.from_mvs = self.addInPort("from_MvS")
  149. self.to_mvi = self.addOutPort("to_MvI")
  150. self.to_mvs = self.addOutPort("to_MvS")
  151. self.rules_per_phase = rules_per_phase
  152. def extTransition(self, inputs):
  153. if self.from_mvi in inputs:
  154. # Got input from MvI, so we queue it
  155. for inp in inputs[self.from_mvi]:
  156. username = inp[0]
  157. data = inp[1]
  158. if data is not None:
  159. self.state.inputs.setdefault(username, []).extend(data)
  160. else:
  161. self.state.outputs.setdefault(username, []).append(None)
  162. if self.from_mvs in inputs:
  163. # Got input from MvS, so we can continue processing
  164. for mvs_input in inputs[self.from_mvs]:
  165. mvs_stripped = [i[0] for i in mvs_input]
  166. if self.state.mvk is None:
  167. # No MvK, so set it with the root we have just received (or should have received)
  168. self.state.root = mvs_stripped[0]
  169. self.state.mvk = MvK(self.state.root)
  170. else:
  171. self.state.reply = mvs_stripped
  172. self.state.waiting = False
  173. return self.state
  174. def intTransition(self):
  175. was_empty = len(self.state.users) == 0
  176. if self.state.commands is not None:
  177. self.state.commands = None
  178. return self.state
  179. if self.state.mvk is None:
  180. # Initializing
  181. self.state.waiting = True
  182. elif not self.state.loaded_primitives:
  183. commands = self.state.mvk.execute_yields("", "load_primitives", [], self.state.reply)
  184. if commands is None:
  185. self.state.loaded_primitives = True
  186. self.state.reply = None
  187. else:
  188. self.state.waiting = True
  189. self.state.commands = commands
  190. else:
  191. # Are initialized and have work to do
  192. if len(self.state.users) == 0:
  193. # Read out new set of users first
  194. if self.state.reply is None:
  195. commands = [("RDK", [self.state.root])]
  196. else:
  197. self.state.users = self.state.reply[0]
  198. commands = None
  199. elif self.state.phase == "init_user":
  200. if self.state.reply is None:
  201. commands = [("RV", [self.state.users[0]])]
  202. else:
  203. self.state.current_user = self.state.reply[0]
  204. if self.state.current_user.startswith("__"):
  205. # Don't process this user and force termination of user
  206. self.state.phase = "output"
  207. commands = None
  208. elif self.state.phase == "input":
  209. # Process inputs
  210. if self.state.inputs.get(self.state.current_user, None):
  211. element_type, value = self.state.inputs[self.state.current_user][0]
  212. commands = self.state.mvk.execute_yields(self.state.current_user, "set_input", [element_type, value], self.state.reply)
  213. if commands is None:
  214. self.state.inputs[self.state.current_user].pop(0)
  215. else:
  216. commands = None
  217. elif self.state.phase == "computation":
  218. commands = self.state.mvk.execute_yields(self.state.current_user, "execute_rule", [], self.state.reply)
  219. elif self.state.phase == "output":
  220. commands = self.state.mvk.execute_yields(self.state.current_user, "get_output", [], self.state.reply)
  221. else:
  222. raise Exception("Phase: " + str(self.state.phase))
  223. # Advance phase
  224. if commands is None:
  225. if was_empty:
  226. self.state.phase = "init_user"
  227. elif self.state.phase == "init_user":
  228. self.state.phase = "input"
  229. elif self.state.phase == "input":
  230. self.state.phase = "computation"
  231. elif self.state.phase == "computation":
  232. if not self.state.mvk.success or (self.state.execution_counter > self.rules_per_phase):
  233. self.state.phase = "output"
  234. self.state.execution_counter = 0
  235. else:
  236. self.state.execution_counter += 1
  237. elif self.state.phase == "output":
  238. self.state.users.pop(0)
  239. self.state.phase = "init_user"
  240. self.state.waiting = False
  241. self.state.reply = None
  242. else:
  243. self.state.waiting = True
  244. # Send the commands to the MvS
  245. self.state.commands = commands
  246. return self.state
  247. def outputFnc(self):
  248. if self.state.mvk is None:
  249. # Ask the root first
  250. return {self.to_mvs: [[("RR", [])]]}
  251. elif self.state.waiting:
  252. return {self.to_mvs: [self.state.commands]}
  253. return {}
  254. def timeAdvance(self):
  255. if self.state.phase == "init_user" and self.state.all_failed:
  256. # Make this a parameter
  257. return 200
  258. if self.state.commands is not None:
  259. return 0
  260. elif self.state.waiting:
  261. return float("inf")
  262. elif self.state.mvk is None:
  263. return 0
  264. else:
  265. return 0
  266. class MvIState():
  267. def __init__(self):
  268. self.operations = []
  269. self.output = []
  270. self.processing = []
  271. self.memory = {}
  272. self.init = True
  273. class ModelverseInterface(AtomicDEVS):
  274. def __init__(self, username, operations):
  275. AtomicDEVS.__init__(self, "MvI_%s" % username)
  276. self.state = MvIState()
  277. self.state.operations = operations
  278. self.username = username
  279. self.to_mvk = self.addOutPort("to_MvK")
  280. self.from_mvk = self.addInPort("from_MvK")
  281. def intTransition(self):
  282. self.state.init = False
  283. while self.state.operations:
  284. i = self.state.operations[0]
  285. if isinstance(i, int) and i not in self.memory:
  286. break
  287. self.state.operations.pop(0)
  288. return self.state
  289. def extTransition(self, inputs):
  290. for inp in inputs[self.from_mvk]:
  291. if inp["value"] == "None" and isinstance(self.state.processing[0], int) and self.state.processing[0] not in self.memory:
  292. self.state.memory[self.state.processing.pop(0)] = int(inp["id"])
  293. else:
  294. self.state.output.append(inp)
  295. return self.state
  296. def outputFnc(self):
  297. send = []
  298. for i in self.state.operations:
  299. if isinstance(i, int) and i not in self.memory:
  300. break
  301. elif isinstance(i, int) and i in self.memory:
  302. # Pass a reference!
  303. send.append(("R", self.memory[i]))
  304. elif not isinstance(i, int):
  305. send.append(("V", i))
  306. return {self.to_mvk: [(self.username, send)]}
  307. def timeAdvance(self):
  308. if self.state.init:
  309. return 0
  310. elif self.state.processing and (not isinstance(self.state.processing[0], int) or self.state.processing[0] in self.memory):
  311. return 0
  312. else:
  313. return float("inf")
  314. class NetworkState(object):
  315. def __init__(self):
  316. self.processing = []
  317. self.timer = float("inf")
  318. class Network(AtomicDEVS):
  319. def __init__(self, name, latency, bandwidth):
  320. AtomicDEVS.__init__(self, name)
  321. self.state = NetworkState()
  322. self.input_port = self.addInPort("input_port")
  323. self.output_port = self.addOutPort("output_port")
  324. self.latency = latency
  325. self.bandwidth = bandwidth
  326. def intTransition(self):
  327. self.state.processing.pop(0)
  328. if self.state.processing:
  329. self.state.timer = int(len(self.state.processing[0]) / float(self.bandwidth) + self.latency)
  330. else:
  331. self.state.timer = float("inf")
  332. return self.state
  333. def extTransition(self, inputs):
  334. self.state.timer -= self.elapsed
  335. if self.state.timer == float("inf"):
  336. self.state.timer = 0
  337. for v in inputs[self.input_port]:
  338. self.state.processing.append(json.dumps(v))
  339. if len(self.state.processing) > 0:
  340. self.state.timer = int(len(self.state.processing[0]) / float(self.bandwidth) + self.latency)
  341. return self.state
  342. def outputFnc(self):
  343. return {self.output_port: [json.loads(self.state.processing[0])]}
  344. def timeAdvance(self):
  345. return self.state.timer
  346. class System(CoupledDEVS):
  347. def __init__(self,
  348. username,
  349. operations,
  350. rules_per_phase,
  351. mvi2mvk_latency,
  352. mvi2mvk_bandwidth,
  353. mvk2mvs_latency,
  354. mvk2mvs_bandwidth,
  355. mvs2mvk_latency,
  356. mvs2mvk_bandwidth,
  357. mvk2mvi_latency,
  358. mvk2mvi_bandwidth,
  359. read_root,
  360. create_node,
  361. create_edge,
  362. create_nodevalue,
  363. create_dict,
  364. read_value,
  365. read_outgoing,
  366. read_incoming,
  367. read_edge,
  368. read_dict,
  369. read_dict_keys,
  370. read_dict_edge,
  371. read_dict_node,
  372. read_dict_node_edge,
  373. read_reverse_dict,
  374. delete_node,
  375. delete_edge):
  376. CoupledDEVS.__init__(self, "System")
  377. self.mvi_manager = self.addSubModel(ModelverseInterface(\
  378. username = "user_manager",
  379. operations = [username],
  380. ))
  381. self.mvi = self.addSubModel(ModelverseInterface(\
  382. username = username,
  383. operations = operations
  384. ))
  385. self.mvk = self.addSubModel(ModelverseKernel(\
  386. rules_per_phase = rules_per_phase
  387. ))
  388. self.mvs = self.addSubModel(ModelverseState(\
  389. read_root = read_root,
  390. create_node = create_node,
  391. create_edge = create_edge,
  392. create_nodevalue = create_nodevalue,
  393. create_dict = create_dict,
  394. read_value = read_value,
  395. read_outgoing = read_outgoing,
  396. read_incoming = read_incoming,
  397. read_edge = read_edge,
  398. read_dict = read_dict,
  399. read_dict_keys = read_dict_keys,
  400. read_dict_edge = read_dict_edge,
  401. read_dict_node = read_dict_node,
  402. read_dict_node_edge = read_dict_node_edge,
  403. read_reverse_dict = read_reverse_dict,
  404. delete_node = delete_node,
  405. delete_edge = delete_edge
  406. ))
  407. self.mvi2mvk = self.addSubModel(Network(\
  408. name = "mvi2mvk",
  409. latency = mvi2mvk_latency,
  410. bandwidth = mvi2mvk_bandwidth
  411. ))
  412. self.mvk2mvs = self.addSubModel(Network(\
  413. name = "mvk2mvs",
  414. latency = mvk2mvs_latency,
  415. bandwidth = mvk2mvs_bandwidth
  416. ))
  417. self.mvs2mvk = self.addSubModel(Network(\
  418. name = "mvs2mvk",
  419. latency = mvs2mvk_latency,
  420. bandwidth = mvs2mvk_bandwidth
  421. ))
  422. self.mvk2mvi = self.addSubModel(Network(\
  423. name = "mvk2mvi",
  424. latency = mvk2mvi_latency,
  425. bandwidth = mvk2mvi_bandwidth
  426. ))
  427. self.connectPorts(self.mvi_manager.to_mvk, self.mvk.from_mvi)
  428. self.connectPorts(self.mvi.to_mvk, self.mvi2mvk.input_port)
  429. self.connectPorts(self.mvi2mvk.output_port, self.mvk.from_mvi)
  430. self.connectPorts(self.mvk.to_mvs, self.mvk2mvs.input_port)
  431. self.connectPorts(self.mvk2mvs.output_port, self.mvs.from_mvk)
  432. self.connectPorts(self.mvs.to_mvk, self.mvs2mvk.input_port)
  433. self.connectPorts(self.mvs2mvk.output_port, self.mvk.from_mvs)
  434. self.connectPorts(self.mvk.to_mvi, self.mvk2mvi.input_port)
  435. self.connectPorts(self.mvk2mvi.output_port, self.mvi.from_mvk)
  436. code = \
  437. """
  438. include "primitives.alh"
  439. Element function main():
  440. \tlog("Executed the code!")
  441. \treturn
  442. """
  443. operations = [get_object_constructor(f.read()) for f in files] + \
  444. get_object_constructor(code) + \
  445. get_link_and_load("main", files + ["code.alc"])
  446. args = {
  447. "username": "test_user",
  448. "operations": operations,
  449. "mvi2mvk_latency": 1,
  450. "mvi2mvk_bandwidth": 2000,
  451. "mvk2mvs_latency": 1,
  452. "mvk2mvs_bandwidth": 2000,
  453. "mvs2mvk_latency": 1,
  454. "mvs2mvk_bandwidth": 2000,
  455. "mvk2mvi_latency": 1,
  456. "mvk2mvi_bandwidth": 2000,
  457. "read_root": lambda: 1,
  458. "create_node": lambda: 1,
  459. "create_edge": lambda: 1,
  460. "create_nodevalue": lambda: 1,
  461. "create_dict": lambda: 1,
  462. "read_value": lambda: 1,
  463. "read_outgoing": lambda: 1,
  464. "read_incoming": lambda: 1,
  465. "read_edge": lambda: 1,
  466. "read_dict": lambda: 1,
  467. "read_dict_keys": lambda: 1,
  468. "read_dict_node": lambda: 1,
  469. "read_dict_edge": lambda: 1,
  470. "read_dict_node_edge": lambda: 1,
  471. "read_reverse_dict": lambda: 1,
  472. "delete_node": lambda: 1,
  473. "delete_edge": lambda: 1,
  474. "rules_per_phase": 200,
  475. }
  476. model = System(**args)
  477. sim = Simulator(model)
  478. sim.setTerminationTime(900000)
  479. #sim.setVerbose()
  480. sim.simulate()