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

ascii_numcoder() A
last analyzed 2023-01-03 16:24 UTC

↳ Parent: osm_poi_matchmaker.libs.file_output

Complexity

Conditions

Size

Total Lines	8
Code Lines	7

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	7
nop	1
dl	0
loc	8
rs	10
c	0
b	0
f	0

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


try:
    import logging
    import sys
    import math
    import os
    import datetime
    from urllib.parse import quote
    from osm_poi_matchmaker.dao.data_structure import OSM_object_type
    from osm_poi_matchmaker.utils import config
    from osm_poi_matchmaker.libs.address import clean_url
    from osm_poi_matchmaker.libs.osm import relationer, timestamp_now

    from osm_poi_matchmaker.libs.compare_strings import compare_strings
    from sqlalchemy.orm import scoped_session, sessionmaker
    from osm_poi_matchmaker.dao.poi_base import POIBase

    from lxml import etree
    import lxml
except ImportError as err:
    logging.error('Error %s import module: %s', __name__, err)
    logging.exception('Exception occurred')

    sys.exit(128)

POI_TAGS = {'poi_name': 'name', 'poi_city': 'addr:city', 'poi_postcode': 'addr:postcode',
            'poi_addr_street': 'addr:street', 'poi_addr_housenumber': 'addr:housenumber',
            'poi_conscriptionnumber': 'addr:conscriptionnumber', 'poi_branch': 'branch', 'poi_email': 'email'}


POI_YESNO_TAGS = {'poi_fuel_adblue': 'fuel:adblue', 'poi_fuel_octane_100': 'fuel:octane_100',
                  'poi_fuel_octane_98': 'fuel:octane_98', 'poi_fuel_octane_95': 'fuel:octane_95',
                  'poi_fuel_diesel_gtl': 'fuel:GTL_diesel', 'poi_fuel_diesel': 'fuel:diesel',
                  'poi_fuel_lpg': 'fuel:lpg', 'poi_fuel_e85': 'fuel:e85', 'poi_rent_lpg_bottles': 'rent:lpg_bottles',

                  'poi_compressed_air': 'compressed_air', 'poi_restaurant': 'restaurant', 'poi_food': 'food',

                  'poi_truck': 'truck', 'poi_authentication_app': 'authentication:app',
                  'poi_authentication_membership_card': 'authentication:membership_card', 'poi_fee': 'fee',

                  'poi_parking_fee': 'parking_fee', 'poi_motorcar': 'motorcar'}

POI_EV_TAGS = {'poi_capacity': 'capacity',
               'poi_socket_chademo': 'socket:chademo', 'poi_socket_chademo_output': 'socket:chademo:output',

               'poi_socket_type2_combo': 'socket:type2_combo',
               'poi_socket_type2_combo_output': 'socket:type2_combo:output',
               'poi_socket_type2_cable': 'socket:type2_cable',
               'poi_socket_type2_cable_output': 'socket:type2_cable:output',
               'poi_socket_type2': 'socket:type2', 'poi_socket_type2_output': 'socket:type2:output',
               'poi_manufacturer': 'manufacturer', 'poi_model': 'model'}

TESTCASE_GEN_KEYS = ('original', 'poi_postcode', 'poi_city', 'poi_addr_street', 'poi_addr_housenumber', 'poi_conscriptionnumber')


TIMESTAMP_FORMAT = '{:{dfmt}T{tfmt}Z}'
DATE_FOTMAT = '%Y-%m-%d'
TIME_FORMAT = '%H:%M:%S'


def ascii_numcoder(text):

    output = ''
    for i in text:
        if i in range(0, 10, 1):
            output += i
        else:
            output += str(ord(i))
    return output


def save_csv_file(path: str, file: str, data, message: str):
    """Save Pandas dataframe to a CSV file

    Args:
        path (str): Path of newly created CVS file
        file (str): Filename of newly created CVS file
        data (pd.DataFrame): Pandas dataframe to write
        message (str): Addtion information to display
    """
    try:
        # Save file to CSV file
        logging.info('Saving {%s to file: %s', message, file)
        res = data.to_csv(os.path.join(path, file))

        logging.info('The %s was sucessfully saved', file)
    except Exception as e:

        logging.error(e)
        logging.exception('Exception occurred')


