def parse_OGR_nodes_paths()

in libs/apls/graphTools.py [0:0]

• 62 lines of code
• 22 McCabe index (conditional complexity)

def parse_OGR_nodes_paths(vectorFileName, osmidx=0, osmNodeidx=0,
                          nodeListGpd=gpd.GeoDataFrame(),
                          valid_road_types=set([]),
                          roadTypeField='type',
                          verbose=True,
                          super_verbose=False):
    """
    Construct dicts of nodes and paths with key=osmid and value=dict of
    attributes.

    Notes
    -----
    valid_road_types is a set of road types to be allowed

    Parameters
    ----------
    vectorFileName : str
        Absolute path to vector file supported by OGR that has
        line segments JSON response from from the Overpass API.

    Returns
    -------
    nodes, paths : tuple
    """

    # ensure valid vectorFileName
    try:
        source = fiona.open(vectorFileName, 'r')
        doit = True
    except:
        doit = False
        return {}, {}

    #dataSource = ogr.Open(vectorFileName, 0)
    # with fiona.open(vectorFileName, 'r') as source:
    if doit:
        #layer = dataSource.GetLayer()
        nodes = {}
        paths = {}
        for i, feature in enumerate(source):

            geom = feature['geometry']
            properties = feature['properties']
            # todo create more adjustable filter
            if roadTypeField in properties:
                road_type = properties['type']
            elif 'highway' in properties:
                road_type = properties['highway']
            elif 'road_type' in properties:
                road_type = properties['road_type']
            else:
                road_type = 'None'

            if ((i % 100) == 0) and verbose:
                print("\n", i, "/", len(source))
                print("   geom:", geom)
                print("   properties:", properties)
                print("   road_type:", road_type)

            ##################
            # check if road type allowable, continue if not
            # first check if road
            if (len(valid_road_types) > 0) and \
                    (geom['type'] == 'LineString' or geom['type'] == 'MultiLineString'):
                if road_type not in valid_road_types:
                    if verbose:
                        print("Invalid road type, skipping...")
                    continue
            ###################

            # skip empty linestrings
            if 'LINESTRING EMPTY' in list(properties.values()):
                continue

            osmidx = int(osmidx + 1)
            # print ("geom:", geom)

            if geom['type'] == 'LineString':
                # print osmNodeidx
                lineString = shapely.geometry.shape(geom)
                if super_verbose:
                    print("lineString.wkt:", lineString.wkt)
                # if len(geom['coordinates']) == 0:
                #    continue

                path, nodeList, osmNodeidx, nodeListGpd = \
                    processLineStringFeature(lineString, osmidx, osmNodeidx,
                                             nodeListGpd, properties=properties)
                # print(nodeListGpd.head())
                osmNodeidx = osmNodeidx+1
                osmidx = osmidx+1
                #print ("nodeList:", nodeList)
                nodes.update(nodeList)
                paths[osmidx] = path
                # print(geom.GetGeometryName())

            elif geom['type'] == 'MultiLineString':
                for linestring in shapely.geometry.shape(geom):

                    path, nodeList, osmNodeidx, nodeListGpd = \
                        processLineStringFeature(linestring, osmidx, osmNodeidx,
                                                 nodeListGpd, properties=properties)
                    osmNodeidx = osmNodeidx + 1
                    osmidx = osmidx+1
                    # print(geom.GetGeometryName())
                    nodes.update(nodeList)
                    paths[osmidx] = path

        source.close()

    return nodes, paths