mirror of
https://github.com/asterisk/asterisk.git
synced 2025-09-19 11:42:27 +00:00
5d9ac58d8e
** This script is in no way finished. Started the initial "cut" at converting a sip.conf file to a res_sip.conf file. Hopefully the bulk of the framework is in place and only a few minor adjustments need to be made when an option mapping is added that "doesn't fit". This script and supporting files should be executable against python version 2.5. An OrderedDict class (backported from a newer version of python) is included. A MultiOrderedDict class is implemented so options, when added, should be able to be added in order and allowed to have multiple values. Currently the scripts supports the majority of endpoint options found in res_sip.conf. Support has also been added for Aor(s) and the ACL/security sections. Inside the sip_to_res_sip.py file one can see a list of options that still need to be mapped. Also items that still need to be done: templates, includes, parsing '=>' delimiter. Note that some code is hopefully in place already to support templates (e.g. lookup/retrieving defaults from them). However, the parsing of and adding of the section needs to be done. git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@394024 65c4cc65-6c06-0410-ace0-fbb531ad65f3
243 lines
8.5 KiB
Python
243 lines
8.5 KiB
Python
from astdicts import MultiOrderedDict
|
|
|
|
def merge_values(left, right, key):
|
|
"""Merges values from right into left."""
|
|
if isinstance(left, list):
|
|
vals0 = left
|
|
else: # assume dictionary
|
|
vals0 = left[key] if key in left else []
|
|
vals1 = right[key] if key in right else []
|
|
|
|
return vals0 + [i for i in vals1 if i not in vals0]
|
|
|
|
###############################################################################
|
|
|
|
class Section(MultiOrderedDict):
|
|
"""A Section is a MultiOrderedDict itself that maintains a list of
|
|
key/value options. However, in the case of an Asterisk config
|
|
file a section may have other defaults sections that is can pull
|
|
data from (i.e. templates). So when an option is looked up by key
|
|
it first checks the base section and if not found looks in the
|
|
added default sections. If not found at that point then a 'KeyError'
|
|
exception is raised.
|
|
"""
|
|
def __init__(self, defaults = []):
|
|
MultiOrderedDict.__init__(self)
|
|
self._defaults = defaults
|
|
|
|
def __getitem__(self, key):
|
|
"""Get the value for the given key. If it is not found in the 'self'
|
|
then check inside the defaults before declaring unable to locate."""
|
|
if key in self:
|
|
return MultiOrderedDict.__getitem__(self, key)
|
|
|
|
for default in self._defaults:
|
|
if key in default:
|
|
return default[key]
|
|
|
|
raise KeyError(key)
|
|
|
|
def keys(self):
|
|
res = MultiOrderedDict.keys(self)
|
|
for d in self._defaults:
|
|
for key in d.keys():
|
|
if key not in res:
|
|
res.append(key)
|
|
return res
|
|
|
|
def add_default(self, default):
|
|
self._defaults.append(default)
|
|
|
|
def get_merged(self, key):
|
|
"""Return a list of values for a given key merged from default(s)"""
|
|
# first merge key/values from defaults together
|
|
merged = []
|
|
for i in self._defaults:
|
|
if not merged:
|
|
merged = i
|
|
continue
|
|
merged = merge_values(merged, i, key)
|
|
# then merge self in
|
|
return merge_values(merged, self, key)
|
|
|
|
###############################################################################
|
|
|
|
def remove_comment(line):
|
|
"""Remove any commented elements from the given line"""
|
|
line = line.partition(COMMENT)[0]
|
|
return line.rstrip()
|
|
|
|
def try_section(line):
|
|
"""Checks to see if the given line is a section. If so return the section
|
|
name, otherwise return 'None'.
|
|
"""
|
|
if not line.startswith('['):
|
|
return None
|
|
|
|
first, second, third = line.partition(']')
|
|
# TODO - third may contain template, parse to see if it is a template
|
|
# or is a list of templates...return?
|
|
return first[1:]
|
|
|
|
def try_option(line):
|
|
"""Parses the line as an option, returning the key/value pair."""
|
|
first, second, third = line.partition('=')
|
|
return first.strip(), third.strip()
|
|
|
|
###############################################################################
|
|
|
|
def get_value(mdict, key, index=-1):
|
|
"""Given a multi-dict, retrieves a value for the given key. If the key only
|
|
holds a single value return that value. If the key holds more than one
|
|
value and an index is given that is greater than or equal to zero then
|
|
return the value at the index. Otherwise return the list of values."""
|
|
vals = mdict[key]
|
|
if len(vals) == 1:
|
|
return vals[0]
|
|
if index >= 0:
|
|
return vals[index]
|
|
return vals
|
|
|
|
def find_value(mdicts, key, index=-1):
|
|
"""Given a list of multi-dicts, try to find value(s) for the given key."""
|
|
if not isinstance(mdicts, list):
|
|
# given a single multi-dict
|
|
return get_value(mdicts, key, index)
|
|
|
|
for d in mdicts:
|
|
if key in d:
|
|
return d[key]
|
|
# not found, throw error
|
|
raise KeyError(key)
|
|
|
|
def find_dict(mdicts, key, val):
|
|
"""Given a list of mult-dicts, return the multi-dict that contains
|
|
the given key/value pair."""
|
|
|
|
def found(d):
|
|
# just check the first value of the key
|
|
return key in d and d[key][0] == val
|
|
|
|
if isinstance(mdicts, list):
|
|
try:
|
|
return [d for d in mdicts if found(d)][0]
|
|
except IndexError:
|
|
pass
|
|
elif found(mdicts):
|
|
return mdicts
|
|
|
|
raise LookupError("Dictionary not located for key = %s, value = %s"
|
|
% (key, val))
|
|
|
|
###############################################################################
|
|
|
|
COMMENT = ';'
|
|
DEFAULTSECT = 'general'
|
|
|
|
class MultiOrderedConfigParser:
|
|
def __init__(self):
|
|
self._default = MultiOrderedDict()
|
|
# sections contain dictionaries of dictionaries
|
|
self._sections = MultiOrderedDict()
|
|
|
|
def default(self):
|
|
return self._default
|
|
|
|
def sections(self):
|
|
return self._sections
|
|
|
|
def section(self, section, index=-1):
|
|
"""Retrieves a section dictionary for the given section. If the section
|
|
holds only a single section dictionary return that dictionary. If
|
|
the section holds more than one dictionary and an index is given
|
|
that is greater than or equal to zero then return the dictionary at
|
|
the index. Otherwise return the list of dictionaries for the given
|
|
section name."""
|
|
try:
|
|
return get_value(self._sections, section, index)
|
|
except KeyError:
|
|
raise LookupError("section %r not found" % section)
|
|
|
|
def add_section(self, section, defaults=[]):
|
|
"""Adds a section with the given name and defaults."""
|
|
self._sections[section] = res = Section(defaults)
|
|
return res
|
|
|
|
def get(self, key, section=DEFAULTSECT, index=-1):
|
|
"""Retrieves a value for the given key from the given section. If the
|
|
key only holds a single value return that value. If the key holds
|
|
more than one value and an index is given that is greater than or
|
|
equal to zero then return the value at the index. Otherwise return
|
|
the list of values."""
|
|
try:
|
|
if section == DEFAULTSECT:
|
|
return get_value(self._default, key, index)
|
|
|
|
# search section(s)
|
|
return find_value(self.section(section), key, index)
|
|
except KeyError:
|
|
# check default section if we haven't already
|
|
if section != DEFAULTSECT:
|
|
return self.get(key, DEFAULTSECT, index)
|
|
raise LookupError("key %r not found in section %r"
|
|
% (key, section))
|
|
|
|
def set(self, key, val, section=DEFAULTSECT):
|
|
"""Sets an option in the given section."""
|
|
if section == DEFAULTSECT:
|
|
self._default[key] = val
|
|
else:
|
|
# for now only set value in first section
|
|
self.section(section, 0)[key] = val
|
|
|
|
def read(self, filename):
|
|
try:
|
|
with open(filename, 'rt') as file:
|
|
self._read(file, filename)
|
|
except IOError:
|
|
print "Could not open file ", filename, " for reading"
|
|
|
|
def _read(self, file, filename):
|
|
for line in file:
|
|
line = remove_comment(line)
|
|
if not line:
|
|
# line was empty or was a comment
|
|
continue
|
|
|
|
section = try_section(line)
|
|
if section:
|
|
if section == DEFAULTSECT:
|
|
sect = self._default
|
|
else:
|
|
self._sections[section] = sect = Section([self._default])
|
|
# TODO - if section has templates add those
|
|
# with sect.add_default
|
|
continue
|
|
|
|
key, val = try_option(line)
|
|
sect[key] = val
|
|
|
|
def write(self, filename):
|
|
try:
|
|
with open(filename, 'wt') as file:
|
|
self._write(file)
|
|
except IOError:
|
|
print "Could not open file ", filename, " for writing"
|
|
pass
|
|
|
|
def _write(self, file):
|
|
# TODO - need to write out default section, but right now in
|
|
# our case res_sip.conf has not default/general section
|
|
for section, sect_list in self._sections.iteritems():
|
|
# every section contains a list of dictionaries
|
|
for sect in sect_list:
|
|
file.write("[%s]\n" % section)
|
|
for key, val_list in sect.iteritems():
|
|
# every value is also a list
|
|
for v in val_list:
|
|
key_val = key
|
|
if (v is not None):
|
|
key_val += " = " + str(v)
|
|
file.write("%s\n" % (key_val))
|
|
file.write("\n")
|