Source code for solaris.preproc.label

import geopandas as gpd
import pandas as pd
import shapely.geometry
import shapely.wkt

from .pipesegment import PipeSegment, LoadSegment, MergeSegment
from ..vector.polygon import convert_poly_coords


[docs]class LoadString(LoadSegment): """ Load a string from a file. """ def __init__(self, pathstring): super().__init__() self.pathstring = pathstring def load(self): infile = open(self.pathstring, 'r') content = infile.read() infile.close() return content
[docs]class SaveString(PipeSegment): """ Write a string to a file. """ def __init__(self, pathstring, append=False): super().__init__() self.pathstring = pathstring self.append = append def transform(self, pin): mode = 'a' if self.append else 'w' outfile = open(self.pathstring, mode) outfile.write(str(pin)) outfile.close() return pin
[docs]class ShowString(PipeSegment): """ Print a string to the screen. """ def transform(self, pin): print(pin) return pin
[docs]class LoadDataFrame(LoadSegment): """ Load a GeoPandas GeoDataFrame from a file. """ def __init__(self, pathstring, geom_col='geometry', projection=None): super().__init__() self.pathstring = pathstring self.geom_col = geom_col self.projection = projection def load(self): if self.pathstring.lower()[-4:] == '.csv': df = pd.read_csv(self.pathstring) geometry = df.apply(lambda row: shapely.wkt.loads(row[self.geom_col]), axis=1) df.drop(columns=[self.geom_col]) gdf = gpd.GeoDataFrame(df, geometry=geometry) if self.projection is not None: gdf.crs = 'epsg:' + str(self.projection) return gdf else: return gpd.read_file(self.pathstring)
[docs]class SaveDataFrame(PipeSegment): """ Save a GeoPandas GeoDataFrame to disk. """ def __init__(self, pathstring, driver='GeoJSON'): super().__init__() self.pathstring = pathstring self.driver = driver def transform(self, pin): pin.to_file(self.pathstring, driver=self.driver) return pin
[docs]class ShowDataFrame(PipeSegment): """ Print a GeoPandas GeoDataFrame to the screen. """ def transform(self, pin): print(pin) return pin
[docs]class ReprojectDataFrame(PipeSegment): """ Reproject a GeoPandas GeoDataFrame. """ def __init__(self, projection=3857): super().__init__() self.projection = projection def transform(self, pin): return pin.to_crs('epsg:' + str(self.projection))
[docs]class ExplodeDataFrame(PipeSegment): """ Given a GeoPandas GeoDataFrame, break multi-part geometries into multiple lines. """ def transform(self, pin): return pin.explode().reset_index()
[docs]class IntersectDataFrames(PipeSegment): """ Given an iterable of GeoPandas GeoDataFrames, returns their intersection """ def __init__(self, master=0): super().__init__() self.master = master def transform(self, pin): result = pin[self.master] for i, gdf in enumerate(pin): if not i==self.master: result = gpd.overlay(result, gdf) result.crs = pin[self.master].crs return result
#class DataFrameToMask(PipeSegment): # """ # Given a GeoPandas GeoDataFrame and an Image-class image, # convert the DataFrame to the corresponding Boolean mask # """ # pass # # #class MaskToDataFrame(PipeSegment): # """ # Given a boolean mask, convert it to a GeoPandas GeoDataFrame of polygons. # """ # pass
[docs]class DataFramePixelCoords(PipeSegment): """ Given a GeoPandas GeoDataFrame, converts between georeferenced coordinates and pixel coordinates. Assumes image has affine geotransform. """ def __init__(self, inverse=False, reverse_order=False, *args, **kwargs): super().__init__() self.inverse = inverse self.reverse_order = reverse_order self.args = args self.kwargs = kwargs def transform(self, pin): if not self.reverse_order: gdf = pin[0] img = pin[1] else: gdf = pin[1] img = pin[0] affine = img.metadata['geotransform'] gdf = gdf.copy() newgeoms = gdf.apply(lambda row: convert_poly_coords( row.geometry, affine_obj=affine, inverse=self.inverse, *self.args, **self.kwargs ), axis=1) gdf.geometry = newgeoms return gdf
[docs]class DataFrameToString(PipeSegment): """ Given a GeoPandas GeoDataFrame, convert it into a GeoJSON string. Caveat emptor: This follows the GeoJSON 2016 standard, which does not include any coordinate reference system information. """ def __init__(self, crs=True, **kwargs): super().__init__() self.crs = crs self.kwargs = kwargs def transform(self, pin): geojson = pin.to_json(**(self.kwargs)) if self.crs: geojson = '{"type": "FeatureCollection", "crs": { "type": "name", "properties": { "name": "urn:ogc:def:crs:EPSG::' \ + str(pin.crs.to_epsg()) \ + '" } }, ' \ + geojson[30:] return geojson
[docs]class BoundsToDataFrame(PipeSegment): """ Given a set of tile bounds [left, lower, right, upper], convert it to a GeoPandas GeoDataFrame. Note: User must specify projection, since a simple set of bounds doesn't include that. """ def __init__(self, projection=None): super().__init__() self.projection = projection def transform(self, pin): gdf = gpd.GeoDataFrame() if self.projection is not None: gdf.crs = 'epsg:' + str(self.projection) gdf.loc[0, 'geometry'] = shapely.geometry.box(*pin) return gdf