KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use save_as method in green
Best Python code snippet using green
UniformRefinerUnitTests.py
Source:UniformRefinerUnitTests.py 
...10 p_size = eMesh.get_parallel_size()11 gmesh_spec = "4x4x"+str(4*p_size)+"|bbox:0,0,0,1,1,1"12 eMesh.new_mesh(GMeshSpec(gmesh_spec))13 eMesh.commit()14 eMesh.save_as("hex_fixture.e")15 eMesh = PerceptMesh()16 eMesh.open("exodus_files/"+input_files_loc+"hex_fixture.e")17 scalarDimension = 018 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)19 breaker = Refiner(eMesh, HEX8_TET4_24, proc_rank_field)20 eMesh.commit()21 breaker.doBreak()22 eMesh.save_as("tet_fixture.e")23 24def fixture_setup_1():25 pm = MPI.COMM_WORLD26 p_size = parallel_machine_size(pm)27 if p_size <= 3:28 n = 1229 nx = n30 ny = n31 fixture = QuadFixture_4(pm, nx, ny, True)32 fixture.meta_data.commit()33 fixture.generate_mesh()34 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data)35 eMesh.print_info("quad fixture", 2)36 eMesh.save_as("quad_fixture.e")37 fixture = QuadFixture_4(pm, nx, ny, False)38 fixture.meta_data.commit()39 fixture.generate_mesh()40 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data)41 eMesh.print_info("quad fixture no sidesets", 2)42 eMesh.save_as("quad_fixture_no_sidesets.e")43is_setup = False44def fixture_setup():45 if is_setup == True:46 pass47 else:48 fixture_setup_0()49 fixture_setup_1()50 is_setup == True51 52class UniformRefinerUnitTests(unittest.TestCase):53 def test_break_quad_to_quad_sierra_1(self):54 pm = MPI.COMM_WORLD55 p_size = parallel_machine_size(pm)56 if p_size <= 2:57 n = 258 nx = 159 ny = n60 createEdgeSets = False61 fixture = QuadFixture_4(pm,nx,ny,createEdgeSets)62 isCommited = False63 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)64 scalarDimension = 065 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)66 eMesh.add_field("proc_rank_edge", eMesh.edge_rank(), scalarDimension)67 breaker = Refiner(eMesh, QUAD4_QUAD4_4, proc_rank_field)68 eMesh.commit()69 fixture.generate_mesh()70 breaker.setRemoveOldElements(False)71 breaker.doBreak()72 eMesh.dump_elements_compact()73 def test_break_tri_to_tri_sierra_1_test(self):74 pm = MPI.COMM_WORLD75 p_size = parallel_machine_size(pm)76 if p_size <= 2:77 n = 278 nx = n79 ny = n80 createEdgeSets = False81 fixture = QuadFixture_4(pm,nx,ny,createEdgeSets)82 isCommited = False83 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)84 scalarDimension = 085 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)86 eMesh.add_field("proc_rank_edge", eMesh.edge_rank(), scalarDimension)87 breaker = Refiner(eMesh, TRI3_TRI3_4, proc_rank_field)88 eMesh.commit()89 fixture.generate_mesh()90 breaker.setRemoveOldElements(False)91 breaker.doBreak()92 eMesh.dump_elements_compact()93 def test_break_quad_to_quad_sierra(self):94 fixture_setup()95 pm = MPI.COMM_WORLD96 p_size = parallel_machine_size(pm)97 if p_size <= 3:98 n = 1299 nx = n100 ny = n101 fixture = QuadFixture_4(pm, nx, ny, True)102 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, False)103 scalarDimension = 0104 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)105 breaker = Refiner(eMesh, QUAD4_QUAD4_4_SIERRA, proc_rank_field)106 eMesh.commit()107 fixture.generate_mesh()108 eMesh.print_info("quad mesh")109 breaker.doBreak()110 def test_hex8_hex8_8_1(self):111 fixture_setup()112 eMesh = PerceptMesh()113 p_size = eMesh.get_parallel_size()114 gmesh_spec = "4x4x"+str(4*p_size)+"|bbox:0,0,0,1,1,1"115 eMesh.new_mesh(GMeshSpec(gmesh_spec))116 scalarDimension = 0117 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)118 breaker = Refiner(eMesh, HEX8_HEX8_8, proc_rank_field)119 eMesh.commit()120 breaker.doBreak121 def test_wedge6_1(self):122 fixture_setup()123 eMesh = PerceptMesh()124 p_size = eMesh.get_parallel_size()125 if p_size == 1:126 wedgeFixture = WedgeFixture()127 wedgeFixture.createMesh(MPI.COMM_WORLD, 4,3,2,0,1,0,1,0,1, "swept_wedge_0.e")128 def test_beam_enrich(self):129 fixture_setup()130 pm = MPI.COMM_WORLD131 p_size = parallel_machine_size(pm)132 if p_size <= 1:133 134 eMesh = PerceptMesh()135 eMesh.open("exodus_files/beam.e")136 scalarDimension = 0137 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)138 eMesh.save_as("beam_enrich_0.e")139 breaker = Refiner(eMesh, BEAM2_BEAM3_1, proc_rank_field)140 eMesh.commit()141 breaker.setIgnoreSideSets(True)142 breaker.doBreak()143 eMesh.save_as("beam_enrich.e")144 def test_beam_refine(self):145 fixture_setup()146 pm = MPI.COMM_WORLD147 p_size = parallel_machine_size(pm)148 if p_size <= 1:149 mesh = BeamFixture(pm, False)150 mesh.m_metaData.commit()151 mesh.populate()152 isCommited = True153 em1 = PerceptMesh(mesh.m_metaData, mesh.m_bulkData, isCommited)154 em1.save_as("beam_0.e")155 eMesh = PerceptMesh()156 eMesh.open("beam_0.e")157 scalarDimension = 0158 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)159 breaker = Refiner(eMesh, BEAM2_BEAM2_2, proc_rank_field)160 eMesh.commit()161 breaker.setIgnoreSideSets(True)162 breaker.doBreak()163 def test_break_quad_to_tri_6(self):164 fixture_setup()165 pm = MPI.COMM_WORLD166 p_size = parallel_machine_size(pm)167 if p_size == 1 or p_size == 3:168 eMesh = PerceptMesh(2)169 eMesh.open("quad_fixture.e")170 scalarDimension = 0171 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)172 breaker = Refiner(eMesh, QUAD4_TRI3_6, proc_rank_field)173 eMesh.commit()174 eMesh.print_info("quad mesh")175 breaker.setIgnoreSideSets(True)176 breaker.setRemoveOldElements(False) 177 breaker.doBreak()178 eMesh.save_as("square_quad4_out.e")179 def test_break_quad_to_tri_4(self):180 fixture_setup()181 pm = MPI.COMM_WORLD182 p_size = parallel_machine_size(pm)183 if p_size == 1 or p_size == 3:184 eMesh = PerceptMesh(2)185 eMesh.open("quad_fixture.e")186 scalarDimension = 0187 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)188 breaker = Refiner(eMesh, QUAD4_TRI3_4, proc_rank_field)189 eMesh.commit()190 eMesh.print_info("quad mesh")191 breaker.setIgnoreSideSets(True)192 breaker.setRemoveOldElements(False)193 breaker.doBreak()194 eMesh.save_as("square_quad4_tri3_4_out.e")195 def test_break_quad_to_quad(self):196 fixture_setup()197 pm = MPI.COMM_WORLD198 p_size = parallel_machine_size(pm)199 if p_size == 1 or p_size == 3:200 eMesh = PerceptMesh(2)201 eMesh.open("quad_fixture_no_sidesets.e")202 scalarDimension = 0203 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)204 breaker = Refiner(eMesh, QUAD4_QUAD4_4, proc_rank_field)205 eMesh.commit()206 eMesh.print_info("quad mesh")207 breaker.setIgnoreSideSets(True)208 breaker.doBreak()209 eMesh.save_as("square_quad4_ref_out.e")210 def test_break_quad_to_quad_sierra_unit1(self):211 fixture_setup()212 pm = MPI.COMM_WORLD213 p_size = parallel_machine_size(pm)214 if p_size == 1 or p_size == 3:215 eMesh = PerceptMesh(2)216 eMesh.open("quad_fixture.e")217 scalarDimension = 0218 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)219 breaker = Refiner(eMesh, QUAD4_QUAD4_4_SIERRA, proc_rank_field)220 eMesh.commit()221 eMesh.print_info("quad mesh")222 breaker.doBreak()223 eMesh.save_as("square_quad4_sierra_ref_out.e")224 def test_break_quad_to_quad_sierra_sidesets(self):225 fixture_setup()226 pm = MPI.COMM_WORLD227 p_size = parallel_machine_size(pm)228 if p_size == 1 or p_size == 2:229 eMesh = PerceptMesh(2)230 eMesh.open("quad_fixture.e")231 scalarDimension = 0232 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)233 breaker = Refiner(eMesh, QUAD4_QUAD4_4_SIERRA, proc_rank_field)234 eMesh.commit()235 eMesh.print_info("after refinement break_quad_to_quad_sierra_sidesets")236 breaker.doBreak()237 eMesh.save_as("quad_sidesets_sierra_out.e")238 def test_break_hex8_tet4_24_1(self):239 fixture_setup()240 pm = MPI.COMM_WORLD241 p_size = parallel_machine_size(pm)242 eMesh = PerceptMesh(3)243 gmesh_spec = "1x1x" + str(p_size) + "|bbox:0,0,0,1,1," + str(p_size)244 eMesh.new_mesh(GMeshSpec(gmesh_spec))245 scalarDimension = 0246 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)247 breaker = Refiner(eMesh, HEX8_TET4_24, proc_rank_field)248 eMesh.commit()249 eMesh.print_info()250 breaker.setRemoveOldElements(True)251 breaker.doBreak()252 eMesh.save_as("hex_tet_24_cube1x1x1.e")253 def test_hex8_tet4_6_12_1(self):254 fixture_setup()255 eMesh = PerceptMesh(3)256 p_size = eMesh.get_parallel_size() 257 gmesh_spec = "1x1x" + str(p_size) + "|bbox:0,0,0,1,1," + str(p_size)258 eMesh.new_mesh(GMeshSpec(gmesh_spec))259 scalarDimension = 0260 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)261 breaker = Refiner(eMesh, HEX8_TET4_6_12, proc_rank_field)262 eMesh.commit()263 eMesh.print_info()264 breaker.setRemoveOldElements(True)265 breaker.doBreak()266 eMesh.save_as("hex_tet_6_12_cube1x1x1.e")267 def test_hex8_tet4_6_12_2(self):268 fixture_setup()269 pm = MPI.COMM_WORLD270 p_size = parallel_machine_size(pm)271 eMesh = PerceptMesh(3)272 eMesh.open("hex_fixture.e")273 scalarDimension = 0274 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)275 breaker = Refiner(eMesh, HEX8_TET4_6_12, proc_rank_field)276 eMesh.commit()277 eMesh.print_info()278 breaker.doBreak()279 eMesh.save_as("hex_tet_6_12_1.e")280 def test_quad4_quad4_test_1(self):281 fixture_setup()282 pm = MPI.COMM_WORLD283 p_size = parallel_machine_size(pm)284 if p_size <= 3:285 n = 12286 nx = n287 ny = n288 fixture = QuadFixture_4(pm,nx,ny,True)289 fixture.meta_data.commit()290 fixture.generate_mesh()291 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data)292 eMesh.print_info("quad fixture")293 eMesh.save_as("quad_fixture_test_1.e")294 def test_break_quad_to_quad_sierra_1_unit1(self):295 fixture_setup()296 pm = MPI.COMM_WORLD297 p_size = parallel_machine_size(pm)298 doGenSideSets = True299 if p_size <= 3:300 n = 12301 nx = n302 ny = n303 fixture = QuadFixture_4(pm,nx,ny,doGenSideSets)304 isCommited = False305 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)306 eMesh.commit()307 fixture.generate_mesh()308 eMesh.save_as("quad_fixture_0.e")309 eMesh.close()310 311 i = 0312 while i < 2:313 print "\n\n\n ================ tmp Refine Pass = ", i314 eMesh1 = PerceptMesh(2)315 eMesh1.open("quad_fixture_" + str(i) + ".e")316 scalarDimension = 0317 proc_rank_field = eMesh1.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)318 breaker = Refiner(eMesh1, QUAD4_QUAD4_4_SIERRA, proc_rank_field)319 eMesh1.commit()320 breaker.doBreak()321 eMesh1.save_as("quad_fixture_" + str(i+1) + ".e")322 eMesh1.close()323 i = i + 1324 def test_break_quad_to_quad_sierra_2(self):325 fixture_setup()326 pm = MPI.COMM_WORLD327 p_size = parallel_machine_size(pm)328 doGenSideSets = True329 if p_size <= 3:330 n = 12331 nx = n332 ny = n333 fixture = QuadFixture_4(pm,nx,ny,doGenSideSets)334 isCommited = False335 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)336 eMesh.commit()337 fixture.generate_mesh()338 eMesh.save_as("quad_fixture_mbreak_0.e")339 eMesh.close()340 eMesh1 = PerceptMesh(2)341 eMesh1.open("quad_fixture_mbreak_0.e")342 scalarDimension = 0343 proc_rank_field = eMesh1.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)344 breaker = Refiner(eMesh1, QUAD4_QUAD4_4_SIERRA, proc_rank_field)345 eMesh1.commit()346 347 i = 0348 while i < 2:349 print "\n\n\n ================ tmp Refine Pass = ", i350 breaker.doBreak()351 eMesh1.save_as("quad_fixture_mbreak_" + str(i) + ".e")352 i = i + 1353 def test_break_quad4_to_quad9(self):354 fixture_setup()355 pm = MPI.COMM_WORLD356 p_size = parallel_machine_size(pm)357 if p_size <= 3:358 n = 12359 nx = n360 ny = n361 doGenSideSets = True362 fixture = QuadFixture_4(pm,nx,ny,doGenSideSets)363 isCommited = False364 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)365 scalarDimension = 0366 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)367 breaker = Refiner(eMesh, QUAD4_QUAD9_1, proc_rank_field)368 eMesh.commit()369 fixture.generate_mesh()370 eMesh.save_as("quad_fixture_quad9_0.e")371 breaker.doBreak()372 eMesh.save_as("quad_fixture_quad9_1.e")373 def test_break_quad4_to_quad9(self):374 fixture_setup()375 pm = MPI.COMM_WORLD376 p_size = parallel_machine_size(pm)377 if p_size <= 3:378 n = 12379 nx = n380 ny = n381 doGenSideSets = True382 fixture = QuadFixture_4(pm,nx,ny,doGenSideSets)383 isCommited = False384 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)385 scalarDimension = 0386 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)387 breaker = Refiner(eMesh, QUAD4_QUAD8_1, proc_rank_field)388 eMesh.commit()389 fixture.generate_mesh()390 eMesh.save_as("quad_fixture_quad8_0.e")391 breaker.doBreak()392 eMesh.save_as("quad_fixture_quad8_1.e")393 eMesh.save_as("quad_fixture_quad8_quad8_0.e")394 def test_quad8_to_quad8(self):395 fixture_setup()396 pm = MPI.COMM_WORLD397 p_size = parallel_machine_size(pm)398 if p_size <= 3:399 eMesh = PerceptMesh(2)400 eMesh.open("quad_fixture_quad8_quad8_0.e")401 scalarDimension = 0402 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)403 breaker = Refiner(eMesh, QUAD8_QUAD8_4, proc_rank_field)404 eMesh.commit()405 breaker.setIgnoreSideSets(False)406 breaker.doBreak()407 eMesh.save_as("quad_fixture_quad8_quad8_1.e")408 def test_break_quad4_to_quad9_to_quad9_0(self):409 fixture_setup()410 pm = MPI.COMM_WORLD411 p_size = parallel_machine_size(pm)412 doGenSideSets = False413 if p_size <= 1:414 n = 12415 nx = n416 ny = n417 fixture = QuadFixture_4(pm,nx,ny,doGenSideSets)418 isCommited = False419 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)420 scalarDimension = 0421 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)422 breaker = Refiner(eMesh, QUAD4_QUAD9_1, proc_rank_field)423 eMesh.commit()424 fixture.generate_mesh()425 breaker.doBreak()426 eMesh.save_as("quad_1x1x_quad9_quad9_0.e")427 428 em1 = PerceptMesh(2)429 em1.open("quad_1x1x_quad9_quad9_0.e")430 proc_rank_field = em1.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)431 breaker = Refiner(em1, QUAD9_QUAD9_4, proc_rank_field)432 em1.commit()433 breaker.setIgnoreSideSets(True)434 breaker.doBreak()435 em1.save_as("quad_1x1x_quad9_quad9_1.e")436 def test_break_quad4_to_quad9_to_quad9(self):437 fixture_setup()438 pm = MPI.COMM_WORLD439 p_size = parallel_machine_size(pm)440 doGenSideSets = True441 if p_size <= 3:442 n = 12443 nx = n444 ny = n445 fixture = QuadFixture_4(pm, nx, ny, doGenSideSets)446 isCommited = False447 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)448 scalarDimension = 0449 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)450 breaker = Refiner(eMesh, QUAD4_QUAD9_1, proc_rank_field)451 eMesh.commit()452 fixture.generate_mesh()453 breaker.doBreak()454 eMesh.save_as("quad_fixture_quad9_quad9_0.e")455 em1 = PerceptMesh(2)456 em1.open("quad_fixture_quad9_quad9_0.e")457 scalarDimension = 0458 proc_rank_field = em1.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)459 breaker = Refiner(em1, QUAD9_QUAD9_4, proc_rank_field)460 em1.commit()461 breaker.doBreak()462 em1.save_as("quad_fixture_quad9_quad9_1.e")463 def test_break_tri_to_tri_sierra_0(self):464 fixture_setup()465 pm = MPI.COMM_WORLD466 p_size = parallel_machine_size(pm)467 if p_size <= 3:468 n = 12469 nx = n470 ny = n471 fixture = QuadFixture_3(pm, nx, ny, True)472 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, False)473 eMesh.commit()474 fixture.generate_mesh()475 eMesh.print_info("tri mesh")476 eMesh.save_as("quad_fixture_tri3.e") 477 def test_break_tri_to_tri_sierra_1(self):478 fixture_setup()479 pm = MPI.COMM_WORLD480 p_size = parallel_machine_size(pm)481 if p_size <= 3:482 n = 12483 nx = n484 ny = n485 createEdgeSets = True486 fixture = QuadFixture_3(pm, nx, ny, createEdgeSets)487 isCommited = False488 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)489 scalarDimension = 0490 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)491 eMesh.add_field("proc_rank_ede", eMesh.edge_rank(), scalarDimension)492 breaker = Refiner(eMesh, TRI3_TRI3_4, proc_rank_field)493 eMesh.commit()494 fixture.generate_mesh()495 eMesh.print_info("tri mesh")496 eMesh.save_as("quad_fixture_tri3_0.e")497 breaker.doBreak()498 eMesh.print_info("tri mesh refined")499 eMesh.save_as("quad_fixture_tri3_1.e")500 def test_break_tri3_to_tri6_sierra(self):501 fixture_setup()502 pm = MPI.COMM_WORLD503 p_size = parallel_machine_size(pm)504 print p_size, "++++++++++++++++++++"505 for i in range(100):506 print "-------------------"507 if p_size <= 3:508 n = 12509 nx = n510 ny = n511 createEdgeSets = True512 fixture = QuadFixture_3(pm, nx, ny, createEdgeSets)513 isCommited = False514 eMesh = PerceptMesh(fixture.meta_data, fixture.bulk_data, isCommited)515 scalarDimension = 0516 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)517 eMesh.add_field("proc_rank_ede", eMesh.edge_rank(), scalarDimension)518 breaker = Refiner(eMesh, TRI3_TRI6_1, proc_rank_field)519 eMesh.commit()520 fixture.generate_mesh()521 eMesh.print_info("tri mesh tri6")522 eMesh.save_as("quad_fixture_tri3_tri6_0.e")523 breaker.doBreak()524 eMesh.print_info("tri mesh enriched")525 eMesh.save_as("quad_fixture_tri6_tri6_0.e")526 eMesh.save_as("quad_fixture_tri3_tri6_1.e")527 528 def test_break_tri3_to_tri6_to_tri6_sierra(self):529 fixture_setup()530 pm = MPI.COMM_WORLD531 p_size = parallel_machine_size(pm)532 if p_size <= 3:533 eMesh = PerceptMesh(2)534 eMesh.open("quad_fixture_tri6_tri6_0.e")535 scalarDimension = 0536 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)537 eMesh.add_field("proc_rank_ede", eMesh.edge_rank(), scalarDimension)538 breaker = Refiner(eMesh, TRI6_TRI6_4, proc_rank_field)539 eMesh.commit() 540 eMesh.print_info("tri mesh tri6")541 eMesh.save_as("quad_fixture_tri6_tri6_0.e")542 breaker.doBreak()543 eMesh.print_info("tri mesh refined")544 eMesh.save_as("quad_fixture_tri6_tri6_1.e")545 def test_break_tet4_tet4_1(self):546 fixture_setup()547 pm = MPI.COMM_WORLD548 p_size = parallel_machine_size(pm)549 if p_size == 1 or p_size == 3:550 eMesh = PerceptMesh(3)551 eMesh.open_read_only("tet_fixture.e")552 eMesh.save_as("tet_from_hex_fixture_0.e")553 if p_size == 1 or p_size == 3:554 eMesh = PerceptMesh(3)555 eMesh.open("tet_from_hex_fixture_0.e")556 scalarDimension = 0557 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)558 breaker = Refiner(eMesh, TET4_TET4_8, proc_rank_field)559 eMesh.commit()560 eMesh.print_info("tet mesh")561 breaker.doBreak()562 eMesh.save_as("tet4_refined_1.e")563 breaker.doBreak()564 eMesh.save_as("tet4_refined_2.e")565 566 def test_break_tet4_tet10_1(self):567 fixture_setup()568 pm = MPI.COMM_WORLD569 p_size = parallel_machine_size(pm)570 if p_size == 1 or p_size == 3:571 eMesh = PerceptMesh(3)572 eMesh.open_read_only("tet_fixture.e")573 eMesh.save_as("tet_from_hex_fixture_0.e")574 if p_size == 1 or p_size == 3:575 eMesh = PerceptMesh(3)576 eMesh.open("tet_from_hex_fixture_0.e")577 scalarDimension = 0578 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)579 breaker = Refiner(eMesh, TET4_TET10_1, proc_rank_field)580 eMesh.commit()581 eMesh.print_info("tet mesh")582 breaker.doBreak()583 eMesh.save_as("tet10_1.e")584 def test_break_tet4_tet10_tet10_1(self):585 fixture_setup()586 pm = MPI.COMM_WORLD587 p_size = parallel_machine_size(pm)588 if p_size == 1 or p_size == 3:589 eMesh = PerceptMesh(3)590 eMesh.open("tet_from_hex_fixture_0.e")591 scalarDimension = 0592 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)593 breaker = Refiner(eMesh, TET4_TET10_1, proc_rank_field)594 eMesh.commit()595 eMesh.print_info("tet mesh")596 breaker.doBreak()597 eMesh.save_as("tet10_1.e")598 eMesh.print_info("tet10_1")599 if p_size == 1 or p_size == 3:600 eMesh = PerceptMesh(3)601 eMesh.open("tet10_1.e")602 scalarDimension = 0603 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)604 breaker = Refiner(eMesh, TET10_TET10_8, proc_rank_field)605 eMesh.commit()606 breaker.doBreak()607 eMesh.save_as("tet10_tet10_1.e")608 609 def test_hex8_hex8_8_1_unit1(self):610 fixture_setup()611 eMesh = PerceptMesh(3)612 p_size = eMesh.get_parallel_size()613 gmesh_spec = "1x1x" + str(p_size) + "|bbox:0,0,0,1,1," + str(p_size)614 eMesh.new_mesh(GMeshSpec(gmesh_spec))615 scalarDimension = 0616 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)617 breaker = Refiner(eMesh, HEX8_HEX8_8, proc_rank_field) 618 eMesh.commit()619 eMesh.print_info()620 eMesh.save_as("hex_hex_cube1x1x" + str(p_size) + "-orig.e")621 breaker.setRemoveOldElements(True)622 breaker.doBreak()623 eMesh.save_as( "hex_hex_cube1x1x" + str(p_size)+".e")624 def test_hex8_hex8_8_2_unit1(self):625 fixture_setup()626 pm = MPI.COMM_WORLD627 p_size = parallel_machine_size(pm)628 if p_size == 1 or p_size == 3:629 eMesh = PerceptMesh(3)630 eMesh.open("hex_fixture.e")631 scalarDimension = 0632 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)633 breaker = Refiner(eMesh, HEX8_HEX8_8, proc_rank_field)634 eMesh.commit()635 eMesh.print_info()636 eMesh.save_as("hex8_0.e")637 breaker.doBreak()638 eMesh.save_as("hex8_1.e")639 breaker.doBreak()640 eMesh.save_as("hex8_2.e")641 642 def test_hex8_hex27_1_2(self):643 fixture_setup()644 pm = MPI.COMM_WORLD645 p_size = parallel_machine_size(pm)646 if p_size == 1 or p_size == 3:647 eMesh = PerceptMesh(3)648 eMesh.open("hex_fixture.e")649 scalarDimension = 0650 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)651 breaker = Refiner(eMesh, HEX8_HEX27_1, proc_rank_field)652 eMesh.commit()653 eMesh.print_info()654 eMesh.save_as("hex27_0.e")655 breaker.doBreak()656 eMesh.save_as("hex27_1.e")657 def test_hex08_hex20_1_1(self):658 fixture_setup()659 eMesh = PerceptMesh(3)660 p_size = eMesh.get_parallel_size()661 gmesh_spec = "1x1x" + str(p_size) + "|bbox:0,0,0,1,1," + str(p_size)662 eMesh.new_mesh(GMeshSpec(gmesh_spec))663 scalarDimension = 0664 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)665 breaker = Refiner(eMesh, HEX8_HEX20_1, proc_rank_field)666 eMesh.commit()667 eMesh.print_info()668 eMesh.save_as("hex8_hex_20_cube1x1x"+str(p_size) + "-orig.e")669 breaker.setRemoveOldElements(True)670 breaker.doBreak()671 eMesh.save_as("hex8_hex20_cube1x1x"+str(p_size) + ".e")672 eMesh.save_as("hex20_hex20_cube1x1x"+str(p_size) + "_0.e")673 674 def test_hex08_hex20_1_2(self):675 fixture_setup()676 pm = MPI.COMM_WORLD677 p_size = parallel_machine_size(pm)678 if p_size == 1 or p_size == 3:679 eMesh = PerceptMesh(3)680 eMesh.open("hex_fixture.e")681 scalarDimension = 0682 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)683 breaker = Refiner(eMesh, HEX8_HEX20_1, proc_rank_field)684 eMesh.commit()685 eMesh.print_info()686 eMesh.save_as("hex20_0.e")687 breaker.doBreak()688 eMesh.save_as("hex20_1.e")689 eMesh.save_as("hex20_hex20_0.e")690 def test_hex20_hex20_1(self):691 fixture_setup()692 eMesh = PerceptMesh(3)693 p_size = eMesh.get_parallel_size()694 if p_size <= 3:695 eMesh.open("hex20_hex20_cube1x1x"+str(p_size) + "_0.e")696 scalarDimension = 0697 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)698 breaker = Refiner(eMesh, HEX20_HEX20_8, proc_rank_field)699 eMesh.commit()700 breaker.setRemoveOldElements(True)701 breaker.doBreak()702 eMesh.save_as("hex20_hex20_cube1x1x" + str(p_size) + "_1.e")703 def test_hex20_hex20_1_2(self):704 fixture_setup()705 pm = MPI.COMM_WORLD706 p_size = parallel_machine_size(pm)707 if p_size == 1 or p_size == 3:708 eMesh = PerceptMesh(3)709 eMesh.open("hex20_hex20_0.e")710 scalarDimension = 0711 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)712 breaker = Refiner(eMesh, HEX20_HEX20_8, proc_rank_field)713 eMesh.commit()714 eMesh.save_as("hex20_hex20_0.e")715 breaker.doBreak()716 eMesh.save_as("hex20_hex20_1.e")717 def test_hex27_hex27_0(self):718 fixture_setup()719 scalarDimension = 0720 eMesh = PerceptMesh()721 p_size = eMesh.get_parallel_size()722 gmesh_spec = "1x1x" + str(p_size) + "|bbox:0,0,0,1,1," + str(p_size)723 eMesh.new_mesh(GMeshSpec(gmesh_spec))724 725 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)726 breaker = Refiner(eMesh, HEX8_HEX27_1, proc_rank_field)727 eMesh.commit()728 eMesh.print_info()729 eMesh.save_as("hex27_hex27_cube1x1x" + str(p_size) + "-orig.e")730 breaker.setRemoveOldElements(True)731 breaker.doBreak()732 eMesh.save_as("hex27_hex27_cube1x1x" + str(p_size) + "_0.e")733 734 em1 = PerceptMesh(3)735 p_size = em1.get_parallel_size()736 em1.open("hex27_hex27_cube1x1x" + str(p_size) + "_0.e")737 proc_rank_field = em1.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)738 breaker = Refiner(em1, HEX27_HEX27_8, proc_rank_field)739 em1.commit()740 breaker.setIgnoreSideSets(True)741 breaker.setRemoveOldElements(True)742 breaker.doBreak()743 em1.save_as("hex27_hex27_cube1x1x" + str(p_size) + "_1.e")744 def test_hex27_hex27_1(self):745 fixture_setup()746 pm = MPI.COMM_WORLD747 p_size = parallel_machine_size(pm)748 if p_size == 1 or p_size == 3:749 eMesh = PerceptMesh(3)750 eMesh.open("hex_fixture.e")751 scalarDimension = 0752 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)753 breaker = Refiner(eMesh, HEX8_HEX27_1, proc_rank_field)754 eMesh.commit()755 eMesh.print_info()756 eMesh.save_as("hex8_hex27_0.e")757 breaker.doBreak()758 eMesh.save_as("hex8_27_1.e")759 if p_size == 1 or p_size == 3:760 eMesh = PerceptMesh(3)761 eMesh.open("hex8_27_1.e")762 scalarDimension = 0763 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)764 breaker = Refiner(eMesh, HEX27_HEX27_8, proc_rank_field)765 eMesh.commit()766 breaker.setRemoveOldElements(True)767 breaker.doBreak()768 eMesh.save_as("hex8_hex27_hex27_1.e")769 def test_wedge6_2(self):770 fixture_setup()771 eMesh = PerceptMesh(3)772 p_size = eMesh.get_parallel_size()773 if p_size == 1:774 wedgeFixture = WedgeFixture()775 wedgeFixture.createMesh(MPI.COMM_WORLD, 4,3,2,0,1,0,1,0,1, "swept_wedge_0.e")776 eMesh.open("swept_wedge_0.e")777 scalarDimension = 0778 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)779 breaker = Refiner(eMesh, WEDGE6_WEDGE6_8, proc_rank_field)780 eMesh.commit()781 breaker.doBreak()782 eMesh.save_as("swept-wedge_1.e")783 784 def test_wedge6_enrich_1(self):785 fixture_setup()786 eMesh = PerceptMesh(3)787 p_size = eMesh.get_parallel_size()788 if p_size == 1:789 wedgeFixture = WedgeFixture()790 wedgeFixture.createMesh(MPI.COMM_WORLD, 4,3,2,0,1,0,1,0,1, "swept_wedge_enrich_0.e")791 eMesh.open("swept_wedge_enrich_0.e")792 scalarDimension = 0793 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)794 breaker = Refiner(eMesh, WEDGE6_WEDGE15_1, proc_rank_field)795 eMesh.commit()796 breaker.doBreak()797 eMesh.save_as("swept-wedge_enrich_1.e")798 eMesh.save_as("swept-wedge_enrich_refine_0.e")799 def test_wedge6_enrich_refine(self):800 fixture_setup()801 p_size = parallel_machine_size(MPI.COMM_WORLD)802 if p_size == 1:803 eMesh = PerceptMesh(3)804 wedgeFixture = WedgeFixture()805 wedgeFixture.createMesh(MPI.COMM_WORLD, 4,2,2,0,1,0,1,0,1, "tmp-swept-wedge_enrich_0.e")806 eMesh.open("tmp-swept-wedge_enrich_0.e")807 scalarDimension = 0808 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)809 breaker = Refiner(eMesh, WEDGE6_WEDGE15_1, proc_rank_field)810 eMesh.commit()811 breaker.doBreak()812 eMesh.save_as("swept-wedge_2_enrich_refine_0.e")813 if p_size == 1:814 eMesh = PerceptMesh(3)815 eMesh.open("swept-wedge_2_enrich_refine_0.e")816 scalarDimension = 0817 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)818 breaker = Refiner(eMesh, WEDGE15_WEDGE15_8, proc_rank_field)819 eMesh.commit()820 breaker.setIgnoreSideSets(True)821 breaker.doBreak()822 eMesh.save_as("swept-wedge_2_enrich_refine_1.e")823 def test_heterogeneous_mesh(self):824 fixture_setup()825 pm = MPI.COMM_WORLD826 p_size = parallel_machine_size(MPI.COMM_WORLD)827 if p_size <= 1:828 mesh = HeterogeneousFixture(MPI.COMM_WORLD, False)829 #put_io_part_attribute(mesh.m_block_hex)830 #put_io_part_attribute(mesh.m_block_wedge)831 #put_io_part_attribute(mesh.m_block_tet)832 mesh.m_metaData.commit()833 mesh.populate()834 isCommited = True835 em1 = PerceptMesh(mesh.m_metaData, mesh.m_bulkData, isCommited)836 em1.save_as("heterogeneous_0.e")837 em1.close()838 839 eMesh = PerceptMesh(3)840 eMesh.open("heterogeneous_0.e")841 scalarDimension = 0842 proc_rank_field = eMesh.add_field("proc_rank", stk::topology::ELEMENT_RANK, scalarDimension)843 #INCOMPLETE844 def test_wedge6_wedge18_enrich(self):845 pm = MPI.COMM_WORLD846 p_size = parallel_machine_size(pm)847 if p_size == 1:848 wedgeFixture = WedgeFixture()849 bulk = wedgeFixture.createMesh(MPI.COMM_WORLD, 4,3,2,0,1,0,1,0,1,"")850 eMesh = PerceptMesh(wedgeFixture.getMetaData(), bulk, False)...
Huffman.py
Source:Huffman.py
1from MinHeap import MinHeap,MinHeapNode2from copy import copy3from CharCounter import get_frequencies4import logging5class Decorator:6 @classmethod7 def has_extension(cls, filename):8 ##split by "\" , obtain last split(the thing after last "\")9 last_in_path = filename.split("\\").pop()10 ##split by . and get the last str after .11 ext = last_in_path.split('.').pop()12 ext = ext.strip()13 if ext == last_in_path or ext == '':14 return False15 if len(ext) > 0:16 return True17 @classmethod18 def get_extension(cls, filename):19 ##split by "\" , obtain last split(the thing after last "\")20 last_in_path = filename.split("\\").pop()21 ##split by . and get the last str after .22 ext = last_in_path.split('.').pop()23 ext = ext.strip()24 ##if result is '' ( meaning the last split after . is '') return None25 if ext == last_in_path or ext == '':26 return None27 return ext28 @classmethod29 def check_extension(cls, filename, extension):30 ##helper function to check if a string has last chars in the extension31 if extension == (filename[len(filename) - len(extension):len(filename)]):32 return True33 else:34 return False35 @classmethod36 def is_empty(cls, filename):37 ##if file has no extension, ext is empty space38 ext = cls.get_extension(filename)39 if ext == None:40 ext = ""41 else:42 ext = "." + ext43 var = filename.replace(ext, "")44 strip = var.strip()45 if strip == "":46 return True47 else:48 return False49 @classmethod50 def replace_extension(cls, filename, ext1, ext2):51 return filename.replace(ext1, ext2)52 @classmethod53 def ensure_proper_names_for_files(cls, to_decorate):54 def wrapper(*args,**kwargs):55 if "filename" in kwargs:56 filename = kwargs["filename"]57 else:58 filename = None59 if "save_as" in kwargs:60 save_as = kwargs["save_as"]61 else:62 save_as = None63 if "debug" in kwargs:64 debug = kwargs["filename"]65 else:66 debug = None67 print(filename)68 ##common for both compress and decompress functions69 if filename == None:70 logging.error(f"The filename is empty\n")71 return72 if save_as == None:73 save_as = ""74 if cls.is_empty(filename):75 logging.error(f"The filename is empty\n")76 return77 if cls.is_empty(save_as):78 logging.warning(79 f"The save_as parameter contains an empty file name.The encoded file will be saved as 'filename'.huf\n")80 ext = cls.get_extension(filename)81 if ext == None:82 ext = ""83 save_as = filename.replace(ext, "")84 if debug == None:85 debug = False86 ##if compress function is called87 if to_decorate.__name__ == "compress":88 if cls.has_extension(filename):89 pass90 else:91 ##by default filename argument is treated as .txt if no extension is present92 filename = filename + ".txt"93 filename = filename.replace("..txt", ".txt")94 if cls.has_extension(save_as):95 if cls.check_extension(save_as, ".huf") == False:96 logging.error(f"Can't save {cls.get_extension(save_as)} as huf")97 return98 else:99 ##by default save_as argument is treated as .huf if no extension is present100 save_as = save_as + ".huf"101 save_as = save_as.replace("..huf", ".huf")102 if to_decorate.__name__ == "decompress":103 if cls.has_extension(filename):104 if cls.check_extension(filename, ".huf") == False:105 logging.error(f"Can't decompress{ cls.get_extension(save_as)}")106 return107 else:108 ##by default filename argument is treated as .huf if no extension is present109 logging.warning(f"File is treated as huf\n")110 filename = filename + ".huf"111 filename = filename.replace("..huf", ".huf")112 if cls.has_extension(save_as):113 pass114 else:115 ##by default save_as argument is treated as .huf if no extension is present116 save_as = save_as + ".txt"117 save_as = save_as.replace("..txt", ".txt")118 to_decorate(args[0],filename=filename, save_as=save_as, debug=debug)119 return wrapper120class HuffmanNode(MinHeapNode):121 def __init__(self,node,left,right):122 super().__init__(node)123 self.left = left124 self.right = right125 self.code = ''126 self.char = ''127class Huffman:128 def __init__(self):129 self.heap = MinHeap()130 self.root = None131 self.frequencies = {}132 self.translation_dict = {}133 @Decorator.ensure_proper_names_for_files134 def compress(self,filename ,save_as ,debug ):135 self.obtain_frequencies(filename,debug)136 ##create a list of Huffman nodes137 node_values = [HuffmanNode(value[0],0,0) for value in self.frequencies.values()]138 ##build a min heap of that list139 self.heap.build_heap(node_values)140 self.heap.print_heap()141 ##create the huffman tree142 self.create_huffman_tree()143 ##obtain codes and corresponding letter of each leaf node144 self.assign_codes_and_chars(self.root,debug)145 self.print_tree(self.root,debug)146 ##create a dictionary of structure char:prefix code147 self.setup_translation_dict(self.root)148 self.translate(filename,save_as,debug)149 @Decorator.ensure_proper_names_for_files150 def decompress(self,filename , save_as , debug ):151 ##function to decompress a .huf file152 ## first take one char -> obtain ascii integer value ->convert to binary and pad it with 0 until you have a 8-lenght string -153 ##do that for all chars in text then put all 8 bit strings together154 ##init node with the root -> if value is 0 go to root's left , if it is 1 go to root's right155 ##do that until a leaf node is reached,then return to root156 file = open(filename, 'r',encoding="utf-8")157 text = file.read()158 translated_text = ''159 for i,char in enumerate(text):160 if char == '\x0c':161 break162 integer_value = ord(char)163 binary_value = bin(integer_value)164 binary_value = binary_value[2:]165 if i != len(text) - 1:166 string_representation = self.add_leading_zeros(binary_value)167 if debug:168 print(f"debug Decompression: UTF-8 char: {char}, integer value: {integer_value}, bin value: {binary_value}, padded bin value: {string_representation}")169 else:170 string_representation = binary_value171 if debug:172 print("debug Decompression: last")173 print(f"debug Decompression: UTF-8 char: {char}, integer value: {integer_value}, bin value: {binary_value}, padded bin value: {string_representation}")174 translated_text = translated_text + string_representation175 if debug:176 print(f"debug Decompression: Original text {text}")177 print(f"debug Decompression: Translated text {translated_text}")178 decompressed_text = ''179 node = self.root180 ##add one additional char that signals the end181 translated_text = translated_text + 'e'182 for char in translated_text:183 #print(char)184 #print(node.char,node.code, node.left, node.right)185 if node.right == 0 :186 if debug:187 print("debug Decompression: reached leaf node ",node.char," node is back at root")188 decompressed_text = decompressed_text + node.char189 node = self.root190 if char == '0':191 if debug:192 print("debug Decompression: at 0 going left")193 node = node.left194 if char == '1':195 if debug:196 print("debug Decompression: at 1 going right")197 node = node.right198 if debug:199 print(decompressed_text)200 file = open(save_as, 'w', encoding="utf-8")201 file.write(decompressed_text)202 def add_leading_zeros(self,byte):203 if(len(byte)<8):204 no = 8 - len(byte)205 for i in range(0, no):206 byte = '0' + byte207 return byte208 return byte209 def translate(self,filename,save_as,debug):210 if debug:211 print(f"debug Compression -> Translation dict {self.translation_dict}")212 file = open(filename, 'r')213 text = file.read()214 translated_text = ''215 ##we take each char from original file and replace it with its prefix code216 for char in text:217 if char == '\x0c':218 break219 translated_text = translated_text + self.translation_dict[char]220 if debug:221 print(f"debug Compression: Original text {text}")222 print(f"debug Compression: Translated text {translated_text}")223 ##we break the text into 8 value strings224 byte_array = self.bytes(translated_text, 8)225 ##we convert the 8 value string into integer, then into char (int value represents the ascii value)226 with open(save_as, "w",encoding="utf-8") as f:227 for byte in byte_array:228 int_val = int(byte, 2)229 char_val = chr(int_val)230 f.write(char_val)231 if debug:232 print(f"debug Compression: bin value: {byte}, integer value: {int_val}, UTF-8 char: {char_val}")233 def bytes(self,list, size ):234 ##helper function to separate binary string into bytes235 return (list[pos:pos + size] for pos in range(0, len(list), size))236 def obtain_frequencies(self,filename,debug):237 ##the count of each distinct symbol in the file238 result = get_frequencies(filename)239 self.frequencies = copy(result)240 if debug:241 print(f"debug Frequencies{self.frequencies}")242 def assign_char(self, node , debug):243 ##this function is used to add char value to Huffman node244 for key in self.frequencies.keys():245 ##the vaalue of key in dict is a list with 2 elems consisting of char frequence and : 1 if char was already assigned, 1 otherwise246 if debug:247 print(f"debug assign char {node.value} {key} {self.frequencies[key][0]} {self.frequencies[key][1]}")248 if self.frequencies[key][0] == node.value and self.frequencies[key][1] == 0:249 node.char = key250 self.frequencies[key][1] = 1251 return252 def create_huffman_tree(self):253 while( len(self.heap.heap) > 1 ):254 ##we extract two values from MinHeap and create two nodes255 ##all leaf nodes have their children 0256 l = self.heap.extract()257 r = self.heap.extract()258 parent_node = HuffmanNode(l.value + r.value,l,r)259 ##push node's value into MinHeap260 self.heap.insert_nodes(parent_node)261 ##print(f"created node {parent_node.value} of {l.value} and {r.value} extracted")262 self.root = self.heap.extract()263 self.root.left = l264 self.root.right = r265 def assign_codes_and_chars(self,node,debug):266 if node.right == 0:267 self.assign_char(node,debug)268 if debug:269 print(f"debug {node.value} {node.char}")270 if hasattr(node.left, 'value'):271 node.left.code = node.code + '0'272 self.assign_codes_and_chars(node.left,debug)273 if hasattr(node.right,'value'):274 node.right.code = node.code + '1'275 self.assign_codes_and_chars(node.right,debug)276 def setup_translation_dict(self,node):277 if node.right == 0:278 self.translation_dict[node.char] = node.code279 if hasattr(node.left, 'value'):280 self.setup_translation_dict(node.left)281 if hasattr(node.right,'value'):282 self.setup_translation_dict(node.right)283 def print_tree(self,node,debug):284 if debug:285 pass286 else:287 return288 print(f"debug tree: \n node: {node.value} , code : {node.code}")289 if hasattr(node.left, 'value') and hasattr(node.right, 'value'):290 print(f" children :{node.left.value},{node.right.value}")291 if node.right == 0:292 print(f"reached end at {node.value} {node.char}")293 if hasattr(node.left, 'value'):294 self.print_tree(node.left,debug)295 if hasattr(node.right,'value'):...
util.py
Source:util.py
1"""Helper functions that don't involve arcpy"""23import os4import fnmatch56import urllib27from urlparse import urlparse89# for downloading files10import time11import dateutil12import dateutil.parser13import datetime1415def ensure_dir(dir):16 """Make sure directory exists"""17 if not os.path.exists(dir):18 os.makedirs(dir)19 # return dir so we can chain this20 return dir212223def try_remove(file):24 """Try to delete file but don't worry if it doesn't work"""25 try:26 os.remove(file)27 except:28 pass293031def find_files(folder, file_mask='*'):32 """Find all files matching mask recursively in folder"""33 # find all files matching mask in folder recursively and return full paths34 paths = []35 for root, dirs, files in os.walk(folder):36 for file in fnmatch.filter(files, file_mask):37 real_folder = apply(os.path.join, root.split('/'))38 path = os.path.join(real_folder, file)39 paths.append(path)40 return paths4142def find_dirs(folder, file_mask='*'):43 """Find all directories matching mask recursively in folder"""44 paths = []45 for root, dirs, files in os.walk(folder):46 for dir in fnmatch.filter(dirs, file_mask):47 real_folder = apply(os.path.join, root.split('/'))48 path = os.path.join(real_folder, dir)49 # HACK: if subdirectory matches then don't match parent50 if root in paths:51 paths.remove(root)52 paths.append(path)53 return paths545556# taken from previous code57def fix_timezone_offset(d):58 """ Convert from UTC to local time, respecting DST if required """59 local_offset = time.timezone60 if time.daylight:61 if time.localtime().tm_isdst > 0:62 local_offset = time.altzone63 localdelta = datetime.timedelta(seconds=-local_offset)64 # convert to local time so other ftp programs would produce same result65 return (d + localdelta).replace(tzinfo=None)666768def save_http(to_dir, url, save_as=None, mode='wb', ignore_existing=False):69 """ Save file at given URL into given directory using an HTTP connection """70 logging.debug("Saving {}".format(url))71 if save_as is None:72 save_as = os.path.join(to_dir, os.path.basename(url))73 if ignore_existing and os.path.exists(save_as):74 logging.debug('Ignoring existing file')75 return save_as76 ensure_dir(to_dir)77 # we want to keep modified times matching on both ends78 do_save = True79 response = urllib2.urlopen(url)80 modlocal = None81 if 'last-modified' in response.headers.keys():82 mod = response.headers['last-modified']83 modtime = dateutil.parser.parse(mod)84 modlocal = fix_timezone_offset(modtime)85 # if file exists then compare mod times86 if os.path.isfile(save_as):87 filetime = os.path.getmtime(save_as)88 filedatetime = datetime.datetime.fromtimestamp(filetime)89 do_save = modlocal != filedatetime90 # NOTE: need to check file size too? Or should it not matter because we91 # only change the timestamp after it's fully written92 if do_save:93 logging.info("Downloading {}".format(save_as))94 try:95 with open(save_as, mode) as f:96 f.write(response.read())97 except:98 try_remove(save_as)99 raise100 if modlocal is not None:101 tt = modlocal.timetuple()102 usetime = time.mktime(tt)103 os.utime(save_as, (usetime, usetime))104 return save_as105106107def save_ftp(to_dir, url, user="anonymous", password=""):108 """ Save file at given URL into given directory using an FTP connection """109 urlp = urlparse(url)110 folder = os.path.dirname(urlp.path)111 site = urlp.netloc112 filename = os.path.basename(urlp.path)113 logging.debug("Saving {}".format(filename))114 import ftplib115 ftp = ftplib.FTP(site)116 ftp.login(user, password)117 ftp.cwd(folder)118 save_as = os.path.join(to_dir, filename)119 do_save = True120 ftptime = ftp.sendcmd('MDTM {}'.format(filename))121 ftpdatetime = datetime.datetime.strptime(ftptime[4:], '%Y%m%d%H%M%S')122 ftplocal = fix_timezone_offset(ftpdatetime)123 # if file exists then compare mod times124 if os.path.isfile(save_as):125 filetime = os.path.getmtime(save_as)126 filedatetime = datetime.datetime.fromtimestamp(filetime)127 do_save = ftplocal != filedatetime128 # NOTE: need to check file size too? Or should it not matter because we129 # only change the timestamp after it's fully written130 if do_save:131 logging.debug("Downloading {}".format(filename))132 with open(save_as, 'wb') as f:133 ftp.retrbinary('RETR {}'.format(filename), f.write)134 tt = ftplocal.timetuple()135 usetime = time.mktime(tt)136 os.utime(save_as, (usetime, usetime))
...
helper.py
Source:helper.py
1from PIL import Image2from PIL import ImageColor,ImageFilter3from PIL import ImageDraw, ImageFont4import requests5def show(path):6 Image.open(path).show()7def create_with_color(path, color='blue', size =(300,300)):8 image = Image.new('RGBA',size, color)9 image.save(path)10def create_transparent(path, size=(500,500)):11 image = Image.new('RGBA',size)12 image.save(path)13def crop(file, rect, save_as):14 img = Image.open(file)15 im1 = img.crop(rect)16 im1.save(save_as)17def merge(from_img, into_img, save_as, position):18 img_from = Image.open(from_img)19 img_into = Image.open(into_img).copy()20 img_into.paste(img_from, position)21 img_into.save(save_as)22 23def resize(file_img, percent, save_as):24 img = Image.open(file_img)25 w,h = img.size26 #print(w,h)27 img1 = img.resize((int(w*percent), int(h*percent)))28 img1.save(save_as)29def rotate(file_image, degree, save_as):30 img = Image.open(file_image)31 img2=img.rotate(degree)32 img2.save(save_as)33def flip_horizontal(file_image, save_as):34 img = Image.open(file_image)35 img.transpose(Image.FLIP_LEFT_RIGHT).save(save_as)36def flip_vertical(file_image, save_as):37 img = Image.open(file_image)38 img.transpose(Image.FLIP_TOP_BOTTOM).save(save_as)39def fill_color(file_image,save_as, color='red', left_top=(0,0), width_height=(100,100)):40 img = Image.open(file_image)41 for x in range(width_height[0]):42 for y in range(width_height[1]):43 x1,y1=x+left_top[0],y+left_top[1]44 img.putpixel((x1,y1), ImageColor.getcolor(color, 'RGBA') )45 img.save(save_as)46def watermark(file_image, save_as, text, xy):47 im = Image.open(file_image)48 draw = ImageDraw.Draw(im)49 font=ImageFont.truetype('arial.ttf', size=20)50 draw.text(xy, text, fill='gray', font=font)51 im.save(save_as)52def blur(file_image, save_as):53 im = Image.open(file_image)54 im1 = im.filter(ImageFilter.BLUR)55 im1.save(save_as)56def gray_scale(file_image, save_as):57 im = Image.open(file_image)58 im1 = im.convert('L')59 im1.save(save_as)60def download(url, save_as, format='png'):61 resp = requests.get(url, stream=True).raw62 img = Image.open(resp)63 img.save(save_as, format)64def draw_sample_lines(save_as, size=(500,500), color='white'):65 im = Image.new('RGBA',size,color)66 draw = ImageDraw.Draw(im)67 lines=[(50,50)]68 x=30069 while x>0:70 lines.append((lines[-1][0]+x,lines[-1][1]))71 lines.append((lines[-1][0],lines[-1][1]+x))72 lines.append((lines[-1][0]-x,lines[-1][1]))73 lines.append((lines[-1][0],lines[-1][1]-x))74 x-=2075 draw.line(lines, fill='black')76 im.save(save_as)77def draw_sample_rect(save_as, size=(500,500), color='white'):78 im = Image.new('RGBA',size,color)79 draw = ImageDraw.Draw(im)80 p = [0,0]81 inc=1082 w=1083 color = ['red','blue','green','black','orange']84 for i in range(7):85 print(p,w)86 draw.rectangle((p[0],p[1],p[0]+w,p[1]+w),fill=color[i%5])87 p[0],p[1] = p[0]+inc,p[1]+inc88 w += inc89 inc+=1090 im.save(save_as)91def draw_sample_circle(save_as, size=(500,500), color='white'):92 im = Image.new('RGBA',size,color)93 draw = ImageDraw.Draw(im)94 p = [120,30]95 draw.ellipse((p[0],p[1],160,60), fill='red')96 97 points = [(10,10)]98 step=1099 for i in range(0,10):100 points.append((points[-1][0]+step,points[-1][1]+step))101 points.append((points[-1][0]+step,points[-1][1]-step))102 draw.polygon(points, fill='green')103 im.save(save_as)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run green automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