def add_osm_node(osm_id: int, node_data: dict, prefix: str = 'poi') -> dict:
    """Generate OpenStreetMap node header information as string

    Args:
        osm_id (int): OpenStreetMap ID
        node_data (dict): [description]
        prefix (str): Prefix for field names in database

    Returns:
        str: [description]
    """
    logging.info(node_data.to_string())
    if node_data.get('osm_timestamp') is None:
        osm_timestamp = datetime.datetime.now()
    else:
        osm_timestamp = node_data.get('osm_timestamp')
    osm_version = '99999' if node_data.get('osm_version') is None else node_data.get('osm_version')
    osm_data = {'action': 'modify', 'id': str(osm_id),
                'lat': '{}'.format(node_data.get('{}_lat'.format(prefix))),
                'lon': '{}'.format(node_data.get('{}_lon'.format(prefix))),
                'user': '{}'.format('osm_poi_matchmaker'), 'uid': '{}'.format('8635934'), 'version': '{}'.format(osm_version),

                'timestamp': TIMESTAMP_FORMAT.format(osm_timestamp, dfmt=DATE_FOTMAT, tfmt=TIME_FORMAT)}

    logging.info(osm_data)
    return osm_data


def list_osm_node(osm_id: int, node_data: dict, prefix='poi') -> dict:
    """Generate OpenStreetMap node header information as string

    Args:
        osm_id (int): OpenStreetMap ID
        node_data (dict): [description]
        prefix (str): Prefix for field names in database

    Returns:
        str: [description]
    """
    logging.debug(node_data)
    osm_user = 'osm_poi_matchmaker' if node_data.get('osm_user') is None else node_data.get('osm_user')

    osm_user_id = '8635934' if node_data.get('osm_user_id') is None else node_data.get('osm_user_id')

    if node_data.get('osm_timestamp') is None:
        osm_timestamp = datetime.datetime.now()
    else:
        osm_timestamp = node_data.get('osm_timestamp')
    osm_version = '99999' if node_data.get('osm_version') is None else node_data.get('osm_version')
    osm_data = {'id': str(osm_id),
                'lat': '{}'.format(node_data.get('{}_lat'.format(prefix))),
                'lon': '{}'.format(node_data.get('{}_lon'.format(prefix))),
                'user': '{}'.format(osm_user), 'uid': '{}'.format(osm_user_id), 'version': '{}'.format(osm_version),

                'timestamp': TIMESTAMP_FORMAT.format(osm_timestamp, dfmt=DATE_FOTMAT, tfmt=TIME_FORMAT)}

    logging.debug(osm_data)
    return osm_data


def add_osm_way(osm_id: int, node_data: dict) -> dict:
    """Generate OpenStreetMap way header information as dictionary

    Args:
        osm_id (int): [description]
        node_data (dict): [description]

    Returns:
        str: [description]
    """
    if node_data.get('osm_timestamp') is None:
        osm_timestamp = datetime.datetime.now()
    else:
        osm_timestamp = node_data.get('osm_timestamp')
    osm_version = '99999' if node_data.get('osm_version') is None else node_data.get('osm_version')
    osm_data = {'action': 'modify', 'id': str(osm_id),
                'user': '{}'.format('osm_poi_matchmaker'), 'uid': '{}'.format('8635934'),
                'version': '{}'.format(osm_version),
                'timestamp': TIMESTAMP_FORMAT.format(osm_timestamp, dfmt=DATE_FOTMAT, tfmt=TIME_FORMAT)}

    return osm_data


def add_osm_link_comment(osm_id: int, osm_type) -> str:
    """Create OpenStreetMap osm.org link from OSM object

    Args:
        osm_id (int): [description]
        osm_type ([type]): [description]

    Returns:
        str: [description]
    """
    osm_comment = ' OSM link: https://osm.org/{}/{} '.format(osm_type.name, str(osm_id))
    return osm_comment


