Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_poac(self):
res = self.indicator.run_poac('ZI001_SDP')
self.assertIsInstance(res[0], SpatialDataFrame)
self.assertIsInstance(res[1], str)
def test_curr(self):
res = self.indicator.run_curr('ZI001_SDV')
self.assertIsInstance(res[0], SpatialDataFrame)
self.assertEqual(res[1][0]['success'], True)
val_dict[tag].append(str(n['tags'].get(tag, 'Null')))
# Populate Geometries & IDs
point = Point({
"x": n['lon'],
"y": n['lat'],
"spatialReference": {"wkid": 4326}
})
geo_dict['geo'].append(point)
val_dict['osm_id'].append(str(n['id']))
except Exception as ex:
print('Node ID {0} Raised Exception: {1}'.format(n['id'], str(ex)))
try:
return SpatialDataFrame(val_dict, geometry=geo_dict['geo'])
except TypeError:
raise Exception('Ensure ArcPy is Included in Python Interpreter')
# Populate Geometries & IDs
coords = [[e['lon'], e['lat']] for e in w.get('geometry')]
if g_type == 'polygon':
poly = Polygon({"rings": [coords], "spatialReference": {"wkid": 4326}})
else:
poly = Polyline({"paths": [coords], "spatialReference": {"wkid": 4326}})
geo_dict['geo'].append(poly)
val_dict['osm_id'].append(str(w['id']))
except Exception as ex:
print('Way ID {0} Raised Exception: {1}'.format(w['id'], str(ex)))
try:
return SpatialDataFrame(val_dict, geometry=geo_dict['geo'])
except TypeError:
raise Exception('Ensure ArcPy is Included in Python Interpreter')
df_before = data_sdf[sq].copy()
geoms_before = df_before.clip(geom.extent)
geoms_before_sdf = SpatialDataFrame(geometry=geoms_before)
q_before = geoms_before_sdf['SHAPE'] == {"paths": []}
geoms_before_sdf = geoms_before_sdf[~q_before].copy()
geoms_before_sdf.reset_index(inplace=True, drop=True)
geometry_type = osm_sdf.geometry_type
sq = osm_sdf[osm_sdf.geometry.notnull()].geometry.disjoint(geom) == False
df_after = osm_sdf[sq].copy()
geoms_after = df_after.clip(geom.extent)
geoms_after_sdf = SpatialDataFrame(geometry=geoms_after)
#geoms_after_sdf = SpatialDataFrame({'Pass': 'Pass'}, geometry=geoms_after, index=[0])
q_after = geoms_after_sdf['SHAPE'] == {"paths": []}
geoms_after_sdf = geoms_after_sdf[~q_after].copy()
geoms_after_sdf.reset_index(inplace=True, drop=True)
# This Need Work
if geometry_type == "Polygon":
if before_val == None:
before_val = geoms_before_sdf.geometry.project_as(4326).get_area('GEODESIC','SQUAREKILOMETERS').sum()
after_val = geoms_after_sdf.geometry.project_as(4326).get_area('GEODESIC','SQUAREKILOMETERS').sum()
if after_val > 0:
score = get_cp_score(ratio=before_val/after_val,
baseVal=before_val,
inputVal=after_val)
else:
if steps < 1: steps = 1
way_chunks = [w_list[i:i + steps] for i in range(0, len(w_list), steps)]
pool = Pool(processes=chunk)
mp_result = pool.map_async(partial(build_ways_dict, n_list, g_type), way_chunks)
pool.close()
pool.join()
if mp_result.ready():
res = mp_result.get()
d_dict = defaultdict(list)
for d in res:
for key, value in d.items():
d_dict[key].extend(value)
return SpatialDataFrame({'ID': d_dict['ids'], 'NAME': d_dict['names']}, geometry=d_dict['geoms'])
# Unpack Geom Extent as OSM Expects
bbox = (geom.extent[1], geom.extent[0], geom.extent[3], geom.extent[2])
# Fetch OSM SpatialDataFrame
osm_sdf = gen_osm_sdf('line', bbox, osm_tag='highway')
data_sdf = df_list[idx]
if len(data_sdf) == 0:
before_val = 0
else:
sq = data_sdf[data_sdf.geometry.notnull()].geometry.disjoint(geom) == False
df_before = data_sdf[sq].copy()
geoms_before = df_before.clip(geom.extent)
geoms_before_sdf = SpatialDataFrame(geometry=geoms_before)
q_before = geoms_before_sdf['SHAPE'] == {"paths": []}
geoms_before_sdf = geoms_before_sdf[~q_before].copy()
geoms_before_sdf.reset_index(inplace=True, drop=True)
geometry_type = osm_sdf.geometry_type
sq = osm_sdf[osm_sdf.geometry.notnull()].geometry.disjoint(geom) == False
df_after = osm_sdf[sq].copy()
geoms_after = df_after.clip(geom.extent)
geoms_after_sdf = SpatialDataFrame(geometry=geoms_after)
#geoms_after_sdf = SpatialDataFrame({'Pass': 'Pass'}, geometry=geoms_after, index=[0])
q_after = geoms_after_sdf['SHAPE'] == {"paths": []}
if steps < 1: steps = 1
lists = [n_list[i:i + steps] for i in range(0, len(n_list), steps)]
pool = Pool(processes=chunk)
mp_result = pool.map_async(build_node_dict, lists)
pool.close()
pool.join()
if mp_result.ready():
res = mp_result.get()
d_dict = defaultdict(list)
for d in res:
for key, value in d.items():
d_dict[key].extend(value)
return SpatialDataFrame({'ID': d_dict['ids'], 'NAME': d_dict['names']}, geometry=d_dict['geoms'])