osm_poi_matchmaker.libs.osm.query_overpass() - Code Metrics - KAMI911/osm_poi_matchmaker - Measure and Improve Code Quality continuously with Scrutinizer

osm_poi_matchmaker.libs.osm.query_overpass() A
last analyzed 2023-01-03 16:24 UTC

↳ Parent: osm_poi_matchmaker.libs.osm

Complexity

Conditions

Size

Total Lines	5
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	3
dl	0
loc	5
rs	10
c	0
b	0
f	0

# -*- coding: utf-8 -*-

try:
    import logging
    import sys
    import sqlalchemy
    import geopandas as gpd

    import datetime
    from OSMPythonTools.overpass import Overpass
    from OSMPythonTools.nominatim import Nominatim
    from OSMPythonTools.overpass import overpassQueryBuilder
except ImportError as err:
    logging.error('Error %s import module: %s', __name__, err)
    logging.exception('Exception occurred')

    sys.exit(128)


def get_area_id(area):

    # Query Nominatom
    nominatim = Nominatim()
    return nominatim.query(area).areaId()


def query_overpass(area_id, query_statement, element_type='node'):

    # Query Overpass based on area
    overpass = Overpass()
    query = overpassQueryBuilder(area=area_id, elementType=element_type, selector=query_statement)
    return overpass.query(query)


def query_osm_postcode_gpd(session, lon, lat):

    if lat is None or lat == '' or lon == '' or lon is None:
        return None
    query = sqlalchemy.text('''
        SELECT name
        FROM planet_osm_polygon, (SELECT ST_SetSRID(ST_MakePoint(:lon, :lat),4326) as geom) point
        WHERE boundary='postal_code' and ST_Contains(way, point.geom) ORDER BY name LIMIT 1;''')
    data = session.execute(query, {'lon': lon, 'lat': lat}).first()
    if data is None:
        return None
    row = dict(zip(data.keys(), data))
    return int(row['name'].split(' ')[0]) if row['name'].split(' ')[0] is not None else None


def query_postcode_osm_external(prefer_osm, session, lon, lat, postcode_ext):

    if prefer_osm is False and postcode_ext is not None:
        return postcode_ext
    query_postcode = query_osm_postcode_gpd(session, lon, lat)
    if prefer_osm is True and query_postcode is not None:

        return query_postcode
    elif prefer_osm is True and query_postcode is None:
        return postcode_ext


def relationer(relation_text):

    if relation_text is None:
        return None
    data = []
    for i in range(0, len(relation_text), 2):
        item = relation_text[i]
        if item[0] == 'n':
            tp = 'node'

        elif item[0] == 'w':
            tp = 'way'

        elif item[0] == 'r':
            tp = 'relation'

        rf = item[1:]

        rl = relation_text[i + 1]

        data.append({'type': tp, 'ref': rf, 'role': rl})

    return data


def timestamp_now():

    return datetime.datetime.now()


def osm_timestamp_now():

    return '{:{dfmt}T{tfmt}Z}'.format(datetime.datetime.now(), dfmt='%Y-%m-%d', tfmt='%H:%M:%S')


def query_osm_city_name_gpd(session, lon, lat):

    if lat is None or lat == '' or lon == '' or lon is None:
        return None
    query = sqlalchemy.text('''
        SELECT name
        FROM planet_osm_polygon, (SELECT ST_SetSRID(ST_MakePoint(:lat,:lon),4326) as geom) point
        WHERE admin_level='8' and ST_Contains(way, point.geom) ORDER BY name LIMIT 1;''')
    data = session.execute(query, {'lon': lon, 'lat': lat}).first()
    if data is None:

        return None
    else:
        return data[0]


def query_osm_city_name(session, name):

    query = sqlalchemy.text('''
        SELECT name
        FROM planet_osm_polygon WHERE admin_level='8' and name=:name ORDER BY name LIMIT 1;''')
    data = session.execute(query, {'name': name}).first()
    if data is None:

        return None
    else:
        return data[0]


1			# -- coding: utf-8 --
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing module docstring Loading history...
2			try:
3			import logging
4			import sys
5			import sqlalchemy
6			import geopandas as gpd
			0 ignored issues – show Unused Code introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Unused geopandas imported as gpd Loading history...
