Changeset 4936
- Timestamp:
- Jan 15, 2008, 12:20:54 PM (17 years ago)
- Location:
- anuga_core/source/anuga/abstract_2d_finite_volumes
- Files:
-
- 2 edited
Legend:
- Unmodified
- Added
- Removed
-
anuga_core/source/anuga/abstract_2d_finite_volumes/test_util.py
r4935 r4936 1627 1627 os.remove(point2_filename) 1628 1628 1629 1630 def test_sww2csv_gauges2(self): 1631 1632 def elevation_function(x, y): 1633 return -x 1634 1635 """Most of this test was copied from test_interpolate test_interpole_sww2csv 1636 1637 This is testing the gauge_sww2csv function, by creating a sww file and 1638 then exporting the gauges and checking the results. 1639 1640 This is the same as sww2csv_gauges except set domain.set_starttime to 5. Therefore 1641 testing the storing of the absolute time in the csv files 1642 """ 1643 1644 # create mesh 1645 mesh_file = tempfile.mktemp(".tsh") 1646 points = [[0.0,0.0],[6.0,0.0],[6.0,6.0],[0.0,6.0]] 1647 m = Mesh() 1648 m.add_vertices(points) 1649 m.auto_segment() 1650 m.generate_mesh(verbose=False) 1651 m.export_mesh_file(mesh_file) 1652 1653 #Create shallow water domain 1654 domain = Domain(mesh_file) 1655 os.remove(mesh_file) 1656 1657 domain.default_order=2 1658 domain.beta_h = 0 1659 1660 #Set some field values 1661 domain.set_quantity('elevation', elevation_function) 1662 domain.set_quantity('friction', 0.03) 1663 domain.set_quantity('xmomentum', 3.0) 1664 domain.set_quantity('ymomentum', 4.0) 1665 domain.set_starttime(5) 1666 1667 ###################### 1668 # Boundary conditions 1669 B = Transmissive_boundary(domain) 1670 domain.set_boundary( {'exterior': B}) 1671 1672 # This call mangles the stage values. 1673 domain.distribute_to_vertices_and_edges() 1674 domain.set_quantity('stage', 1.0) 1675 1676 1677 1678 domain.set_name('datatest' + str(time.time())) 1679 domain.format = 'sww' 1680 domain.smooth = True 1681 domain.reduction = mean 1682 1683 sww = get_dataobject(domain) 1684 sww.store_connectivity() 1685 sww.store_timestep(['stage', 'xmomentum', 'ymomentum','elevation']) 1686 domain.set_quantity('stage', 10.0) # This is automatically limited 1687 # so it will not be less than the elevation 1688 domain.time = 2. 1689 sww.store_timestep(['stage','elevation', 'xmomentum', 'ymomentum']) 1690 1691 # test the function 1692 points = [[5.0,1.],[0.5,2.]] 1693 1694 points_file = tempfile.mktemp(".csv") 1695 # points_file = 'test_point.csv' 1696 file_id = open(points_file,"w") 1697 file_id.write("name, easting, northing, elevation \n\ 1698 point1, 5.0, 1.0, 3.0\n\ 1699 point2, 0.5, 2.0, 9.0\n") 1700 file_id.close() 1701 1702 1703 sww2csv_gauges(sww.filename, 1704 points_file, 1705 verbose=False, 1706 use_cache=False) 1707 1708 # point1_answers_array = [[0.0,1.0,-5.0,3.0,4.0], [2.0,10.0,-5.0,3.0,4.0]] 1709 point1_answers_array = [[5.0,1.0,6.0,-5.0,3.0,4.0], [7.0,10.0,15.0,-5.0,3.0,4.0]] 1710 point1_filename = 'gauge_point1.csv' 1711 point1_handle = file(point1_filename) 1712 point1_reader = reader(point1_handle) 1713 point1_reader.next() 1714 1715 line=[] 1716 for i,row in enumerate(point1_reader): 1717 #print 'i',i,'row',row 1718 line.append([float(row[0]),float(row[1]),float(row[2]),float(row[3]),float(row[4]),float(row[5])]) 1719 #print 'assert line',line[i],'point1',point1_answers_array[i] 1720 assert allclose(line[i], point1_answers_array[i]) 1721 1722 point2_answers_array = [[5.0,1.0,1.5,-0.5,3.0,4.0], [7.0,10.0,10.5,-0.5,3.0,4.0]] 1723 point2_filename = 'gauge_point2.csv' 1724 point2_handle = file(point2_filename) 1725 point2_reader = reader(point2_handle) 1726 point2_reader.next() 1727 1728 line=[] 1729 for i,row in enumerate(point2_reader): 1730 #print 'i',i,'row',row 1731 line.append([float(row[0]),float(row[1]),float(row[2]),float(row[3]),float(row[4]),float(row[5])]) 1732 #print 'assert line',line[i],'point1',point1_answers_array[i] 1733 assert allclose(line[i], point2_answers_array[i]) 1734 1735 # clean up 1736 point1_handle.close() 1737 point2_handle.close() 1738 #print "sww.filename",sww.filename 1739 os.remove(sww.filename) 1740 os.remove(points_file) 1741 os.remove(point1_filename) 1742 os.remove(point2_filename) 1743 1744 1629 1745 def test_greens_law(self): 1630 1746 -
anuga_core/source/anuga/abstract_2d_finite_volumes/util.py
r4935 r4936 9 9 import os 10 10 11 from os import remove, mkdir, access, F_OK, R_OK, W_OK, sep, mkdir11 from os import remove, mkdir, access, F_OK, R_OK, W_OK, sep,getcwd 12 12 from os.path import exists, basename, split,join 13 13 from warnings import warn … … 2201 2201 attribute_dic, title_index_dic = csv2dict(dir_filename+ 2202 2202 '.csv') 2203 2204 2203 directory_start_time = directories_dic[directory][1] 2205 2204 directory_add_tide = directories_dic[directory][2] 2206 2205 2206 if verbose: print 'reading: %s.csv' %dir_filename 2207 2207 # print 'keys',attribute_dic.keys() 2208 2208 #add time to get values … … 2260 2260 quantities_axis[quantity][3], 2261 2261 quantities_label[quantity]) 2262 print quantities_axis[quantity]2262 #print quantities_axis[quantity] 2263 2263 2264 2264 cstr = ['b', 'r', 'g', 'c', 'm', 'y', 'k'] … … 2537 2537 2538 2538 dir_name, base = os.path.split(sww_file) 2539 # print 'dirname',dir_name, base 2539 #print 'dirname',dir_name, base 2540 #need to get current directory so when path and file 2541 #are "joined" below the directory is correct 2542 if dir_name == '': 2543 dir_name =getcwd() 2544 2540 2545 if access(sww_file,R_OK): 2541 2546 if verbose: print 'File %s exists' %(sww_file) … … 2574 2579 heading.insert(0,'time') 2575 2580 2576 # print heading, quantities2581 # print 'start time', callable_sww.starttime, heading, quantities 2577 2582 2578 2583 #create a list of csv writers for all the points and write header 2579 2584 points_writer = [] 2580 2585 for i,point in enumerate(points): 2586 #print 'gauge file:',dir_name+sep+'gauge_'+point_name[i]+'.csv' 2581 2587 points_writer.append(writer(file(dir_name+sep+'gauge_'+point_name[i]+'.csv', "wb"))) 2582 2588 points_writer[i].writerow(heading) 2583 2589 2590 2591 if verbose: print 'Writing csv files' 2592 2584 2593 for time in callable_sww.get_time(): 2585 # points_list = []2586 2594 2587 2595 for point_i, point in enumerate(points_array): 2588 points_list = [time] 2596 #add domain starttime to relative time. 2597 points_list = [time+callable_sww.starttime] 2589 2598 # print'time',time,'point_i',point_i,point, points_array 2590 2599 point_quantities = callable_sww(time,point_i) 2591 2600 # print "quantities", point_quantities 2592 2601 2593 # for i, quantity in enumerate(quantities):2594 # points_list.append(quantities)2595 2602 for quantity in quantities: 2596 2603 if quantity==NAN: 2597 if verbose:print 'quantity does not exist in' %callable_sww.get_name2604 print 'quantity does not exist in' %callable_sww.get_name 2598 2605 else: 2599 2606 if quantity == 'stage':
Note: See TracChangeset
for help on using the changeset viewer.