daemon: Initial support for importing a scenario using the new NRL
Network Modeling Framework (NMF) XML representation.
This commit is contained in:
tgoff0
parent
08c9fd8bf5
commit
1112da1417
3 changed files with 1093 additions and 12 deletions
|
|
@ -3,10 +3,13 @@
|
|||
# See the LICENSE file included in this distribution.
|
||||
|
||||
from xml.dom.minidom import parse
|
||||
from xmlutils import getoneelement
|
||||
from xmlutils import getFirstChildByTagName
|
||||
from xmlparser0 import CoreDocumentParser0
|
||||
from xmlparser1 import CoreDocumentParser1
|
||||
|
||||
class CoreVersionParser(object):
|
||||
DEFAULT_SCENARIO_VERSION = '1.0'
|
||||
|
||||
'''\
|
||||
Helper class to check the version of Network Plan document. This
|
||||
simply looks for a "Scenario" element; when present, this
|
||||
|
|
@ -19,9 +22,14 @@ class CoreVersionParser(object):
|
|||
self.dom = options['dom']
|
||||
else:
|
||||
self.dom = parse(filename)
|
||||
self.scenario = getoneelement(self.dom, 'Scenario')
|
||||
if self.scenario is not None:
|
||||
self.version = 0.0
|
||||
scenario = getFirstChildByTagName(self.dom, 'scenario')
|
||||
if scenario:
|
||||
version = scenario.getAttribute('version')
|
||||
if not version:
|
||||
version = self.DEFAULT_SCENARIO_VERSION
|
||||
self.version = version
|
||||
elif getFirstChildByTagName(self.dom, 'Scenario'):
|
||||
self.version = '0.0'
|
||||
else:
|
||||
self.version = 'unknown'
|
||||
|
||||
|
|
@ -29,8 +37,10 @@ def core_document_parser(session, filename, options):
|
|||
vp = CoreVersionParser(filename, options)
|
||||
if 'dom' not in options:
|
||||
options['dom'] = vp.dom
|
||||
if vp.version == 0.0:
|
||||
if vp.version == '0.0':
|
||||
doc = CoreDocumentParser0(session, filename, options)
|
||||
elif vp.version == '1.0':
|
||||
doc = CoreDocumentParser1(session, filename, options)
|
||||
else:
|
||||
raise ValueError, 'unsupported document version: %s' % vp.version
|
||||
return doc
|
||||
|
|
|
|||
942
daemon/core/misc/xmlparser1.py
Normal file
942
daemon/core/misc/xmlparser1.py
Normal file
|
|
@ -0,0 +1,942 @@
|
|||
#
|
||||
# CORE
|
||||
# Copyright (c) 2015 the Boeing Company.
|
||||
# See the LICENSE file included in this distribution.
|
||||
#
|
||||
|
||||
import sys
|
||||
import random
|
||||
from core.netns import nodes
|
||||
from core import constants
|
||||
from core.misc.ipaddr import MacAddr
|
||||
from xml.dom.minidom import parse
|
||||
from xmlutils import *
|
||||
|
||||
class CoreDocumentParser1(object):
|
||||
|
||||
layer2_device_types = 'hub', 'switch'
|
||||
layer3_device_types = 'host', 'router'
|
||||
device_types = layer2_device_types + layer3_device_types
|
||||
|
||||
# TODO: support CORE interface classes:
|
||||
# RJ45Node
|
||||
# TunnelNode
|
||||
|
||||
def __init__(self, session, filename, options):
|
||||
self.session = session
|
||||
self.verbose = self.session.getcfgitembool('verbose', False)
|
||||
self.filename = filename
|
||||
if 'dom' in options:
|
||||
# this prevents parsing twice when detecting file versions
|
||||
self.dom = options['dom']
|
||||
else:
|
||||
self.dom = parse(filename)
|
||||
self.start = options['start']
|
||||
self.nodecls = options['nodecls']
|
||||
self.scenario = self.get_scenario(self.dom)
|
||||
self.location_refgeo_set = False
|
||||
self.location_refxyz_set = False
|
||||
# saved link parameters saved when parsing networks and applied later
|
||||
self.link_params = {}
|
||||
# map from id-string to objid, for files having node names but
|
||||
# not node numbers
|
||||
self.objidmap = {}
|
||||
self.objids = set()
|
||||
self.default_services = {}
|
||||
if self.scenario:
|
||||
self.parse_scenario()
|
||||
|
||||
def info(self, msg):
|
||||
s = 'XML parsing \'%s\': %s' % (self.filename, msg)
|
||||
if self.session:
|
||||
self.session.info(s)
|
||||
else:
|
||||
sys.stdout.write(s + '\n')
|
||||
|
||||
def warn(self, msg):
|
||||
s = 'WARNING XML parsing \'%s\': %s' % (self.filename, msg)
|
||||
if self.session:
|
||||
self.session.warn(s)
|
||||
else:
|
||||
sys.stderr.write(s + '\n')
|
||||
|
||||
@staticmethod
|
||||
def get_scenario(dom):
|
||||
scenario = getFirstChildByTagName(dom, 'scenario')
|
||||
if not scenario:
|
||||
raise ValueError, 'no scenario element found'
|
||||
version = scenario.getAttribute('version')
|
||||
if version and version != '1.0':
|
||||
raise ValueError, \
|
||||
'unsupported scenario version found: \'%s\'' % version
|
||||
return scenario
|
||||
|
||||
def parse_scenario(self):
|
||||
self.parse_default_services()
|
||||
self.parse_session_config()
|
||||
self.parse_network_plan()
|
||||
|
||||
def assign_id(self, idstr, idval):
|
||||
if idstr in self.objidmap:
|
||||
assert self.objidmap[idstr] == idval and idval in self.objids
|
||||
return
|
||||
self.objidmap[idstr] = idval
|
||||
self.objids.add(idval)
|
||||
|
||||
def rand_id(self):
|
||||
while True:
|
||||
x = random.randint(0, 0xffff)
|
||||
if x not in self.objids:
|
||||
return x
|
||||
|
||||
def get_id(self, idstr):
|
||||
'''\
|
||||
Get a, possibly new, object id (node number) corresponding to
|
||||
the given XML string id.
|
||||
'''
|
||||
if not idstr:
|
||||
idn = self.rand_id()
|
||||
self.objids.add(idn)
|
||||
return idn
|
||||
elif idstr in self.objidmap:
|
||||
return self.objidmap[idstr]
|
||||
else:
|
||||
try:
|
||||
idn = int(idstr)
|
||||
except ValueError:
|
||||
idn = self.rand_id()
|
||||
self.assign_id(idstr, idn)
|
||||
return idn
|
||||
|
||||
def get_common_attributes(self, node):
|
||||
'''\
|
||||
Return id, name attributes for the given XML element. These
|
||||
attributes are common to nodes and networks.
|
||||
'''
|
||||
idstr = node.getAttribute('id')
|
||||
# use an explicit set COREID if it exists
|
||||
coreid = self.find_core_id(node)
|
||||
if coreid:
|
||||
idn = int(coreid)
|
||||
if idstr:
|
||||
self.assign_id(idstr, idn)
|
||||
else:
|
||||
idn = self.get_id(idstr)
|
||||
# TODO: consider supporting unicode; for now convert to an
|
||||
# ascii string
|
||||
namestr = str(node.getAttribute('name'))
|
||||
return idn, namestr
|
||||
|
||||
def iter_network_member_devices(self, element):
|
||||
# element can be a network or a channel
|
||||
for interface in iterChildrenWithAttribute(element, 'member',
|
||||
'type', 'interface'):
|
||||
if_id = getChildTextTrim(interface)
|
||||
assert if_id # XXX for testing
|
||||
if not if_id:
|
||||
continue
|
||||
device, if_name = self.find_device_with_interface(if_id)
|
||||
assert device, 'no device for if_id: %s' % if_id # XXX for testing
|
||||
if device:
|
||||
yield device, if_name
|
||||
|
||||
def network_class(self, network, network_type):
|
||||
'''\
|
||||
Return the corresponding CORE network class for the given
|
||||
network/network_type.
|
||||
'''
|
||||
if network_type == 'ethernet':
|
||||
return nodes.PtpNet
|
||||
elif network_type == 'satcom':
|
||||
return nodes.PtpNet
|
||||
elif network_type == 'wireless':
|
||||
channel = getFirstChildByTagName(network, 'channel')
|
||||
if channel:
|
||||
# use an explicit CORE type if it exists
|
||||
coretype = getFirstChildTextTrimWithAttribute(channel, 'type',
|
||||
'domain', 'CORE')
|
||||
if coretype:
|
||||
if coretype == 'basic_range':
|
||||
return nodes.WlanNode
|
||||
elif coretype.startswith('emane'):
|
||||
return nodes.EmaneNode
|
||||
else:
|
||||
self.warn('unknown network type: \'%s\'' % coretype)
|
||||
return xmltypetonodeclass(self.session, coretype)
|
||||
return nodes.WlanNode
|
||||
self.warn('unknown network type: \'%s\'' % network_type)
|
||||
return None
|
||||
|
||||
def create_core_object(self, objcls, objid, objname, element, node_type):
|
||||
obj = self.session.addobj(cls = objcls, objid = objid,
|
||||
name = objname, start = self.start)
|
||||
if self.verbose:
|
||||
self.info('added object objid=%s name=%s cls=%s' % \
|
||||
(objid, objname, objcls))
|
||||
self.set_object_position(obj, element)
|
||||
self.set_object_presentation(obj, element, node_type)
|
||||
return obj
|
||||
|
||||
def get_core_object(self, idstr):
|
||||
if idstr and idstr in self.objidmap:
|
||||
objid = self.objidmap[idstr]
|
||||
return self.session.obj(objid)
|
||||
return None
|
||||
|
||||
def parse_network_plan(self):
|
||||
# parse the scenario in the following order:
|
||||
# 1. layer-2 devices
|
||||
# 2. other networks (ptp/wlan)
|
||||
# 3. layer-3 devices
|
||||
self.parse_layer2_devices()
|
||||
self.parse_networks()
|
||||
self.parse_layer3_devices()
|
||||
|
||||
def set_ethernet_link_parameters(self, channel, link_params,
|
||||
mobility_model_name, mobility_params):
|
||||
# save link parameters for later use, indexed by the tuple
|
||||
# (device_id, interface_name)
|
||||
for dev, if_name in self.iter_network_member_devices(channel):
|
||||
if self.device_type(dev) in self.device_types:
|
||||
dev_id = dev.getAttribute('id')
|
||||
key = (dev_id, if_name)
|
||||
self.link_params[key] = link_params
|
||||
if mobility_model_name or mobility_params:
|
||||
raise NotImplementedError
|
||||
|
||||
def set_wireless_link_parameters(self, channel, link_params,
|
||||
mobility_model_name, mobility_params):
|
||||
network = self.find_channel_network(channel)
|
||||
network_id = network.getAttribute('id')
|
||||
if network_id in self.objidmap:
|
||||
nodenum = self.objidmap[network_id]
|
||||
else:
|
||||
self.warn('unknown network: %s' % network.toxml('utf-8'))
|
||||
assert False # XXX for testing
|
||||
return
|
||||
model_name = getFirstChildTextTrimWithAttribute(channel, 'type',
|
||||
'domain', 'CORE')
|
||||
if not model_name:
|
||||
model_name = 'basic_range'
|
||||
if model_name == 'basic_range':
|
||||
mgr = self.session.mobility
|
||||
elif model_name.startswith('emane'):
|
||||
mgr = self.session.emane
|
||||
elif model_name.startswith('xen'):
|
||||
mgr = self.session.xen
|
||||
else:
|
||||
# TODO: any other config managers?
|
||||
raise NotImplementedError
|
||||
mgr.setconfig_keyvalues(nodenum, model_name, link_params.items())
|
||||
if mobility_model_name and mobility_params:
|
||||
mgr.setconfig_keyvalues(nodenum, mobility_model_name,
|
||||
mobility_params.items())
|
||||
|
||||
def link_layer2_devices(self, device1, ifname1, device2, ifname2):
|
||||
'''\
|
||||
Link two layer-2 devices together.
|
||||
'''
|
||||
devid1 = device1.getAttribute('id')
|
||||
dev1 = self.get_core_object(devid1)
|
||||
devid2 = device2.getAttribute('id')
|
||||
dev2 = self.get_core_object(devid2)
|
||||
assert dev1 and dev2 # XXX for testing
|
||||
if dev1 and dev2:
|
||||
# TODO: review this
|
||||
if isinstance(dev2, nodes.RJ45Node):
|
||||
# RJ45 nodes have different linknet()
|
||||
netif = dev2.linknet(dev1)
|
||||
else:
|
||||
netif = dev1.linknet(dev2)
|
||||
self.set_wired_link_parameters(dev1, netif, devid1, ifname1)
|
||||
|
||||
@classmethod
|
||||
def parse_xml_value(cls, valtext):
|
||||
if not valtext:
|
||||
return None
|
||||
try:
|
||||
if not valtext.translate(None, '0123456789'):
|
||||
val = int(valtext)
|
||||
else:
|
||||
val = float(valtext)
|
||||
except ValueError:
|
||||
val = str(valtext)
|
||||
return val
|
||||
|
||||
@classmethod
|
||||
def parse_parameter_children(cls, parent):
|
||||
params = {}
|
||||
for parameter in iterChildrenWithName(parent, 'parameter'):
|
||||
param_name = parameter.getAttribute('name')
|
||||
assert param_name # XXX for testing
|
||||
if not param_name:
|
||||
continue
|
||||
# TODO: consider supporting unicode; for now convert
|
||||
# to an ascii string
|
||||
param_name = str(param_name)
|
||||
param_val = cls.parse_xml_value(getChildTextTrim(parameter))
|
||||
# TODO: check if the name already exists?
|
||||
if param_name and param_val:
|
||||
params[param_name] = param_val
|
||||
return params
|
||||
|
||||
def parse_network_channel(self, channel):
|
||||
element = self.search_for_element(channel, 'type',
|
||||
lambda x: not x.hasAttributes())
|
||||
channel_type = getChildTextTrim(element)
|
||||
link_params = self.parse_parameter_children(channel)
|
||||
|
||||
mobility = getFirstChildByTagName(channel, 'CORE:mobility')
|
||||
if mobility:
|
||||
mobility_model_name = \
|
||||
getFirstChildTextTrimByTagName(mobility, 'type')
|
||||
mobility_params = self.parse_parameter_children(mobility)
|
||||
else:
|
||||
mobility_model_name = None
|
||||
mobility_params = None
|
||||
if channel_type == 'wireless':
|
||||
self.set_wireless_link_parameters(channel, link_params,
|
||||
mobility_model_name,
|
||||
mobility_params)
|
||||
elif channel_type == 'ethernet':
|
||||
# TODO: maybe this can be done in the loop below to avoid
|
||||
# iterating through channel members multiple times
|
||||
self.set_ethernet_link_parameters(channel, link_params,
|
||||
mobility_model_name,
|
||||
mobility_params)
|
||||
else:
|
||||
raise NotImplementedError
|
||||
layer2_device = []
|
||||
for dev, if_name in self.iter_network_member_devices(channel):
|
||||
if self.device_type(dev) in self.layer2_device_types:
|
||||
layer2_device.append((dev, if_name))
|
||||
assert len(layer2_device) <= 2
|
||||
if len(layer2_device) == 2:
|
||||
self.link_layer2_devices(layer2_device[0][0], layer2_device[0][1],
|
||||
layer2_device[1][0], layer2_device[1][1])
|
||||
|
||||
def parse_network(self, network):
|
||||
'''\
|
||||
Each network element should have an 'id' and 'name' attribute
|
||||
and include the following child elements:
|
||||
|
||||
type (one)
|
||||
member (zero or more with type="interface" or type="channel")
|
||||
channel (zero or more)
|
||||
'''
|
||||
layer2_members = set()
|
||||
layer3_members = 0
|
||||
for dev, if_name in self.iter_network_member_devices(network):
|
||||
if not dev:
|
||||
continue
|
||||
devtype = self.device_type(dev)
|
||||
if devtype in self.layer2_device_types:
|
||||
layer2_members.add(dev)
|
||||
elif devtype in self.layer3_device_types:
|
||||
layer3_members += 1
|
||||
else:
|
||||
raise NotImplementedError
|
||||
if len(layer2_members) == 0:
|
||||
net_type = getFirstChildTextTrimByTagName(network, 'type')
|
||||
if not net_type:
|
||||
msg = 'no network type found for network: \'%s\'' % \
|
||||
network.toxml('utf-8')
|
||||
self.warn(msg)
|
||||
assert False # XXX for testing
|
||||
return
|
||||
net_cls = self.network_class(network, net_type)
|
||||
objid, net_name = self.get_common_attributes(network)
|
||||
if self.verbose:
|
||||
self.info('parsing network: %s %s' % (net_name, objid))
|
||||
if objid in self.session._objs:
|
||||
return
|
||||
n = self.create_core_object(net_cls, objid, net_name,
|
||||
network, None)
|
||||
# handle channel parameters
|
||||
for channel in iterChildrenWithName(network, 'channel'):
|
||||
self.parse_network_channel(channel)
|
||||
|
||||
def parse_networks(self):
|
||||
'''\
|
||||
Parse all 'network' elements.
|
||||
'''
|
||||
for network in iterDescendantsWithName(self.scenario, 'network'):
|
||||
self.parse_network(network)
|
||||
|
||||
def parse_addresses(self, interface):
|
||||
mac = []
|
||||
ipv4 = []
|
||||
ipv6= []
|
||||
hostname = []
|
||||
for address in iterChildrenWithName(interface, 'address'):
|
||||
addr_type = address.getAttribute('type')
|
||||
if not addr_type:
|
||||
msg = 'no type attribute found for address ' \
|
||||
'in interface: \'%s\'' % interface.toxml('utf-8')
|
||||
self.warn(msg)
|
||||
assert False # XXX for testing
|
||||
continue
|
||||
addr_text = getChildTextTrim(address)
|
||||
if not addr_text:
|
||||
msg = 'no text found for address ' \
|
||||
'in interface: \'%s\'' % interface.toxml('utf-8')
|
||||
self.warn(msg)
|
||||
assert False # XXX for testing
|
||||
continue
|
||||
if addr_type == 'mac':
|
||||
mac.append(addr_text)
|
||||
elif addr_type == 'IPv4':
|
||||
ipv4.append(addr_text)
|
||||
elif addr_type == 'IPv6':
|
||||
ipv6.append(addr_text)
|
||||
elif addr_type == 'hostname':
|
||||
hostname.append(addr_text)
|
||||
else:
|
||||
msg = 'skipping unknown address type \'%s\' in ' \
|
||||
'interface: \'%s\'' % (addr_type, interface.toxml('utf-8'))
|
||||
self.warn(msg)
|
||||
assert False # XXX for testing
|
||||
continue
|
||||
return mac, ipv4, ipv6, hostname
|
||||
|
||||
def parse_interface(self, node, device_id, interface):
|
||||
'''\
|
||||
Each interface can have multiple 'address' elements.
|
||||
'''
|
||||
if_name = interface.getAttribute('name')
|
||||
network = self.find_interface_network_object(interface)
|
||||
if not network:
|
||||
msg = 'skipping node \'%s\' interface \'%s\': ' \
|
||||
'unknown network' % (node.name, if_name)
|
||||
self.warn(msg)
|
||||
assert False # XXX for testing
|
||||
return
|
||||
mac, ipv4, ipv6, hostname = self.parse_addresses(interface)
|
||||
if mac:
|
||||
hwaddr = MacAddr.fromstring(mac[0])
|
||||
else:
|
||||
hwaddr = None
|
||||
ifindex = node.newnetif(network, addrlist = ipv4 + ipv6,
|
||||
hwaddr = hwaddr, ifindex = None,
|
||||
ifname = if_name)
|
||||
# TODO: 'hostname' addresses are unused
|
||||
if self.verbose:
|
||||
msg = 'node \'%s\' interface \'%s\' connected ' \
|
||||
'to network \'%s\'' % (node.name, if_name, network.name)
|
||||
self.info(msg)
|
||||
# set link parameters for wired links
|
||||
if isinstance(network,
|
||||
(nodes.HubNode, nodes.PtpNet, nodes.SwitchNode)):
|
||||
netif = node.netif(ifindex)
|
||||
self.set_wired_link_parameters(network, netif, device_id)
|
||||
|
||||
def set_wired_link_parameters(self, network, netif,
|
||||
device_id, netif_name = None):
|
||||
if netif_name is None:
|
||||
netif_name = netif.name
|
||||
key = (device_id, netif_name)
|
||||
if key in self.link_params:
|
||||
link_params = self.link_params[key]
|
||||
if self.start:
|
||||
bw = link_params.get('bw')
|
||||
delay = link_params.get('delay')
|
||||
loss = link_params.get('loss')
|
||||
duplicate = link_params.get('duplicate')
|
||||
jitter = link_params.get('jitter')
|
||||
network.linkconfig(netif, bw = bw, delay = delay, loss = loss,
|
||||
duplicate = duplicate, jitter = jitter)
|
||||
else:
|
||||
for k, v in link_params.iteritems():
|
||||
netif.setparam(k, v)
|
||||
|
||||
@staticmethod
|
||||
def search_for_element(node, tagName, match = None):
|
||||
'''\
|
||||
Search the given node and all ancestors for an element named
|
||||
tagName that satisfies the given matching function.
|
||||
'''
|
||||
while True:
|
||||
for child in iterChildren(node, Node.ELEMENT_NODE):
|
||||
if child.tagName == tagName and \
|
||||
(match is None or match(child)):
|
||||
return child
|
||||
node = node.parentNode
|
||||
if not node:
|
||||
break
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def find_core_id(cls, node):
|
||||
def match(x):
|
||||
domain = x.getAttribute('domain')
|
||||
return domain == 'COREID'
|
||||
alias = cls.search_for_element(node, 'alias', match)
|
||||
if alias:
|
||||
return getChildTextTrim(alias)
|
||||
return None
|
||||
|
||||
@classmethod
|
||||
def find_point(cls, node):
|
||||
return cls.search_for_element(node, 'point')
|
||||
|
||||
@staticmethod
|
||||
def find_channel_network(channel):
|
||||
p = channel.parentNode
|
||||
if p and p.tagName == 'network':
|
||||
return p
|
||||
return None
|
||||
|
||||
def find_interface_network_object(self, interface):
|
||||
network_id = getFirstChildTextTrimWithAttribute(interface, 'member',
|
||||
'type', 'network')
|
||||
if not network_id:
|
||||
# support legacy notation: <interface net="netid" ...
|
||||
network_id = interface.getAttribute('net')
|
||||
obj = self.get_core_object(network_id)
|
||||
if obj:
|
||||
# the network_id should exist for ptp or wlan/emane networks
|
||||
return obj
|
||||
# the network should correspond to a layer-2 device if the
|
||||
# network_id does not exist
|
||||
channel_id = getFirstChildTextTrimWithAttribute(interface, 'member',
|
||||
'type', 'channel')
|
||||
if not network_id or not channel_id:
|
||||
return None
|
||||
network = getFirstChildWithAttribute(self.scenario, 'network',
|
||||
'id', network_id)
|
||||
if not network:
|
||||
return None
|
||||
channel = getFirstChildWithAttribute(network, 'channel',
|
||||
'id', channel_id)
|
||||
if not channel:
|
||||
return None
|
||||
device = None
|
||||
for dev, if_name in self.iter_network_member_devices(channel):
|
||||
if self.device_type(dev) in self.layer2_device_types:
|
||||
assert not device # XXX
|
||||
device = dev
|
||||
if device:
|
||||
obj = self.get_core_object(device.getAttribute('id'))
|
||||
if obj:
|
||||
return obj
|
||||
return None
|
||||
|
||||
def set_object_position_pixel(self, obj, point):
|
||||
x = float(point.getAttribute('x'))
|
||||
y = float(point.getAttribute('y'))
|
||||
z = point.getAttribute('z')
|
||||
if z:
|
||||
z = float(z)
|
||||
else:
|
||||
z = 0.0
|
||||
# TODO: zMode is unused
|
||||
# z_mode = point.getAttribute('zMode'))
|
||||
if x < 0.0:
|
||||
self.warn('limiting negative x position of \'%s\' to zero: %s' %
|
||||
(obj.name, x))
|
||||
x = 0.0
|
||||
if y < 0.0:
|
||||
self.warn('limiting negative y position of \'%s\' to zero: %s' %
|
||||
(obj.name, y))
|
||||
y = 0.0
|
||||
obj.setposition(x, y, z)
|
||||
|
||||
def set_object_position_gps(self, obj, point):
|
||||
lat = float(point.getAttribute('lat'))
|
||||
lon = float(point.getAttribute('lon'))
|
||||
zalt = point.getAttribute('z')
|
||||
if zalt:
|
||||
zalt = float(zalt)
|
||||
else:
|
||||
zalt = 0.0
|
||||
# TODO: zMode is unused
|
||||
# z_mode = point.getAttribute('zMode'))
|
||||
if not self.location_refgeo_set:
|
||||
# for x,y,z conversion, we need a reasonable refpt; this
|
||||
# picks the first coordinates as the origin
|
||||
self.session.location.setrefgeo(lat, lon, zalt)
|
||||
self.location_refgeo_set = True
|
||||
x, y, z = self.session.location.getxyz(lat, lon, zalt)
|
||||
if x < 0.0:
|
||||
self.warn('limiting negative x position of \'%s\' to zero: %s' %
|
||||
(obj.name, x))
|
||||
x = 0.0
|
||||
if y < 0.0:
|
||||
self.warn('limiting negative y position of \'%s\' to zero: %s' %
|
||||
(obj.name, y))
|
||||
y = 0.0
|
||||
obj.setposition(x, y, z)
|
||||
|
||||
def set_object_position_cartesian(self, obj, point):
|
||||
# TODO: review this
|
||||
xm = float(point.getAttribute('x'))
|
||||
ym = float(point.getAttribute('y'))
|
||||
zm = point.getAttribute('z')
|
||||
if zm:
|
||||
zm = float(zm)
|
||||
else:
|
||||
zm = 0.0
|
||||
# TODO: zMode is unused
|
||||
# z_mode = point.getAttribute('zMode'))
|
||||
if not self.location_refxyz_set:
|
||||
self.session.location.refxyz = xm, ym, zm
|
||||
self.location_refxyz_set = True
|
||||
# need to convert meters to pixels
|
||||
x = self.session.location.m2px(xm) + self.session.location.refxyz[0]
|
||||
y = self.session.location.m2px(ym) + self.session.location.refxyz[1]
|
||||
z = self.session.location.m2px(zm) + self.session.location.refxyz[2]
|
||||
if x < 0.0:
|
||||
self.warn('limiting negative x position of \'%s\' to zero: %s' %
|
||||
(obj.name, x))
|
||||
x = 0.0
|
||||
if y < 0.0:
|
||||
self.warn('limiting negative y position of \'%s\' to zero: %s' %
|
||||
(obj.name, y))
|
||||
y = 0.0
|
||||
obj.setposition(x, y, z)
|
||||
|
||||
def set_object_position(self, obj, element):
|
||||
'''\
|
||||
Set the x,y,x position of obj from the point associated with
|
||||
the given element.
|
||||
'''
|
||||
point = self.find_point(element)
|
||||
if not point:
|
||||
return False
|
||||
point_type = point.getAttribute('type')
|
||||
if not point_type:
|
||||
msg = 'no type attribute found for point: \'%s\'' % \
|
||||
point.toxml('utf-8')
|
||||
self.warn(msg)
|
||||
assert False # XXX for testing
|
||||
return False
|
||||
elif point_type == 'pixel':
|
||||
self.set_object_position_pixel(obj, point)
|
||||
elif point_type == 'gps':
|
||||
self.set_object_position_gps(obj, point)
|
||||
elif point_type == 'cart':
|
||||
self.set_object_position_cartesian(obj, point)
|
||||
else:
|
||||
self.warn("skipping unknown point type: '%s'" % point_type)
|
||||
assert False # XXX for testing
|
||||
return False
|
||||
if self.verbose:
|
||||
msg = 'set position of %s from point element: \'%s\'' % \
|
||||
(obj.name, point.toxml('utf-8'))
|
||||
self.info(msg)
|
||||
return True
|
||||
|
||||
def parse_device_service(self, service, node):
|
||||
name = service.getAttribute('name')
|
||||
session_service = self.session.services.getservicebyname(name)
|
||||
if not session_service:
|
||||
assert False # XXX for testing
|
||||
return None
|
||||
values = []
|
||||
startup_idx = service.getAttribute('startup_idx')
|
||||
if startup_idx:
|
||||
values.append('startidx=%s' % startup_idx)
|
||||
startup_time = service.getAttribute('start_time')
|
||||
if startup_time:
|
||||
values.append('starttime=%s' % startup_time)
|
||||
dirs = []
|
||||
for directory in iterChildrenWithName(service, 'directory'):
|
||||
dirname = directory.getAttribute('name')
|
||||
dirs.append(str(dirname))
|
||||
if dirs:
|
||||
values.append("dirs=%s" % dirs)
|
||||
startup = []
|
||||
shutdown = []
|
||||
validate = []
|
||||
for command in iterChildrenWithName(service, 'command'):
|
||||
command_type = command.getAttribute('type')
|
||||
command_text = getChildTextTrim(command)
|
||||
if not command_text:
|
||||
continue
|
||||
if command_type == 'start':
|
||||
startup.append(str(command_text))
|
||||
elif command_type == 'stop':
|
||||
shutdown.append(str(command_text))
|
||||
elif command_type == 'validate':
|
||||
validate.append(str(command_text))
|
||||
if startup:
|
||||
values.append('cmdup=%s' % startup)
|
||||
if shutdown:
|
||||
values.append('cmddown=%s' % shutdown)
|
||||
if validate:
|
||||
values.append('cmdval=%s' % validate)
|
||||
filenames = []
|
||||
files = []
|
||||
for f in iterChildrenWithName(service, 'file'):
|
||||
filename = f.getAttribute('name')
|
||||
if not filename:
|
||||
continue;
|
||||
filenames.append(filename)
|
||||
data = getChildTextTrim(f)
|
||||
if data:
|
||||
data = str(data)
|
||||
else:
|
||||
data = None
|
||||
typestr = 'service:%s:%s' % (name, filename)
|
||||
files.append((typestr, filename, data))
|
||||
if filenames:
|
||||
values.append('files=%s' % filenames)
|
||||
custom = service.getAttribute('custom')
|
||||
if custom and custom.lower() == 'true':
|
||||
self.session.services.setcustomservice(node.objid,
|
||||
session_service, values)
|
||||
# NOTE: if a custom service is used, setservicefile() must be
|
||||
# called after the custom service exists
|
||||
for typestr, filename, data in files:
|
||||
self.session.services.setservicefile(nodenum = node.objid,
|
||||
type = typestr,
|
||||
filename = filename,
|
||||
srcname = None,
|
||||
data = data)
|
||||
return str(name)
|
||||
|
||||
def parse_device_services(self, services, node):
|
||||
'''\
|
||||
Use session.services manager to store service customizations
|
||||
before they are added to a node.
|
||||
'''
|
||||
service_names = []
|
||||
for service in iterChildrenWithName(services, 'service'):
|
||||
name = self.parse_device_service(service, node)
|
||||
if name:
|
||||
service_names.append(name)
|
||||
return '|'.join(service_names)
|
||||
|
||||
def add_device_services(self, node, device, node_type):
|
||||
'''\
|
||||
Add services to the given node.
|
||||
'''
|
||||
services = getFirstChildByTagName(device, 'CORE:services')
|
||||
if services:
|
||||
services_str = self.parse_device_services(services, node)
|
||||
if self.verbose:
|
||||
self.info('services for node \'%s\': %s' % \
|
||||
(node.name, services_str))
|
||||
elif node_type in self.default_services:
|
||||
services_str = None # default services will be added
|
||||
else:
|
||||
return
|
||||
self.session.services.addservicestonode(node = node,
|
||||
nodetype = node_type,
|
||||
services_str = services_str,
|
||||
verbose = self.verbose)
|
||||
|
||||
def set_object_presentation(self, obj, element, node_type):
|
||||
# defaults from the CORE GUI
|
||||
default_icons = {
|
||||
'router': 'router.gif',
|
||||
'host': 'host.gif',
|
||||
'PC': 'pc.gif',
|
||||
'mdr': 'mdr.gif',
|
||||
# 'prouter': 'router_green.gif',
|
||||
# 'xen': 'xen.gif'
|
||||
}
|
||||
icon_set = False
|
||||
for child in iterChildrenWithName(element, 'CORE:presentation'):
|
||||
canvas = child.getAttribute('canvas')
|
||||
if canvas:
|
||||
obj.canvas = int(canvas)
|
||||
icon = child.getAttribute('icon')
|
||||
if icon:
|
||||
icon = str(icon).replace("$CORE_DATA_DIR",
|
||||
constants.CORE_DATA_DIR)
|
||||
obj.icon = icon
|
||||
icon_set = True
|
||||
if not icon_set and node_type in default_icons:
|
||||
obj.icon = default_icons[node_type]
|
||||
|
||||
def device_type(self, device):
|
||||
if device.tagName in self.device_types:
|
||||
return device.tagName
|
||||
return None
|
||||
|
||||
def core_node_type(self, device):
|
||||
# use an explicit CORE type if it exists
|
||||
coretype = getFirstChildTextTrimWithAttribute(device, 'type',
|
||||
'domain', 'CORE')
|
||||
if coretype:
|
||||
return coretype
|
||||
return self.device_type(device)
|
||||
|
||||
def find_device_with_interface(self, interface_id):
|
||||
# TODO: suport generic 'device' elements
|
||||
for device in iterDescendantsWithName(self.scenario,
|
||||
self.device_types):
|
||||
interface = getFirstChildWithAttribute(device, 'interface',
|
||||
'id', interface_id)
|
||||
if interface:
|
||||
if_name = interface.getAttribute('name')
|
||||
return device, if_name
|
||||
return None, None
|
||||
|
||||
def parse_layer2_device(self, device):
|
||||
objid, device_name = self.get_common_attributes(device)
|
||||
if self.verbose:
|
||||
self.info('parsing layer-2 device: %s %s' % (device_name, objid))
|
||||
try:
|
||||
return self.session.obj(objid)
|
||||
except KeyError:
|
||||
pass
|
||||
device_type = self.device_type(device)
|
||||
if device_type == 'hub':
|
||||
device_class = nodes.HubNode
|
||||
elif device_type == 'switch':
|
||||
device_class = nodes.SwitchNode
|
||||
else:
|
||||
self.warn('unknown layer-2 device type: \'%s\'' % device_type)
|
||||
assert False # XXX for testing
|
||||
return None
|
||||
n = self.create_core_object(device_class, objid, device_name,
|
||||
device, None)
|
||||
return n
|
||||
|
||||
def parse_layer3_device(self, device):
|
||||
objid, device_name = self.get_common_attributes(device)
|
||||
if self.verbose:
|
||||
self.info('parsing layer-3 device: %s %s' % (device_name, objid))
|
||||
try:
|
||||
return self.session.obj(objid)
|
||||
except KeyError:
|
||||
pass
|
||||
device_cls = self.nodecls
|
||||
core_node_type = self.core_node_type(device)
|
||||
n = self.create_core_object(device_cls, objid, device_name,
|
||||
device, core_node_type)
|
||||
n.type = core_node_type
|
||||
self.add_device_services(n, device, core_node_type)
|
||||
for interface in iterChildrenWithName(device, 'interface'):
|
||||
self.parse_interface(n, device.getAttribute('id'), interface)
|
||||
return n
|
||||
|
||||
def parse_layer2_devices(self):
|
||||
'''\
|
||||
Parse all layer-2 device elements. A device can be: 'switch',
|
||||
'hub'.
|
||||
'''
|
||||
# TODO: suport generic 'device' elements
|
||||
for device in iterDescendantsWithName(self.scenario,
|
||||
self.layer2_device_types):
|
||||
self.parse_layer2_device(device)
|
||||
|
||||
def parse_layer3_devices(self):
|
||||
'''\
|
||||
Parse all layer-3 device elements. A device can be: 'host',
|
||||
'router'.
|
||||
'''
|
||||
# TODO: suport generic 'device' elements
|
||||
for device in iterDescendantsWithName(self.scenario,
|
||||
self.layer3_device_types):
|
||||
self.parse_layer3_device(device)
|
||||
|
||||
def parse_session_origin(self, session_config):
|
||||
'''\
|
||||
Parse the first origin tag and set the CoreLocation reference
|
||||
point appropriately.
|
||||
'''
|
||||
# defaults from the CORE GUI
|
||||
self.session.location.setrefgeo(47.5791667, -122.132322, 2.0)
|
||||
self.session.location.refscale = 150.0
|
||||
origin = getFirstChildByTagName(session_config, 'origin')
|
||||
if not origin:
|
||||
return
|
||||
lat = origin.getAttribute('lat')
|
||||
lon = origin.getAttribute('lon')
|
||||
alt = origin.getAttribute('alt')
|
||||
if lat and lon and alt:
|
||||
self.session.location.setrefgeo(float(lat), float(lon), float(alt))
|
||||
self.location_refgeo_set = True
|
||||
scale100 = origin.getAttribute("scale100")
|
||||
if scale100:
|
||||
self.session.location.refscale = float(scale100)
|
||||
point = getFirstChildTextTrimByTagName(origin, 'point')
|
||||
if point:
|
||||
xyz = point.split(',')
|
||||
if len(xyz) == 2:
|
||||
xyz.append('0.0')
|
||||
if len(xyz) == 3:
|
||||
self.session.location.refxyz = \
|
||||
(float(xyz[0]), float(xyz[1]), float(xyz[2]))
|
||||
self.location_refxyz_set = True
|
||||
|
||||
def parse_session_options(self, session_config):
|
||||
options = getFirstChildByTagName(session_config, 'options')
|
||||
if not options:
|
||||
return
|
||||
params = self.parse_parameter_children(options)
|
||||
for name, value in params.iteritems():
|
||||
if name and value:
|
||||
setattr(self.session.options, str(name), str(value))
|
||||
|
||||
def parse_session_hooks(self, session_config):
|
||||
'''\
|
||||
Parse hook scripts.
|
||||
'''
|
||||
hooks = getFirstChildByTagName(session_config, 'hooks')
|
||||
if not hooks:
|
||||
return
|
||||
for hook in iterChildrenWithName(hooks, 'hook'):
|
||||
filename = hook.getAttribute('name')
|
||||
state = hook.getAttribute('state')
|
||||
data = getChildTextTrim(hook)
|
||||
if data is None:
|
||||
data = '' # allow for empty file
|
||||
hook_type = "hook:%s" % state
|
||||
self.session.sethook(hook_type, filename = str(filename),
|
||||
srcname = None, data = str(data))
|
||||
|
||||
def parse_session_metadata(self, session_config):
|
||||
metadata = getFirstChildByTagName(session_config, 'metadata')
|
||||
if not metadata:
|
||||
return
|
||||
params = self.parse_parameter_children(metadata)
|
||||
for name, value in params.iteritems():
|
||||
if name and value:
|
||||
self.session.metadata.additem(str(name), str(value))
|
||||
|
||||
def parse_session_config(self):
|
||||
session_config = \
|
||||
getFirstChildByTagName(self.scenario, 'CORE:sessionconfig')
|
||||
if not session_config:
|
||||
return
|
||||
self.parse_session_origin(session_config)
|
||||
self.parse_session_options(session_config)
|
||||
self.parse_session_hooks(session_config)
|
||||
self.parse_session_metadata(session_config)
|
||||
|
||||
def parse_default_services(self):
|
||||
# defaults from the CORE GUI
|
||||
self.default_services = {
|
||||
'router': ['zebra', 'OSPFv2', 'OSPFv3', 'vtysh', 'IPForward'],
|
||||
'host': ['DefaultRoute', 'SSH'],
|
||||
'PC': ['DefaultRoute',],
|
||||
'mdr': ['zebra', 'OSPFv3MDR', 'vtysh', 'IPForward'],
|
||||
# 'prouter': ['zebra', 'OSPFv2', 'OSPFv3', 'vtysh', 'IPForward'],
|
||||
# 'xen': ['zebra', 'OSPFv2', 'OSPFv3', 'vtysh', 'IPForward'],
|
||||
}
|
||||
default_services = \
|
||||
getFirstChildByTagName(self.scenario, 'CORE:defaultservices')
|
||||
if not default_services:
|
||||
return
|
||||
for device in iterChildrenWithName(default_services, 'device'):
|
||||
device_type = device.getAttribute('type')
|
||||
if not device_type:
|
||||
self.warn('parse_default_services: no type attribute ' \
|
||||
'found for device')
|
||||
continue
|
||||
services = []
|
||||
for service in iterChildrenWithName(device, 'service'):
|
||||
name = service.getAttribute('name')
|
||||
if name:
|
||||
services.append(str(name))
|
||||
self.default_services[device_type] = services
|
||||
# store default services for the session
|
||||
for t, s in self.default_services.iteritems():
|
||||
self.session.services.defaultservices[t] = s
|
||||
if self.verbose:
|
||||
self.info('default services for node type \'%s\' ' \
|
||||
'set to: %s' % (t, s))
|
||||
|
|
@ -23,10 +23,10 @@ def addelementsfromlist(dom, parent, iterable, name, attr_name):
|
|||
element = dom.createElement(name)
|
||||
element.setAttribute(attr_name, item)
|
||||
parent.appendChild(element)
|
||||
|
||||
|
||||
def addtextelementsfromlist(dom, parent, iterable, name, attrs):
|
||||
''' XML helper to iterate through a list and add items to parent using tags
|
||||
of the given name, attributes specified in the attrs tuple, and having the
|
||||
of the given name, attributes specified in the attrs tuple, and having the
|
||||
text of the item within the tags.
|
||||
Example: addtextelementsfromlist(dom, parent, ('a','b','c'), "letter",
|
||||
(('show','True'),))
|
||||
|
|
@ -47,7 +47,7 @@ def addtextelementsfromlist(dom, parent, iterable, name, attrs):
|
|||
def addtextelementsfromtuples(dom, parent, iterable, attrs=()):
|
||||
''' XML helper to iterate through a list of tuples and add items to
|
||||
parent using tags named for the first tuple element,
|
||||
attributes specified in the attrs tuple, and having the
|
||||
attributes specified in the attrs tuple, and having the
|
||||
text of second tuple element.
|
||||
Example: addtextelementsfromtuples(dom, parent,
|
||||
(('first','a'),('second','b'),('third','c')),
|
||||
|
|
@ -95,7 +95,7 @@ def addparamtoparent(dom, parent, name, value):
|
|||
p.setAttribute("name", name)
|
||||
p.setAttribute("value", "%s" % value)
|
||||
return p
|
||||
|
||||
|
||||
def addtextparamtoparent(dom, parent, name, value):
|
||||
''' XML helper to add a <param name="name">value</param> tag to the parent
|
||||
element, when value is not None.
|
||||
|
|
@ -108,7 +108,7 @@ def addtextparamtoparent(dom, parent, name, value):
|
|||
txt = dom.createTextNode(value)
|
||||
p.appendChild(txt)
|
||||
return p
|
||||
|
||||
|
||||
def addparamlisttoparent(dom, parent, name, values):
|
||||
''' XML helper to return a parameter list and optionally add it to the
|
||||
parent element:
|
||||
|
|
@ -128,13 +128,76 @@ def addparamlisttoparent(dom, parent, name, values):
|
|||
item.setAttribute("value", str(v))
|
||||
p.appendChild(item)
|
||||
return p
|
||||
|
||||
|
||||
def getoneelement(dom, name):
|
||||
e = dom.getElementsByTagName(name)
|
||||
if len(e) == 0:
|
||||
return None
|
||||
return e[0]
|
||||
|
||||
def iterDescendants(dom, max_depth = 0):
|
||||
'''\
|
||||
Iterate over all descendant element nodes in breadth first order.
|
||||
Only consider nodes up to max_depth deep when max_depth is greater
|
||||
than zero.
|
||||
'''
|
||||
nodes = [dom]
|
||||
depth = 0
|
||||
current_depth_nodes = 1
|
||||
next_depth_nodes = 0
|
||||
while nodes:
|
||||
n = nodes.pop(0)
|
||||
for child in n.childNodes:
|
||||
if child.nodeType == Node.ELEMENT_NODE:
|
||||
yield child
|
||||
nodes.append(child)
|
||||
next_depth_nodes += 1
|
||||
current_depth_nodes -= 1
|
||||
if current_depth_nodes == 0:
|
||||
depth += 1
|
||||
if max_depth > 0 and depth == max_depth:
|
||||
return
|
||||
current_depth_nodes = next_depth_nodes
|
||||
next_depth_nodes = 0
|
||||
|
||||
def iterMatchingDescendants(dom, matchFunction, max_depth = 0):
|
||||
'''\
|
||||
Iterate over descendant elements where matchFunction(descendant)
|
||||
returns true. Only consider nodes up to max_depth deep when
|
||||
max_depth is greater than zero.
|
||||
'''
|
||||
for d in iterDescendants(dom, max_depth):
|
||||
if matchFunction(d):
|
||||
yield d
|
||||
|
||||
def iterDescendantsWithName(dom, tagName, max_depth = 0):
|
||||
'''\
|
||||
Iterate over descendant elements whose name is contained in
|
||||
tagName (or is named tagName if tagName is a string). Only
|
||||
consider nodes up to max_depth deep when max_depth is greater than
|
||||
zero.
|
||||
'''
|
||||
if isinstance(tagName, basestring):
|
||||
tagName = (tagName,)
|
||||
def match(d):
|
||||
return d.tagName in tagName
|
||||
return iterMatchingDescendants(dom, match, max_depth)
|
||||
|
||||
def iterDescendantsWithAttribute(dom, tagName, attrName, attrValue,
|
||||
max_depth = 0):
|
||||
'''\
|
||||
Iterate over descendant elements whose name is contained in
|
||||
tagName (or is named tagName if tagName is a string) and have an
|
||||
attribute named attrName with value attrValue. Only consider
|
||||
nodes up to max_depth deep when max_depth is greater than zero.
|
||||
'''
|
||||
if isinstance(tagName, basestring):
|
||||
tagName = (tagName,)
|
||||
def match(d):
|
||||
return d.tagName in tagName and \
|
||||
d.getAttribute(attrName) == attrValue
|
||||
return iterMatchingDescendants(dom, match, max_depth)
|
||||
|
||||
def iterChildren(dom, nodeType):
|
||||
'''\
|
||||
Iterate over all child elements of the given type.
|
||||
|
|
@ -151,9 +214,15 @@ def gettextchild(dom):
|
|||
return str(child.nodeValue)
|
||||
return None
|
||||
|
||||
def getChildTextTrim(dom):
|
||||
text = gettextchild(dom)
|
||||
if text:
|
||||
text = text.strip()
|
||||
return text
|
||||
|
||||
def getparamssetattrs(dom, param_names, target):
|
||||
''' XML helper to get <param name="name" value="value"/> tags and set
|
||||
the attribute in the target object. String type is used. Target object
|
||||
the attribute in the target object. String type is used. Target object
|
||||
attribute is unchanged if the XML attribute is not present.
|
||||
'''
|
||||
params = dom.getElementsByTagName("param")
|
||||
|
|
@ -172,3 +241,63 @@ def xmltypetonodeclass(session, type):
|
|||
return eval("nodes.%s" % type)
|
||||
else:
|
||||
return None
|
||||
|
||||
def iterChildrenWithName(dom, tagName):
|
||||
return iterDescendantsWithName(dom, tagName, 1)
|
||||
|
||||
def iterChildrenWithAttribute(dom, tagName, attrName, attrValue):
|
||||
return iterDescendantsWithAttribute(dom, tagName, attrName, attrValue, 1)
|
||||
|
||||
def getFirstChildByTagName(dom, tagName):
|
||||
'''\
|
||||
Return the first child element whose name is contained in tagName
|
||||
(or is named tagName if tagName is a string).
|
||||
'''
|
||||
for child in iterChildrenWithName(dom, tagName):
|
||||
return child
|
||||
return None
|
||||
|
||||
def getFirstChildTextByTagName(dom, tagName):
|
||||
'''\
|
||||
Return the corresponding text of the first child element whose
|
||||
name is contained in tagName (or is named tagName if tagName is a
|
||||
string).
|
||||
'''
|
||||
child = getFirstChildByTagName(dom, tagName)
|
||||
if child:
|
||||
return gettextchild(child)
|
||||
return None
|
||||
|
||||
def getFirstChildTextTrimByTagName(dom, tagName):
|
||||
text = getFirstChildTextByTagName(dom, tagName)
|
||||
if text:
|
||||
text = text.strip()
|
||||
return text
|
||||
|
||||
def getFirstChildWithAttribute(dom, tagName, attrName, attrValue):
|
||||
'''\
|
||||
Return the first child element whose name is contained in tagName
|
||||
(or is named tagName if tagName is a string) that has an attribute
|
||||
named attrName with value attrValue.
|
||||
'''
|
||||
for child in \
|
||||
iterChildrenWithAttribute(dom, tagName, attrName, attrValue):
|
||||
return child
|
||||
return None
|
||||
|
||||
def getFirstChildTextWithAttribute(dom, tagName, attrName, attrValue):
|
||||
'''\
|
||||
Return the corresponding text of the first child element whose
|
||||
name is contained in tagName (or is named tagName if tagName is a
|
||||
string) that has an attribute named attrName with value attrValue.
|
||||
'''
|
||||
child = getFirstChildWithAttribute(dom, tagName, attrName, attrValue)
|
||||
if child:
|
||||
return gettextchild(child)
|
||||
return None
|
||||
|
||||
def getFirstChildTextTrimWithAttribute(dom, tagName, attrName, attrValue):
|
||||
text = getFirstChildTextWithAttribute(dom, tagName, attrName, attrValue)
|
||||
if text:
|
||||
text = text.strip()
|
||||
return text
|
||||
|
|
|
|||
Loading…
Reference in a new issue