The source code and dockerfile for the GSW2024 AI Lab.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
This repo is archived. You can view files and clone it, but cannot push or open issues/pull-requests.

253 lines
9.1 KiB

4 weeks ago
  1. import os
  2. import stormpy
  3. from configurations import gspn, xml
  4. @gspn
  5. class TestGSPNBuilder:
  6. def test_layout_info(self):
  7. layout = stormpy.gspn.LayoutInfo()
  8. assert layout.x == 0
  9. assert layout.y == 0
  10. assert layout.rotation == 0
  11. layout.x = 1
  12. assert layout.x == 1
  13. layout_xy = stormpy.gspn.LayoutInfo(2, 3)
  14. assert layout_xy.x == 2
  15. assert layout_xy.rotation == 0
  16. layout_xyr = stormpy.gspn.LayoutInfo(2, 3, 4)
  17. assert layout_xyr.rotation == 4
  18. def test_place(self):
  19. p_id = 4
  20. place = stormpy.gspn.Place(id=p_id)
  21. assert p_id == place.get_id()
  22. assert not place.has_restricted_capacity()
  23. place.set_capacity(cap=5)
  24. assert place.has_restricted_capacity()
  25. assert place.get_capacity() == 5
  26. p_name = "P_0"
  27. place.set_name(name=p_name)
  28. assert place.get_name() == p_name
  29. p_tokens = 2
  30. place.set_number_of_initial_tokens(p_tokens)
  31. assert place.get_number_of_initial_tokens() == p_tokens
  32. def test_transition(self):
  33. # test TimedTransition
  34. tt = stormpy.gspn.TimedTransition()
  35. tt_name = " tt"
  36. tt.set_name(tt_name)
  37. assert tt_name == tt.get_name()
  38. tt_rate = 0.2
  39. tt.set_rate(tt_rate)
  40. assert tt_rate == tt.get_rate()
  41. # connect a place to this transition and test arcs
  42. place = stormpy.gspn.Place(0)
  43. # test input arcs
  44. assert not tt.exists_input_arc(place)
  45. tt.set_input_arc_multiplicity(place, 2)
  46. assert tt.exists_input_arc(place)
  47. assert tt.get_input_arc_multiplicity(place) == 2
  48. tt.remove_input_arc(place)
  49. assert not tt.exists_input_arc(place)
  50. # test output arcs
  51. assert not tt.exists_output_arc(place)
  52. tt.set_output_arc_multiplicity(place, 3)
  53. assert tt.exists_output_arc(place)
  54. assert tt.get_output_arc_multiplicity(place) == 3
  55. tt.remove_output_arc(place)
  56. assert not tt.exists_output_arc(place)
  57. # test inhibition arcs
  58. assert not tt.exists_inhibition_arc(place)
  59. tt.set_inhibition_arc_multiplicity(place, 5)
  60. assert tt.exists_inhibition_arc(place)
  61. assert tt.get_inhibition_arc_multiplicity(place) == 5
  62. tt.remove_inhibition_arc(place)
  63. assert not tt.exists_inhibition_arc(place)
  64. # test ImmediateTransition
  65. ti = stormpy.gspn.ImmediateTransition()
  66. ti_name = " ti"
  67. ti.set_name(ti_name)
  68. assert ti_name == ti.get_name()
  69. assert ti.no_weight_attached()
  70. ti_weight = 0.2
  71. ti.set_weight(ti_weight)
  72. assert ti_weight == ti.get_weight()
  73. assert not ti.no_weight_attached()
  74. def test_build_gspn(self):
  75. gspn_name = "gspn_test"
  76. builder = stormpy.gspn.GSPNBuilder()
  77. id_p_0 = builder.add_place()
  78. id_p_1 = builder.add_place(initial_tokens=1)
  79. id_p_2 = builder.add_place(initial_tokens=0, name="place_2")
  80. id_p_3 = builder.add_place(capacity=2, initial_tokens=3, name="place_3")
  81. p_layout = stormpy.gspn.LayoutInfo(1, 2)
  82. builder.set_place_layout_info(id_p_0, p_layout)
  83. id_ti_0 = builder.add_immediate_transition(priority=1, weight=0.5, name="ti_0")
  84. id_ti_1 = builder.add_immediate_transition()
  85. id_tt_0 = builder.add_timed_transition(priority=2, rate=0.4, name="tt_0")
  86. id_tt_1 = builder.add_timed_transition(priority=0, rate=0.5, num_servers=2, name="tt_1")
  87. t_layout = stormpy.gspn.LayoutInfo(1, 2)
  88. builder.set_transition_layout_info(id_ti_0, t_layout)
  89. # add arcs
  90. builder.add_input_arc(id_p_0, id_ti_1, multiplicity=2)
  91. builder.add_input_arc("place_2", "ti_0", multiplicity=2)
  92. builder.add_output_arc(id_ti_1, id_p_2, multiplicity=2)
  93. builder.add_output_arc("tt_0", "place_3", multiplicity=2)
  94. builder.add_inhibition_arc(id_p_2, id_tt_0, multiplicity=2)
  95. builder.add_inhibition_arc("place_3", "tt_0", multiplicity=2)
  96. builder.add_normal_arc("place_3", "tt_0", multiplicity=2)
  97. builder.add_normal_arc("tt_0", "place_3", multiplicity=2)
  98. # test gspn composition
  99. builder.set_name(gspn_name)
  100. gspn = builder.build_gspn()
  101. assert gspn.get_name() == gspn_name
  102. assert gspn.is_valid()
  103. assert gspn.get_number_of_immediate_transitions() == 2
  104. assert gspn.get_number_of_timed_transitions() == 2
  105. assert gspn.get_number_of_places() == 4
  106. assert len(gspn.get_places()) == 4
  107. assert len(gspn.get_immediate_transitions()) == 2
  108. assert len(gspn.get_timed_transitions()) == 2
  109. # test places
  110. p_0 = gspn.get_place(id_p_0)
  111. assert p_0.get_id() == id_p_0
  112. p_1 = gspn.get_place(id_p_1)
  113. assert p_1.get_id() == id_p_1
  114. assert p_1.get_number_of_initial_tokens() == 1
  115. p_2 = gspn.get_place(id_p_2)
  116. assert p_2.get_id() == id_p_2
  117. assert p_2.get_name() == "place_2"
  118. p_3 = gspn.get_place(id_p_3)
  119. assert p_3.get_name() == "place_3"
  120. assert p_3.get_capacity() == 2
  121. assert p_3.get_number_of_initial_tokens() == 3
  122. assert p_3.has_restricted_capacity()
  123. # test transitions
  124. ti_0 = gspn.get_immediate_transition("ti_0")
  125. assert ti_0.get_id() == id_ti_0
  126. assert ti_0.get_weight() == 0.5
  127. assert ti_0.get_priority() == 1
  128. tt_0 = gspn.get_timed_transition("tt_0")
  129. assert tt_0.get_id() == id_tt_0
  130. assert tt_0.get_rate() == 0.4
  131. assert tt_0.get_priority() == 2
  132. tt_1 = gspn.get_timed_transition("tt_1")
  133. assert tt_1.get_id() == id_tt_1
  134. assert tt_1.get_number_of_servers() == 2
  135. # test new name
  136. gspn_new_name = "new_name"
  137. gspn.set_name(gspn_new_name)
  138. assert gspn.get_name() == gspn_new_name
  139. def test_export_to_pnpro(self, tmpdir):
  140. builder = stormpy.gspn.GSPNBuilder()
  141. builder.set_name("gspn_test")
  142. # add places and transitions
  143. id_p_0 = builder.add_place()
  144. id_p_1 = builder.add_place(initial_tokens=3, name="place_1", capacity=2)
  145. id_ti_0 = builder.add_immediate_transition(priority=0, weight=0.5, name="ti_0")
  146. id_tt_0 = builder.add_timed_transition(priority=0, rate=0.5, num_servers=2, name="tt_0")
  147. gspn = builder.build_gspn()
  148. export_file = os.path.join(str(tmpdir), "gspn.pnpro")
  149. # export gspn to pnml
  150. gspn.export_gspn_pnpro_file(export_file)
  151. # import gspn
  152. gspn_parser = stormpy.gspn.GSPNParser()
  153. gspn_import = gspn_parser.parse(export_file)
  154. # test imported gspn
  155. assert gspn_import.get_name() == gspn.get_name()
  156. assert gspn_import.get_number_of_timed_transitions() == gspn.get_number_of_timed_transitions()
  157. assert gspn_import.get_number_of_immediate_transitions() == gspn.get_number_of_immediate_transitions()
  158. assert gspn_import.get_number_of_places() == gspn.get_number_of_places()
  159. p_0 = gspn_import.get_place(id_p_0)
  160. assert p_0.get_id() == id_p_0
  161. p_1 = gspn_import.get_place(id_p_1)
  162. assert p_1.get_name() == "place_1"
  163. # todo capacity info lost
  164. # assert p_1.get_capacity() == 2
  165. # assert p_1.has_restricted_capacity() == True
  166. assert p_1.get_number_of_initial_tokens() == 3
  167. ti_0 = gspn_import.get_immediate_transition("ti_0")
  168. assert ti_0.get_id() == id_ti_0
  169. tt_0 = gspn_import.get_timed_transition("tt_0")
  170. assert tt_0.get_id() == id_tt_0
  171. @xml
  172. def test_export_to_pnml(self, tmpdir):
  173. builder = stormpy.gspn.GSPNBuilder()
  174. builder.set_name("gspn_test")
  175. # add places and transitions
  176. id_p_0 = builder.add_place()
  177. id_p_1 = builder.add_place(initial_tokens=3, name="place_1", capacity=2)
  178. id_ti_0 = builder.add_immediate_transition(priority=0, weight=0.5, name="ti_0")
  179. id_tt_0 = builder.add_timed_transition(priority=0, rate=0.5, num_servers=2, name="tt_0")
  180. gspn = builder.build_gspn()
  181. export_file = os.path.join(str(tmpdir), "gspn.pnml")
  182. # export gspn to pnml
  183. gspn.export_gspn_pnml_file(export_file)
  184. # import gspn
  185. gspn_parser = stormpy.gspn.GSPNParser()
  186. gspn_import = gspn_parser.parse(export_file)
  187. # test imported gspn
  188. assert gspn_import.get_name() == gspn.get_name()
  189. assert gspn_import.get_number_of_timed_transitions() == gspn.get_number_of_timed_transitions()
  190. assert gspn_import.get_number_of_immediate_transitions() == gspn.get_number_of_immediate_transitions()
  191. assert gspn_import.get_number_of_places() == gspn.get_number_of_places()
  192. p_0 = gspn_import.get_place(id_p_0)
  193. assert p_0.get_id() == id_p_0
  194. p_1 = gspn_import.get_place(id_p_1)
  195. assert p_1.get_name() == "place_1"
  196. assert p_1.get_capacity() == 2
  197. assert p_1.get_number_of_initial_tokens() == 3
  198. assert p_1.has_restricted_capacity()
  199. ti_0 = gspn_import.get_immediate_transition("ti_0")
  200. assert ti_0.get_id() == id_ti_0
  201. tt_0 = gspn_import.get_timed_transition("tt_0")
  202. assert tt_0.get_id() == id_tt_0