def generate_osm_xml(df, session=None):

    """Crete OpenStreetMap (OSM XML) file from passed Panda Dataframe

    Args:
        df ([type]): [description]
        session ([type], optional): [description]. Defaults to None.

    Returns:
        [type]: [description]
    """
    db = POIBase('{}://{}:{}@{}:{}/{}'.format(config.get_database_type(), config.get_database_writer_username(),

                                              config.get_database_writer_password(),
                                              config.get_database_writer_host(),
                                              config.get_database_writer_port(),
                                              config.get_database_poi_database()))
    pgsql_pool = db.pool
    session_factory = sessionmaker(pgsql_pool)
    Session = scoped_session(session_factory)

    session = Session()
    osm_xml_data = etree.Element('osm', version='0.6', generator='JOSM')

    default_osm_id = -1
    current_osm_id = default_osm_id
    added_nodes = []
    try:

        for index, row in df.iterrows():

            tags = {}
            osm_live_tags = {}
            main_data = {}
            current_osm_id = default_osm_id if row.get('osm_id') is None else row.get('osm_id')
            osm_version = '99999' if row.get('osm_version') is None else row.get('osm_version')

            if row.get('osm_node') is None or row.get('osm_node') == OSM_object_type.node:
                try:
                    josm_object = 'n{}'.format(current_osm_id)
                    main_data = etree.SubElement(osm_xml_data, 'node', add_osm_node(current_osm_id, row))

                except Exception as e:

                    logging.exception('Exception occurred')
            elif row.get('osm_node') is not None and row.get('osm_node') == OSM_object_type.way:
                try:
                    main_data = etree.SubElement(osm_xml_data, 'way', add_osm_way(current_osm_id, row))

                    josm_object = 'w{}'.format(current_osm_id)
                # Add way nodes without any modification)
                    node_data = []
                    for n in row.get('osm_nodes'):

                        data = etree.SubElement(main_data, 'nd', ref=str(n))

                    if session is not None:
                        # Go through the list except the last value (which is same as the first)
                        for n in row.get('osm_nodes'):

                            # Add nodes only when it is not already added.
                            if n not in added_nodes:
                                added_nodes.append(n)
                                way_node = db.query_from_cache(n, OSM_object_type.node)
                                if way_node is not None:
                                    node_data = etree.SubElement(osm_xml_data, 'node',

                                                                 list_osm_node(n, way_node, 'osm'))
                                    if node_data.get('osm_live_tags') is not None and \
                                       node_data.get('osm_live_tags') != '':
                                        node_osm_live_tags = node_data.get('osm_live_tags')
                                        for k, v in sorted(node_osm_live_tags).items():

                                            xml_node_tags = etree.SubElement(node_data, 'tag', k=k, v='{}'.format(v))

                except TypeError as e:

                    logging.warning('Missing nodes on this way: %s.', row.get('osm_id'))
                    logging.exception('Exception occurred')
            elif row.get('osm_node') is not None and row.get('osm_node') == OSM_object_type.relation:

                try:
                    main_data = etree.SubElement(osm_xml_data, 'relation', add_osm_way(current_osm_id, row))

                    josm_object = 'r{}'.format(current_osm_id)
                    relations = relationer(row.get('osm_nodes'))
                    for i in relations:
                        data = etree.SubElement(main_data, 'member', type=i.get('type'), ref=i.get('ref'),

                                                role=i.get('role'))
                except TypeError as e:

                    logging.warning('Missing nodes on this relation: %s.', row['osm_id'])
                    logging.exception('Exception occurred')
            # Add already existing node, way, relation OpenStreetMap reference as comment
            try:
                if current_osm_id > 0:
                    osm_xml_data.append(etree.Comment(add_osm_link_comment(current_osm_id, row.get('osm_node'))))

                # Add original POI coordinates as comment
                comment = etree.Comment(' Original coordinates: {} '.format(row.get('poi_geom')))

                osm_xml_data.append(comment)
                if 'poi_distance' in row:
                    if row.get('poi_distance') is not None:
                        comment = etree.Comment(' OSM <-> POI distance: {} m'.format(row.get('poi_distance')))

                    else:
                        comment = etree.Comment(' OSM <-> POI distance: Non exist')

                    osm_xml_data.append(comment)
                if 'poi_good' in row and 'poi_bad' in row:
                    comment = etree.Comment(' Checker good: {}; bad {}'.format(row.get('poi_good'), row.get('poi_bad')))

                    osm_xml_data.append(comment)
                # Using already definied OSM tags if exists
                if row.get('osm_live_tags') is not None:
                    tags.update(row.get('osm_live_tags').copy())
                    logging.critical(row.get('osm_live_tags'))
                    osm_live_tags.update(row.get('osm_live_tags').copy())
                # Adding POI common tags
                if row.get('poi_tags') is not None:
                    tags.update(row.get('poi_tags'))
                # Save live name tags if preserve name is enabled
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                if row.get('preserve_original_name') is True:
                    preserved_name = tags.get('name')
            except KeyError as e:

                logging.debug('No name tag is specified to save in original OpenStreetMap data.')
            try:
                # Overwriting with data from data providers
                for k, v in POI_TAGS.items():

                    if row.get(k) is not None:
                        tags[v] = row.get(k)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                if config.get_geo_alternative_opening_hours():
                    alternative_oh_tag = config.get_geo_alternative_opening_hours_tag()
                    # Alternative opening_hours handling for COVID-19 code path
                    if tags.get('opening_hours') is not None and tags.get('opening_hours') != '':
                        if row.get('poi_opening_hours') is not None and row.get('poi_opening_hours') != '':

                            if tags.get('opening_hours') == row.get('poi_opening_hours'):
                                tags[alternative_oh_tag] = 'same'
                            else:
                                tags[alternative_oh_tag] = row.get('poi_opening_hours')
                    else:
                        if row.get('poi_opening_hours') is not None and row.get('poi_opening_hours') != '':

                            tags['opening_hours'] = row.get('poi_opening_hours')
                            tags[alternative_oh_tag] = 'same'
                else:
                    # Alternative opening_hours handling for NON COVID-19 code path: just simply add opening_hours to tags

                    if row.get('poi_opening_hours') is not None and row.get('poi_opening_hours') != '':

                        tags['opening_hours'] = row.get('poi_opening_hours')
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # If we got POI phone tag use it as OSM contact:phone tag
                if row.get('poi_phone') is not None and row.get('poi_phone') != '':
                    tags['contact:phone'] = row.get('poi_phone')
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # If we got POI website tag use it as OSM contact:website tag
                if row.get('poi_url_base') is not None and row.get('poi_website') is not None:
                    if row['poi_url_base'] in row.get('poi_website'):
                        # The POI website contains the base URL use the POI website field only
                        tags['contact:website'] = clean_url('{}'.format((row.get('poi_website'))))
                    else:
                        # The POI website does not contain the base URL use the merged base URL and POI website field

                        tags['contact:website'] = clean_url('{}/{}'.format(row.get('poi_url_base'), row.get('poi_website')))

                # If only the base URL is available
                elif row.get('poi_url_base') is not None:
                    tags['contact:website'] = row.get('poi_url_base')
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # Short URL for source
                if row['poi_url_base'] is not None:
                    source_url = 'source:{}:date'.format(row.get('poi_url_base').split('/')[2])
                else:
                    source_url = 'source:website:date'
                tags[source_url] = '{:{dfmt}}'.format(datetime.datetime.now(), dfmt=DATE_FOTMAT)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # Write back the saved name tag
                if 'preserved_name' in locals():
                    tags['name'] = preserved_name
                # Rewrite old contact tags to contact:* tag form
                tags_rewrite = ['website', 'phone', 'email', 'facebook', 'instagram', 'youtube', 'pinterest', 'fax']

                for tr in tags_rewrite:

                    if tr in tags:
                        # Never overwrite already existing contact:* tags
                        if 'contact:' + tr in tags:
                            # We already have this contact:* tag so remove the simple contact tag
                            tags.pop(tr, None)
                        else:
                            # Rewrite simple contact tag to contact:* tag
                            tags['contact:' + tr] = tags.pop(tr, None)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                if row.get('poi_description') is not None and row.get('poi_description') != '':
                    tags['description'] = row.get('poi_description')
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # Write tags with yes/no value
                for k, v in POI_YESNO_TAGS.items():

                    if row.get(k) is not None and row.get(k) != '':
                        tags[v] = 'yes' if row.get(k) is True else 'no'
                for k, v in POI_EV_TAGS.items():

                    if row.get(k) is not None and row.get(k) != '':
                        if isinstance(row.get(k), float):
                            if not math.isnan(row.get(k)):
                                tags[v] = int(row.get(k))
                        else:
                            tags[v] = row.get(k)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # This is a new POI - will add fix me tag to the new items.
                if row.get('poi_new') is not None and row.get('poi_new') is True:
                    tags['fixme'] = 'verify import'
                # Remove unwanted addr:country from file output as we discussed in Issue #33
                tags.pop('addr:country', None)
                # tags['import'] = 'osm_poi_matchmaker'
                # Rendering tags to the XML file and JOSM magic link
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                josm_link = ''
                comment = '\nKey\t\t\t\tStatus\t\tNew value\t\tOSM value\n'
                for k, v in sorted(tags.items()):

                    xml_tags = etree.SubElement(main_data, 'tag', k=k, v='{}'.format(v))

                    josm_link = '{}|{}={}'.format(josm_link, k, v)
                    # Add original POI tags as comment
                    try:
                        if isinstance(v, str):
                            v = v.replace('-', '\-').replace('\n', '')

                        w = osm_live_tags[k]

                    except KeyError:
                        comment += "{:32} N\t\t'{}'\n".format(k, v)
                    else:
                        if isinstance(w, str):
                            w = w.replace('-', '\-').replace('\n', '')

                        comment += "{:32} {}\t\t'{}'\t\t\t'{}'\n".format(k, compare_strings(v, w), v, w)

                comment = etree.Comment(comment)

            except Exception as e:

                logging.exception('Exception occurred')
            try:
                osm_xml_data.append(comment)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                # URL encode link and '--' in comment
                josm_link = quote(josm_link)
                josm_link = josm_link.replace('--', '%2D%2D')
                comment = etree.Comment(' JOSM magic link: {}?new_layer=false&objects={}&addtags={} '.format

                                        ('http://localhost:8111/load_object', josm_object, josm_link))

                osm_xml_data.append(comment)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                test_case = {k: row.get(k, None) for k in TESTCASE_GEN_KEYS}
                comment = etree.Comment("ˇ'original': '{t[original]}', 'postcode': '{t[poi_postcode]}', 'city': '{t[poi_city]}', 'street': '{t[poi_addr_street]}', 'housenumber': '{t[poi_addr_housenumber]}', 'conscriptionnumber': '{t[poi_conscriptionnumber]}'°".format(t=test_case))

                osm_xml_data.append(comment)
            except Exception as e:

                logging.exception('Exception occurred')
            try:
                osm_xml_data.append(xml_node_tags)

            except UnboundLocalError as e:

                logging.debug('Unbound local error extra node tags')
            try:
                osm_xml_data.append(main_data)
                # Next deafult OSM id is one more less for non existing objects
                default_osm_id -= 1
            except Exception as e:

                logging.exception('Exception occurred')
    except ValueError as e:

        logging.error(e)
        logging.exception('Exception occurred')

    except Exception as e:

        logging.error(e)
        logging.exception('Exception occurred')

    return lxml.etree.tostring(osm_xml_data, pretty_print=True, xml_declaration=True, encoding="UTF-8")



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
3			try:
4			import logging
5			import sys
6			import math
7			import os
8			import datetime
9			from urllib.parse import quote
10			from osm_poi_matchmaker.dao.data_structure import OSM_object_type
11			from osm_poi_matchmaker.utils import config
12			from osm_poi_matchmaker.libs.address import clean_url
13			from osm_poi_matchmaker.libs.osm import relationer, timestamp_now
			0 ignored issues – show Unused Code introduced 2020-10-23 06:34 UTC by Report Bug Copy Issue Report Unused timestamp_now imported from osm_poi_matchmaker.libs.osm Loading history...