7			import datetime
8			from OSMPythonTools.overpass import Overpass
9			from OSMPythonTools.nominatim import Nominatim
10			from OSMPythonTools.overpass import overpassQueryBuilder
11			except ImportError as err:
12			logging.error('Error %s import module: %s', __name__, err)
13			logging.exception('Exception occurred')
14
15			sys.exit(128)
16
17
18			def get_area_id(area):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
19			# Query Nominatom
20			nominatim = Nominatim()
21			return nominatim.query(area).areaId()
22
23
24			def query_overpass(area_id, query_statement, element_type='node'):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
25			# Query Overpass based on area
26			overpass = Overpass()
27			query = overpassQueryBuilder(area=area_id, elementType=element_type, selector=query_statement)
28			return overpass.query(query)
29
30
31			def query_osm_postcode_gpd(session, lon, lat):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
32			if lat is None or lat == '' or lon == '' or lon is None:
33			return None
34			query = sqlalchemy.text('''
35			SELECT name
36			FROM planet_osm_polygon, (SELECT ST_SetSRID(ST_MakePoint(:lon, :lat),4326) as geom) point
37			WHERE boundary='postal_code' and ST_Contains(way, point.geom) ORDER BY name LIMIT 1;''')
38			data = session.execute(query, {'lon': lon, 'lat': lat}).first()
39			if data is None:
40			return None
41			row = dict(zip(data.keys(), data))
42			return int(row['name'].split(' ')[0]) if row['name'].split(' ')[0] is not None else None
43
44
45			def query_postcode_osm_external(prefer_osm, session, lon, lat, postcode_ext):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history... Unused Code introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Either all return statements in a function should return an expression, or none of them should. Loading history...
46			if prefer_osm is False and postcode_ext is not None:
47			return postcode_ext
48			query_postcode = query_osm_postcode_gpd(session, lon, lat)
49			if prefer_osm is True and query_postcode is not None:
			0 ignored issues – show unused-code introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Unnecessary "elif" after "return" Loading history...
50			return query_postcode
51			elif prefer_osm is True and query_postcode is None:
52			return postcode_ext
53
54
55			def relationer(relation_text):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
56			if relation_text is None:
57			return None
58			data = []
59			for i in range(0, len(relation_text), 2):
60			item = relation_text[i]
61			if item[0] == 'n':
62			tp = 'node'
			0 ignored issues – show Coding Style Naming introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Variable name "tp" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
63			elif item[0] == 'w':
64			tp = 'way'
			0 ignored issues – show Coding Style Naming introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Variable name "tp" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
65			elif item[0] == 'r':
66			tp = 'relation'
			0 ignored issues – show Coding Style Naming introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Variable name "tp" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
67			rf = item[1:]
			0 ignored issues – show Coding Style Naming introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Variable name "rf" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
68			rl = relation_text[i + 1]
			0 ignored issues – show Coding Style Naming introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Variable name "rl" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,\|_[^\\WA-Z]*\|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern) This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
69			data.append({'type': tp, 'ref': rf, 'role': rl})
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report The variable `tp` does not seem to be defined for all execution paths. Loading history...
70			return data
71
72
73			def timestamp_now():
			0 ignored issues – show introduced 2020-10-23 06:34 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
74			return datetime.datetime.now()
75
76
77			def osm_timestamp_now():
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
78			return '{:{dfmt}T{tfmt}Z}'.format(datetime.datetime.now(), dfmt='%Y-%m-%d', tfmt='%H:%M:%S')
79
80
81			def query_osm_city_name_gpd(session, lon, lat):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
82			if lat is None or lat == '' or lon == '' or lon is None:
83			return None
84			query = sqlalchemy.text('''
85			SELECT name
86			FROM planet_osm_polygon, (SELECT ST_SetSRID(ST_MakePoint(:lat,:lon),4326) as geom) point
87			WHERE admin_level='8' and ST_Contains(way, point.geom) ORDER BY name LIMIT 1;''')
88			data = session.execute(query, {'lon': lon, 'lat': lat}).first()
89			if data is None:
			0 ignored issues – show unused-code introduced 2020-10-19 20:05 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
90			return None
91			else:
92			return data[0]
93
94
95			def query_osm_city_name(session, name):
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Missing function or method docstring Loading history...
96			query = sqlalchemy.text('''
97			SELECT name
98			FROM planet_osm_polygon WHERE admin_level='8' and name=:name ORDER BY name LIMIT 1;''')
99			data = session.execute(query, {'name': name}).first()
100			if data is None:
			0 ignored issues – show unused-code introduced 2020-10-19 20:05 UTC by Report Bug Copy Issue Report Unnecessary "else" after "return" Loading history...
101			return None
102			else:
103			return data[0]
104

KAMI911 / osm_poi_matchmaker

osm_poi_matchmaker.libs.osm.query_overpass() A last analyzed 2023-01-03 16:24 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

osm_poi_matchmaker.libs.osm.query_overpass() A
last analyzed 2023-01-03 16:24 UTC