Changeset 3292

Timestamp:

Jul 7, 2006, 4:09:39 PM (18 years ago)

Author:

duncan

Message:

Adding class Exposure_csv, to interact with exposure csv files.

Location:

inundation

Files:

• coordinate_transforms/redfearn.py (modified) (1 diff)
• coordinate_transforms/test_redfearn.py (modified) (1 diff)
• geospatial_data/test_geospatial_data.py (modified) (5 diffs)
• pyvolution/data_manager.py (modified) (8 diffs)
• pyvolution/test_data_manager.py (modified) (1 diff)

inundation/coordinate_transforms/redfearn.py

@@ -178 +178 @@
     
     for lat, long in map(None, latitudes, longitudes,):
-        zone, easting, northing = redfearn(lat,
-                                           long,
+        zone, easting, northing = redfearn(float(lat),
+                                           float(long),
                                            false_easting=false_easting,
                                            false_northing=false_northing)

inundation/coordinate_transforms/test_redfearn.py

@@ -187 +187 @@
             self.failUnless(0 ==1,
                         'Error not thrown error!')
-
+            
+    def test_convert_lats_longs3(self):
+
+        #Site Name:    GDA-MGA: (UTM with GRS80 ellipsoid) 
+        #Zone:   56    
+        #Easting:  222908.705  Northing: 6233785.284 
+        #Latitude:   -34  0 ' 0.00000 ''  Longitude: 150  0 ' 0.00000 '' 
+        #Grid Convergence:  -1  40 ' 43.13 ''  Point Scale: 1.00054660 
+
+        lat_gong = "-34.5"
+        lon_gong = "150.916666667"
+        lat_2 = degminsec2decimal_degrees(34,00,0.)
+        lon_2 = degminsec2decimal_degrees(100,00,0.)
+        
+        lats = [lat_gong, lat_2]
+        longs = [lon_gong, lon_2]
+        try:
+            zone, points = convert_lats_longs(lats, longs)
+        except ANUGAError:
+            pass
+        else:
+            self.failUnless(0 ==1,
+                        'Error not thrown error!')
 #-------------------------------------------------------------
 if __name__ == "__main__":

inundation/geospatial_data/test_geospatial_data.py

@@ -1274 +1274 @@
             pass
         else:
-            self.failUnless(0 ==1,
-                        'Error not thrown error!')
+            self.failUnless(0 ==1,  'Error not thrown error!')
         try:
             results = Geospatial_data(latitudes=lats)
@@ -1281 +1280 @@
             pass
         else:
-            self.failUnless(0 ==1,
-                        'Error not thrown error!')
+            self.failUnless(0 ==1,  'Error not thrown error!')
         try:
             results = Geospatial_data(longitudes=lats)
@@ -1288 +1286 @@
             pass
         else:
-            self.failUnless(0 ==1,
-                        'Error not thrown error!')
+            self.failUnless(0 ==1, 'Error not thrown error!')
         try:
             results = Geospatial_data(latitudes=lats, longitudes=longs,
@@ -1296 +1293 @@
             pass
         else:
-            self.failUnless(0 ==1,
-                        'Error not thrown error!')
+            self.failUnless(0 ==1,  'Error not thrown error!')
             
         try:
@@ -1305 +1301 @@
             pass
         else:
-            self.failUnless(0 ==1,
-                        'Error not thrown error!')
+            self.failUnless(0 ==1,  'Error not thrown error!')
             
 if __name__ == "__main__":

inundation/pyvolution/data_manager.py

@@ -50 +50 @@
 
 import exceptions
+class TitleValueError(exceptions.Exception): pass
 class DataMissingValuesError(exceptions.Exception): pass
 class DataFileNotOpenError(exceptions.Exception): pass
@@ -57 +58 @@
 
 
+import csv
 import os
 
@@ -64 +66 @@
 
 from coordinate_transforms.geo_reference import Geo_reference
+from geospatial_data.geospatial_data import Geospatial_data
 
 
@@ -646 +649 @@
 #Function for storing xya output
 #FIXME Not done yet for this version
+#This is obsolete.  Use geo_spatial_data instead
 class Data_format_xya(Data_format):
     """Generic interface to data formats
@@ -715 +719 @@
         """
         pass
+
+
+#### NBED national exposure database
+    
+LAT_TITLE = 'LATITUDE'
+LONG_TITLE = 'LONGITUDE'
+class Exposure_csv:
+    def __init__(self,file_name, latitude_title=LAT_TITLE,
+                 longitude_title=LONG_TITLE, refine_polygon=None):
+        """
+        Not for adding and deleting rows
+        Can geospatial handle string attributes? It's not made for them.
+        Currently it can't load and save string att's.
+
+        So just use geospatial to hold the x, y and georef? Bad, since
+        different att's are in diferent structures.  Not so bad, the info
+        to write if the .csv file is saved is in attribute_dic
+
+        The location info is in the geospatial attribute.
+        
+        
+        """
+        self._file_name = file_name
+        self._geospatial = None #
+
+        # self._attribute_dic is a dictionary.
+        #The keys are the column titles.
+        #The values are lists of column data
+        
+        # self._title_index_dic is a dictionary.
+        #The keys are the column titles.
+        #The values are the index positions of file columns.
+        self._attribute_dic, self._title_index_dic = \
+        self._load_exposure_csv(self._file_name)
+        try:
+            lats = self._attribute_dic[latitude_title]
+            longs = self._attribute_dic[longitude_title]
+            
+        except KeyError:
+            # maybe a warning..
+            pass
+        else:
+            self._geospatial = Geospatial_data(latitudes = lats,
+                 longitudes = longs)
+            
+        # create a list of points that are in the refining_polygon
+        # described by a list of indexes representing the points
+
+    def __cmp__(self, other):
+        #print "self._attribute_dic",self._attribute_dic
+        #print "other._attribute_dic",other._attribute_dic
+        #print "self._title_index_dic", self._title_index_dic
+        #print "other._title_index_dic", other._title_index_dic
+        
+        #check that a is an instance of this class
+        if isinstance(self, type(other)):
+            result = cmp(self._attribute_dic, other._attribute_dic)
+            if result <>0:
+                return result
+            # The order of the columns is important. Therefore.. 
+            result = cmp(self._title_index_dic, other._title_index_dic)
+            if result <>0:
+                return result
+            for self_ls, other_ls in map(None,self._attribute_dic, \
+                    other._attribute_dic):
+                result = cmp(self._attribute_dic[self_ls],
+                             other._attribute_dic[other_ls])
+                if result <>0:
+                    return result
+            return 0
+        else:
+            return 1
+    
+    def _load_exposure_csv(self, file_name):
+        """
+        Load in the csv as a dic, title as key and column info as value, .
+        Also, create a dic, title as key and column index as value. 
+        """
+        #
+        attribute_dic = {}
+        title_index_dic = {}
+        titles_stripped = [] # list of titles
+        
+        reader = csv.reader(file(file_name))
+
+        # Read in and manipulate the title info
+        titles = reader.next()
+        for i,title in enumerate(titles):
+            titles_stripped.append(title.strip())
+            title_index_dic[title.strip()] = i
+        title_count = len(titles_stripped)       
+        #print "title_index_dic",title_index_dic
+
+        
+        #create a dic of colum values, indexed by column title
+        for line in reader:
+            if len(line) <> title_count:
+                raise IOError #FIXME make this nicer
+            for i, value in enumerate(line):
+                attribute_dic.setdefault(titles_stripped[i],[]).append(value)
+            
+        return attribute_dic, title_index_dic
+
+    def get_column(self, column_name, use_refind_polygon=False):
+        """
+        Given a column name return a list of the column values
+
+        Note, the type of the values will be String!
+        Thinking of semi-hard coding this to not be so..
+        
+        Not implemented:
+        if use_refind_polygon is True, only return values in the
+        refined polygon
+        """
+        if not self._attribute_dic.has_key(column_name):
+            msg = 'Therer is  no column called %s!' %column_name
+            raise TitleValueError, msg
+        return self._attribute_dic[column_name]
+
+    def get_location(self, use_refind_polygon=False):
+        """
+        Return a geospatial object which describes the
+        locations of the location file.
+
+        Note, if there is not location info, this returns None.
+        
+        Not implemented:
+        if use_refind_polygon is True, only return values in the
+        refined polygon
+        """
+        return self._geospatial
+
+    def set_column(self, column_name, column_values, overwrite=False):
+        """
+        Add a column to the 'end' (with the right most column being the end)
+        of the csv file.
+
+        Set overwrite to True if you want to overwrite a column.
+        
+        Note, in column_name white space is removed and case is not checked.
+        Precondition
+        The column_name and column_values cannot have comma's in it.
+        """
+        
+        value_row_count = \
+            len(self._attribute_dic[self._title_index_dic.keys()[0]])
+        if len(column_values) <> value_row_count: 
+            msg = 'The number of column values must equal the number of rows.'
+            raise DataMissingValuesError, msg
+        
+        if self._attribute_dic.has_key(column_name):
+            if not overwrite:
+                msg = 'Column name %s already in use!' %column_name
+                raise TitleValueError, msg
+        else:
+            # New title.  Add it to the title index.
+            self._title_index_dic[column_name] = len(self._title_index_dic)
+        self._attribute_dic[column_name] = column_values
+        #print "self._title_index_dic[column_name]",self._title_index_dic 
+
+    def save(self, file_name=None):
+        """
+        Save the exposure csv file
+        """
+        if file_name is None:
+            file_name = self._file_name
+        
+        fd = open(file_name,'wb')
+        writer = csv.writer(fd)
+        
+        #Write the title to a cvs file
+        line = [None]* len(self._title_index_dic)
+        for title in self._title_index_dic.iterkeys():
+            line[self._title_index_dic[title]]= title
+        writer.writerow(line)
+        
+        # Write the values to a cvs file
+        value_row_count = \
+            len(self._attribute_dic[self._title_index_dic.keys()[0]])
+        for row_i in range(value_row_count):
+            line = [None]* len(self._title_index_dic)
+            for title in self._title_index_dic.iterkeys():
+                line[self._title_index_dic[title]]= \
+                     self._attribute_dic[title][row_i]
+            writer.writerow(line)
 
 
@@ -951 +1140 @@
     return cls(domain, mode)
 
+#FIXME move into geospatial.  There should only be one method that
+# reads xya, and writes pts.
 def xya2pts(basename_in, basename_out=None, verbose=False,
             #easting_min=None, easting_max=None,
@@ -3301 +3492 @@
     """
 
-    #FIXME Move to data_manager
     from shallow_water import Domain
     from pmesh2domain import pmesh_to_domain_instance
@@ -3802 +3992 @@
 
     return
+
+#-------------------------------------------------------------
+if __name__ == "__main__":
+    import csv
+    reader = csv.reader(file("some.csv"))
+    heading = reader.next()
+    clean_heading = heading.strip()
+    print "heading",heading 
+
+    #a = csv.DictReader("some.csv")
+    
+
+
+

inundation/pyvolution/test_data_manager.py

@@ -4659 +4659 @@
 
 
-
+    def test_exposure_csv_loading(self):
+        
+
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LATITUDE, LONGITUDE ,sound  , speed \n\
+115.0, -21.0, splat, 0.0\n\
+114.0, -21.7, pow, 10.0\n\
+114.5, -21.4, bang, 40.0\n")
+        file.close()
+        exposure = Exposure_csv(file_name)
+        exposure.get_column("sound")
+       
+        self.failUnless(exposure._attribute_dic['sound'][2]==' bang',
+                        'FAILED!')
+        self.failUnless(exposure._attribute_dic['speed'][2]==' 40.0',
+                        'FAILED!')
+        
+        os.remove(file_name)
+        
+    def test_exposure_csv_loading(self):
+        
+
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LATITUDE, LONGITUDE ,sound  , speed \n\
+115.0, -21.0, splat, 0.0\n\
+114.0, -21.7, pow, 10.0\n\
+114.5, -21.4, bang, 40.0\n")
+        file.close()
+        exposure = Exposure_csv(file_name)
+        exposure.get_column("sound")
+       
+        self.failUnless(exposure._attribute_dic['sound'][2]==' bang',
+                        'FAILED!')
+        self.failUnless(exposure._attribute_dic['speed'][2]==' 40.0',
+                        'FAILED!')
+        
+        os.remove(file_name)
+
+    def test_exposure_csv_cmp(self):
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LATITUDE, LONGITUDE ,sound  , speed \n\
+115.0, -21.0, splat, 0.0\n\
+114.0, -21.7, pow, 10.0\n\
+114.5, -21.4, bang, 40.0\n")
+        file.close()
+        
+        e1 = Exposure_csv(file_name)
+        e2 = Exposure_csv(file_name)
+        os.remove(file_name)
+
+        self.failUnless(cmp(e1,e2)==0,
+                        'FAILED!')
+        
+        self.failUnless(cmp(e1,"hey")==1,
+                        'FAILED!')
+        
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        # Note, this has less spaces in the title,
+        # the instances will be the same.
+        file.write("LATITUDE,LONGITUDE ,sound, speed \n\
+115.0, -21.0, splat, 0.0\n\
+114.0, -21.7, pow, 10.0\n\
+114.5, -21.4, bang, 40.0\n")
+        file.close()
+        e3 = Exposure_csv(file_name)
+        os.remove(file_name)
+
+        self.failUnless(cmp(e3,e2)==0,
+                        'FAILED!')
+        
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        # Note, 40 changed to 44 .
+        file.write("LATITUDE,LONGITUDE ,sound, speed \n\
+115.0, -21.0, splat, 0.0\n\
+114.0, -21.7, pow, 10.0\n\
+114.5, -21.4, bang, 44.0\n")
+        file.close()
+        e4 = Exposure_csv(file_name)
+        os.remove(file_name)
+        #print "e4",e4._attribute_dic 
+        #print "e2",e2._attribute_dic 
+        self.failUnless(cmp(e4,e2)<>0,
+                        'FAILED!')
+        
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        # Note, the first two columns are swapped.
+        file.write("LONGITUDE,LATITUDE ,sound, speed \n\
+ -21.0,115.0, splat, 0.0\n\
+ -21.7,114.0, pow, 10.0\n\
+ -21.4,114.5, bang, 40.0\n")
+        file.close()
+        e5 = Exposure_csv(file_name)
+        os.remove(file_name)
+
+        self.failUnless(cmp(e3,e5)<>0,
+                        'FAILED!')
+        
+    def test_exposure_csv_saving(self):
+        
+
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LATITUDE, LONGITUDE ,sound  , speed \n\
+115.0, -21.0, splat, 0.0\n\
+114.0, -21.7, pow, 10.0\n\
+114.5, -21.4, bang, 40.0\n")
+        file.close()
+        e1 = Exposure_csv(file_name)
+        
+        file_name2 = tempfile.mktemp(".xya")
+        e1.save(file_name = file_name2)
+        e2 = Exposure_csv(file_name2)
+       
+        self.failUnless(cmp(e1,e2)==0,
+                        'FAILED!')
+        os.remove(file_name)
+        os.remove(file_name2)
+
+    def test_exposure_csv_get_location(self):
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LONGITUDE , LATITUDE, sound  , speed \n\
+150.916666667, -34.5, splat, 0.0\n\
+150.0, -34.0, pow, 10.0\n")
+        file.close()
+        e1 = Exposure_csv(file_name)
+
+        gsd = e1.get_location()
+        
+        points = gsd.get_data_points(absolute=True)
+        
+        assert allclose(points[0][0], 308728.009)
+        assert allclose(points[0][1], 6180432.601)
+        assert allclose(points[1][0],  222908.705)
+        assert allclose(points[1][1], 6233785.284)
+        self.failUnless(gsd.get_geo_reference().get_zone() == 56,
+                        'Bad zone error!')
+
+        os.remove(file_name)
+        
+    def test_exposure_csv_set_column_get_column(self):
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LONGITUDE , LATITUDE, sound  , speed \n\
+150.916666667, -34.5, splat, 0.0\n\
+150.0, -34.0, pow, 10.0\n")
+        file.close()
+        e1 = Exposure_csv(file_name)      
+        os.remove(file_name)
+
+        new_title = "feast"
+        new_values = ["chicken","soup"]
+        e1.set_column(new_title, new_values)
+        returned_values = e1.get_column(new_title)
+        self.failUnless(returned_values == new_values,
+                        ' Error!')
+        
+        file_name2 = tempfile.mktemp(".xya")
+        e1.save(file_name = file_name2)
+        e2 = Exposure_csv(file_name2)
+        returned_values = e2.get_column(new_title)
+        self.failUnless(returned_values == new_values,
+                        ' Error!')       
+        os.remove(file_name2)
+
+    def test_exposure_csv_set_column_get_column_error_checking(self):
+        file_name = tempfile.mktemp(".xya")
+        file = open(file_name,"w")
+        file.write("LONGITUDE , LATITUDE, sound  , speed \n\
+150.916666667, -34.5, splat, 0.0\n\
+150.0, -34.0, pow, 10.0\n")
+        file.close()
+        e1 = Exposure_csv(file_name)      
+        os.remove(file_name)
+
+        new_title = "sound"
+        new_values = [12.5,7.6]
+        try:
+            e1.set_column(new_title, new_values)
+        except TitleValueError:
+            pass
+        else:
+            self.failUnless(0 ==1,  'Error not thrown error!')
+            
+        e1.set_column(new_title, new_values, overwrite=True)
+        returned_values = e1.get_column(new_title)
+        self.failUnless(returned_values == new_values,
+                        ' Error!')       
+        
+        new2_title = "short list"
+        new2_values = [12.5]
+        try:
+            e1.set_column(new2_title, new2_values)
+        except DataMissingValuesError:
+            pass
+        else:
+            self.failUnless(0 ==1,  'Error not thrown error!')
+            
+        new2_title = "long list"
+        new2_values = [12.5, 7,8]
+        try:
+            e1.set_column(new2_title, new2_values)
+        except DataMissingValuesError:
+            pass
+        else:
+            self.failUnless(0 ==1,  'Error not thrown error!')
+        file_name2 = tempfile.mktemp(".xya")
+        e1.save(file_name = file_name2)
+        e2 = Exposure_csv(file_name2)
+        returned_values = e2.get_column(new_title)
+        for returned, new in map(None, returned_values, new_values):
+            self.failUnless(returned == str(new), ' Error!')
+        #self.failUnless(returned_values == new_values, ' Error!')       
+        os.remove(file_name2)
+        
+        try:
+            e1.get_column("toe jam")
+        except TitleValueError:
+            pass
+        else:
+            self.failUnless(0 ==1,  'Error not thrown error!')
+        
+        
 #-------------------------------------------------------------
 if __name__ == "__main__":
+    #suite = unittest.makeSuite(Test_Data_Manager,'test_exposure_csv')
     suite = unittest.makeSuite(Test_Data_Manager,'test')
     runner = unittest.TextTestRunner()