-
Notifications
You must be signed in to change notification settings - Fork 0
/
load_gml.py
112 lines (95 loc) · 2.73 KB
/
load_gml.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
#!/usr/bin/python
# -*- coding: utf-8 -*-
""" load_gml.py : load_gml.py """
'''
Temporarily created python script to load GML
data to the application. Will replace later with
a user interface.
@TODO: add WebUI for data loading.
@TODO: add support for more formats.
'''
from lxml import objectify
from lxml.objectify import ObjectifiedElement
from configparser import ConfigParser
import sys
import re
import json
import os
__author__ = "Abhay Arora (@BzFTMxc)"
if len(sys.argv) < 2:
print '\nMissing required parameters!\nUsage: load_gml.py <path-to-gml>\n'
exit(1)
gml = None
try:
gml = objectify.parse(sys.argv[1])
except:
print 'Could not read GML!\n Error while reading ' + sys.argv[1] + '.'
exit(1)
#
# Read configs
#
config = ConfigParser()
try:
config.read(os.path.dirname(os.path.abspath(__file__)) +\
'/config/geoweb.ini')
except:
print('Could not read config!\n')
exit()
#
# Iterates xml and generates dict.
# XML hierarchy not persisted in dict. Need to fix.
#
# @TODO: rewrite function to iterate single depth level
#
def to_dict(node):
iterator = node.iter()
_self = iterator.next()
data = dict()
while True:
try:
_next = iterator.next()
if type(_next) is ObjectifiedElement:
continue
# data[re.sub(r'{.*}', '', _next.tag)] = to_dict(_next)
# @TODO: Enable above written recursion
# after solving iteration depth issue
else:
data[re.sub(r'{.*}', '', _next.tag)] = _next.text
except:
break
return data
# ------------------------
# Initialize SAM directory
# ------------------------
data_dir = config.get('storage', 'data_dir')
sam_name = str(gml.getroot().iter().next().header.requestID.text)
if not os.path.exists(os.path.join(data_dir, sam_name)):
os.makedirs(os.path.join(data_dir, sam_name))
sam_dir = os.path.join(data_dir, sam_name)
#
# Dumps GML data into JSON file
# Uses to_dict() function
#
def dump_json(node, tag, file_path=None):
elems = gml.iter(tag)
elems_data = dict(
element=re.sub(r'{.*}', '', tag),
data=[]
)
while True:
try:
elem = elems.next()
elems_data['data'].append(to_dict(elem))
except:
break
if file_path is None:
file_path = os.path.join(sam_dir, re.sub(r'{.*}', '', tag) + '.json')
elems_file = open(file_path, 'w')
elems_file.write(json.dumps(elems_data))
elems_file.close()
#
# Dump required data
#
dump_json(gml, r'{http://www.telstra.com.au/nbn}coaxialCable')
dump_json(gml, r'{http://www.telstra.com.au/nbn}structureRoute')
dump_json(gml, r'{http://www.telstra.com.au/nbn}structurePoint')