14			from osm_poi_matchmaker.libs.compare_strings import compare_strings
15			from sqlalchemy.orm import scoped_session, sessionmaker
16			from osm_poi_matchmaker.dao.poi_base import POIBase
			0 ignored issues – show introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report Imports from package osm_poi_matchmaker are not grouped Loading history...
17			from lxml import etree
18			import lxml
19			except ImportError as err:
20			logging.error('Error %s import module: %s', __name__, err)
21			logging.exception('Exception occurred')
22
23			sys.exit(128)
24
25			POI_TAGS = {'poi_name': 'name', 'poi_city': 'addr:city', 'poi_postcode': 'addr:postcode',
26			'poi_addr_street': 'addr:street', 'poi_addr_housenumber': 'addr:housenumber',
27			'poi_conscriptionnumber': 'addr:conscriptionnumber', 'poi_branch': 'branch', 'poi_email': 'email'}
			0 ignored issues – show Coding Style introduced 2020-10-11 06:10 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (110/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
28
29			POI_YESNO_TAGS = {'poi_fuel_adblue': 'fuel:adblue', 'poi_fuel_octane_100': 'fuel:octane_100',
30			'poi_fuel_octane_98': 'fuel:octane_98', 'poi_fuel_octane_95': 'fuel:octane_95',
31			'poi_fuel_diesel_gtl': 'fuel:GTL_diesel', 'poi_fuel_diesel': 'fuel:diesel',
32			'poi_fuel_lpg': 'fuel:lpg', 'poi_fuel_e85': 'fuel:e85', 'poi_rent_lpg_bottles': 'rent:lpg_bottles',
			0 ignored issues – show Coding Style introduced 2020-10-19 19:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (117/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
33			'poi_compressed_air': 'compressed_air', 'poi_restaurant': 'restaurant', 'poi_food': 'food',
			0 ignored issues – show Coding Style introduced 2020-10-19 19:19 UTC by Report Bug Copy Issue Report This line is too long as per the coding-style (109/100). This check looks for lines that are too long. You can specify the maximum line length. Loading history...
34			'poi_truck': 'truck', 'poi_authentication_app': 'authentication:app',

KAMI911 / osm_poi_matchmaker

ascii_numcoder() A last analyzed 2023-01-03 16:24 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

ascii_numcoder() A
last analyzed 2023-01-03 16:24 UTC