<commit_before>#!/usr/local/bin/python
#
# Copyright (c) 2009 Sippy Software, Inc. All rights reserved.
#
# This file is part of SIPPY, a free RFC3261 SIP stack and B2BUA.
#
# SIPPY is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# For a license to use the SIPPY software under conditions
# other than those described here, or to purchase support for this
# software, please contact Sippy Software, Inc. by e-mail at the
# following addresses: <EMAIL>.
#
# SIPPY is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
#
# $Id: MyConfigParser.py,v 1.3 2009/11/19 20:52:20 sobomax Exp $

from ConfigParser import RawConfigParser
from SipConf import SipConf

SUPPORTED_OPTIONS = { \
 'acct_enable':       ('B', 'enable or disable Radius accounting'), \
 'precise_acct':      ('B', 'do Radius accounting with millisecond precision'), \
 'alive_acct_int':    ('I', 'interval for sending alive Radius accounting in ' \
                             'second (0 to disable alive accounting)'), \
 'config':            ('S', 'load configuration from file (path to file)'), \
 'auth_enable':       ('B', 'enable or disable Radius authentication'), \
 'b2bua_socket':      ('S', 'path to the B2BUA command socket or address to listen ' \
                             'for commands in the format "udp:host[:port]"'), \
 'digest_auth':       ('B', 'enable or disable SIP Digest authentication of ' \
                             'incoming INVITE requests'), \
 'foreground':        ('B', 'run in foreground'), \
 'hide_call_id':      ('B', 'do not pass Call-ID header value from ingress call ' \
                             'leg to egress call leg'), \
 'keepalive_ans':     ('I', 'send periodic "keep-alive" re-INVITE requests on ' \
                             'answering (ingress) call leg and disconnect a call ' \
                             'if the re-INVITE fails (period in seconds, 0 to ' \
                             'disable)'), \
 'keepalive_orig':    ('I', 'send periodic "keep-alive" re-INVITE requests on ' \
                             'originating (egress) call leg and disconnect a call ' \
                             'if the re-INVITE fails (period in seconds, 0 to ' \
                             'disable)'), \
 'logfile':           ('S', 'path to the B2BUA log file'), \
 'max_credit_time':   ('I', 'upper limit of session time for all calls in ' \
                             'seconds'), \
 'max_radiusclients': ('I', 'maximum number of Radius Client helper ' \
                             'processes to start'), \
 'pidfile':           ('S', 'path to the B2BUA PID file'), \
 'radiusclient.conf': ('S', 'path to the radiusclient.conf file'), \
 'sip_address':       ('S', 'local SIP address to listen for incoming SIP requests ' \
                             '("*", "0.0.0.0" or "::" to listen on all IPv4 ' \
                             'or IPv6 interfaces)'),
 'sip_port':          ('I', 'local UDP port to listen for incoming SIP requests'), \
 'start_acct_enable': ('B', 'enable start Radius accounting'), \
 'static_route':      ('S', 'static route for all SIP calls'), \
 'static_tr_in':      ('S', 'translation rule (regexp) to apply to all incoming ' \
                             '(ingress) destination numbers'), \
 'static_tr_out':     ('S', 'translation rule (regexp) to apply to all outgoing ' \
                             '(egress) destination numbers'), \
 'allowed_pts':       ('S', 'list of allowed media (RTP) IANA-assigned payload ' \
                             'types that the B2BUA will pass from input to ' \
                             'output, payload types not in this list will be ' \
                             'filtered out (comma separated list)'), \
 'pass_headers':      ('S', 'list of SIP header field names that the B2BUA will ' \
                             'pass from ingress call leg to egress call leg ' \
                             'unmodified (comma-separated list)'), \
 'accept_ips':        ('S', 'list of IP addresses that we will try to ' \
                             'authenticate INVITE from without sending ' \
                             'SIP Digest challenge (comma-separated list)'),
 'rtp_proxy_clients': ('S', 'comma-separated list of paths or addresses of the ' \
                             'RTPproxy control socket. Address in the format ' \
                             '"udp:host[:port]" (comma-separated list)')}

class MyConfigParser(RawConfigParser):
    default_section = None
    _private_keys = None

    def __init__(self, default_section = 'general'):
        self.default_section = default_section
        self._private_keys = {}
        RawConfigParser.__init__(self)
        self.add_section(self.default_section)

    def __getitem__(self, key):
        if key.startswith('_'):
            return self._private_keys[key]
        value_type  = SUPPORTED_OPTIONS[key][0]
        if value_type  == 'B':
            return self.getboolean(self.default_section, key)
        elif value_type == 'I':
            return self.getint(self.default_section, key)
        return self.get(self.default_section, key)

    def __setitem__(self, key, value):
        if key.startswith('_'):
            self._private_keys[key] = value
        else:
            self.set(self.default_section, key, str(value))
        return

    def has_key(self, key):
        return self.__contains__(key)

    def __contains__(self, key):
        if key.startswith('_'):
            return self._private_keys.has_key(key)
        return self.has_option(self.default_section, key)

    def get(self, *args):
        if len(args) == 1:
            return self.__getitem__(args[0])
        return RawConfigParser.get(self, *args)

    def getdefault(self, key, default_value):
        if self.__contains__(key):
            return self.__getitem__(key)
        return default_value

    def get_longopts(self):
        return tuple([x + '=' for x in SUPPORTED_OPTIONS.keys()])

    def read(self, fname):
        RawConfigParser.readfp(self, open(fname))
        for key in tuple(self.options(self.default_section)):
            self.check_and_set(key, RawConfigParser.get(self, \
              self.default_section, key), False)

    def check_and_set(self, key, value, compat = True):
        value = value.strip()
        if compat:
            if key == 'rtp_proxy_client':
                # XXX compatibility option
                if self.has_key('_rtp_proxy_clients'):
                    self['_rtp_proxy_clients'].append(value)
                else:
                    self['_rtp_proxy_clients'] = [value,]
                if self.has_key('rtp_proxy_clients'):
                    self['rtp_proxy_clients'] += ',' + value
                else:
                    self['rtp_proxy_clients'] = value
                return
            elif key == 'pass_header':
                # XXX compatibility option
                if self.has_key('_pass_headers'):
                    self['_pass_headers'].append(value)
                else:
                    self['_pass_headers'] = [value,]
                if self.has_key('pass_headers'):
                    self['pass_headers'] += ',' + value
                else:
                    self['pass_headers'] = value
                return

        value_type  = SUPPORTED_OPTIONS[key][0]
        if value_type == 'B':
            if value.lower() not in self._boolean_states:
                raise ValueError, 'Not a boolean: %s' % value
        elif value_type == 'I':
            _value = int(value)
        if key in ('keepalive_ans', 'keepalive_orig'):
            if _value < 0:
                raise ValueError, 'keepalive_ans should be non-negative'
        elif key == 'max_credit_time':
            if _value <= 0:
                raise ValueError, 'max_credit_time should be more than zero'
        elif key == 'allowed_pts':
            self['_allowed_pts'] = [int(x) for x in value.split(',')]
        elif key in ('accept_ips', 'pass_headers', 'rtp_proxy_clients'):
            self['_' + key] = [x.strip() for x in value.split(',')]
        elif key == 'sip_address':
            if 'my' in dir(value):
                self['_sip_address'] = value
                value = '*'
            elif value in ('*', '0.0.0.0', '::'):
                self['_sip_address'] = SipConf.my_address
        elif key == 'sip_port':
            if _value <= 0 or _value > 65535:
                raise ValueError, 'sip_port should be in the range 1-65535'
            self['_sip_port'] = _value
        self[key] = value

    def options_help(self):
        supported_options = SUPPORTED_OPTIONS.items()
        supported_options.sort()
        for option, (value_type, helptext) in supported_options:
            if value_type == 'B':
                value = 'on/off'
            elif value_type == 'I':
                value = 'number'
            else:
                value = '"string"'
            print '--%s=%s\n\t%s\n' % (option, value, helptext)

if __name__ == '__main__':
    m = MyConfigParser()
    m['_foo'] = 'bar'
    m['b2bua_socket'] = 'bar1'
    m['acct_enable'] = True
    m['auth_enable'] = 'False'
    assert m.has_key('_foo')
    assert m['_foo'] == 'bar'<commit_msg>Properly set _sip_address.<commit_after>#!/usr/local/bin/python
#
# Copyright (c) 2009 Sippy Software, Inc. All rights reserved.
#
# This file is part of SIPPY, a free RFC3261 SIP stack and B2BUA.
#
# SIPPY is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# For a license to use the SIPPY software under conditions
# other than those described here, or to purchase support for this
# software, please contact Sippy Software, Inc. by e-mail at the
# following addresses: <EMAIL>.
#
# SIPPY is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA.
#
# $Id: MyConfigParser.py,v 1.4 2009/11/20 07:28:00 sobomax Exp $

from ConfigParser import RawConfigParser
from SipConf import SipConf

SUPPORTED_OPTIONS = { \
 'acct_enable':       ('B', 'enable or disable Radius accounting'), \
 'precise_acct':      ('B', 'do Radius accounting with millisecond precision'), \
 'alive_acct_int':    ('I', 'interval for sending alive Radius accounting in ' \
                             'second (0 to disable alive accounting)'), \
 'config':            ('S', 'load configuration from file (path to file)'), \
 'auth_enable':       ('B', 'enable or disable Radius authentication'), \
 'b2bua_socket':      ('S', 'path to the B2BUA command socket or address to listen ' \
                             'for commands in the format "udp:host[:port]"'), \
 'digest_auth':       ('B', 'enable or disable SIP Digest authentication of ' \
                             'incoming INVITE requests'), \
 'foreground':        ('B', 'run in foreground'), \
 'hide_call_id':      ('B', 'do not pass Call-ID header value from ingress call ' \
                             'leg to egress call leg'), \
 'keepalive_ans':     ('I', 'send periodic "keep-alive" re-INVITE requests on ' \
                             'answering (ingress) call leg and disconnect a call ' \
                             'if the re-INVITE fails (period in seconds, 0 to ' \
                             'disable)'), \
 'keepalive_orig':    ('I', 'send periodic "keep-alive" re-INVITE requests on ' \
                             'originating (egress) call leg and disconnect a call ' \
                             'if the re-INVITE fails (period in seconds, 0 to ' \
                             'disable)'), \
 'logfile':           ('S', 'path to the B2BUA log file'), \
 'max_credit_time':   ('I', 'upper limit of session time for all calls in ' \
                             'seconds'), \
 'max_radiusclients': ('I', 'maximum number of Radius Client helper ' \
                             'processes to start'), \
 'pidfile':           ('S', 'path to the B2BUA PID file'), \
 'radiusclient.conf': ('S', 'path to the radiusclient.conf file'), \
 'sip_address':       ('S', 'local SIP address to listen for incoming SIP requests ' \
                             '("*", "0.0.0.0" or "::" to listen on all IPv4 ' \
                             'or IPv6 interfaces)'),
 'sip_port':          ('I', 'local UDP port to listen for incoming SIP requests'), \
 'start_acct_enable': ('B', 'enable start Radius accounting'), \
 'static_route':      ('S', 'static route for all SIP calls'), \
 'static_tr_in':      ('S', 'translation rule (regexp) to apply to all incoming ' \
                             '(ingress) destination numbers'), \
 'static_tr_out':     ('S', 'translation rule (regexp) to apply to all outgoing ' \
                             '(egress) destination numbers'), \
 'allowed_pts':       ('S', 'list of allowed media (RTP) IANA-assigned payload ' \
                             'types that the B2BUA will pass from input to ' \
                             'output, payload types not in this list will be ' \
                             'filtered out (comma separated list)'), \
 'pass_headers':      ('S', 'list of SIP header field names that the B2BUA will ' \
                             'pass from ingress call leg to egress call leg ' \
                             'unmodified (comma-separated list)'), \
 'accept_ips':        ('S', 'list of IP addresses that we will try to ' \
                             'authenticate INVITE from without sending ' \
                             'SIP Digest challenge (comma-separated list)'),
 'rtp_proxy_clients': ('S', 'comma-separated list of paths or addresses of the ' \
                             'RTPproxy control socket. Address in the format ' \
                             '"udp:host[:port]" (comma-separated list)')}

class MyConfigParser(RawConfigParser):
    default_section = None
    _private_keys = None

    def __init__(self, default_section = 'general'):
        self.default_section = default_section
        self._private_keys = {}
        RawConfigParser.__init__(self)
        self.add_section(self.default_section)

    def __getitem__(self, key):
        if key.startswith('_'):
            return self._private_keys[key]
        value_type  = SUPPORTED_OPTIONS[key][0]
        if value_type  == 'B':
            return self.getboolean(self.default_section, key)
        elif value_type == 'I':
            return self.getint(self.default_section, key)
        return self.get(self.default_section, key)

    def __setitem__(self, key, value):
        if key.startswith('_'):
            self._private_keys[key] = value
        else:
            self.set(self.default_section, key, str(value))
        return

    def has_key(self, key):
        return self.__contains__(key)

    def __contains__(self, key):
        if key.startswith('_'):
            return self._private_keys.has_key(key)
        return self.has_option(self.default_section, key)

    def get(self, *args):
        if len(args) == 1:
            return self.__getitem__(args[0])
        return RawConfigParser.get(self, *args)

    def getdefault(self, key, default_value):
        if self.__contains__(key):
            return self.__getitem__(key)
        return default_value

    def get_longopts(self):
        return tuple([x + '=' for x in SUPPORTED_OPTIONS.keys()])

    def read(self, fname):
        RawConfigParser.readfp(self, open(fname))
        for key in tuple(self.options(self.default_section)):
            self.check_and_set(key, RawConfigParser.get(self, \
              self.default_section, key), False)

    def check_and_set(self, key, value, compat = True):
        value = value.strip()
        if compat:
            if key == 'rtp_proxy_client':
                # XXX compatibility option
                if self.has_key('_rtp_proxy_clients'):
                    self['_rtp_proxy_clients'].append(value)
                else:
                    self['_rtp_proxy_clients'] = [value,]
                if self.has_key('rtp_proxy_clients'):
                    self['rtp_proxy_clients'] += ',' + value
                else:
                    self['rtp_proxy_clients'] = value
                return
            elif key == 'pass_header':
                # XXX compatibility option
                if self.has_key('_pass_headers'):
                    self['_pass_headers'].append(value)
                else:
                    self['_pass_headers'] = [value,]
                if self.has_key('pass_headers'):
                    self['pass_headers'] += ',' + value
                else:
                    self['pass_headers'] = value
                return

        value_type  = SUPPORTED_OPTIONS[key][0]
        if value_type == 'B':
            if value.lower() not in self._boolean_states:
                raise ValueError, 'Not a boolean: %s' % value
        elif value_type == 'I':
            _value = int(value)
        if key in ('keepalive_ans', 'keepalive_orig'):
            if _value < 0:
                raise ValueError, 'keepalive_ans should be non-negative'
        elif key == 'max_credit_time':
            if _value <= 0:
                raise ValueError, 'max_credit_time should be more than zero'
        elif key == 'allowed_pts':
            self['_allowed_pts'] = [int(x) for x in value.split(',')]
        elif key in ('accept_ips', 'pass_headers', 'rtp_proxy_clients'):
            self['_' + key] = [x.strip() for x in value.split(',')]
        elif key == 'sip_address':
            if 'my' in dir(value):
                self['_sip_address'] = value
                value = '*'
            elif value in ('*', '0.0.0.0', '::'):
                self['_sip_address'] = SipConf.my_address
            else:
                self['_sip_address'] = value
        elif key == 'sip_port':
            if _value <= 0 or _value > 65535:
                raise ValueError, 'sip_port should be in the range 1-65535'
            self['_sip_port'] = _value
        self[key] = value

    def options_help(self):
        supported_options = SUPPORTED_OPTIONS.items()
        supported_options.sort()
        for option, (value_type, helptext) in supported_options:
            if value_type == 'B':
                value = 'on/off'
            elif value_type == 'I':
                value = 'number'
            else:
                value = '"string"'
            print '--%s=%s\n\t%s\n' % (option, value, helptext)

if __name__ == '__main__':
    m = MyConfigParser()
    m['_foo'] = 'bar'
    m['b2bua_socket'] = 'bar1'
    m['acct_enable'] = True
    m['auth_enable'] = 'False'
    assert m.has_key('_foo')
    assert m['_foo'] == 'bar'
{'fraction_non_alphanumeric': 0.11421733374070679, 'fraction_numerical': 0.008045625827477339, 'mean_word_length': 1.967512843759444, 'pattern_counts': {'":': 2, '<': 11, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7511107', 'n_tokens_mistral': 5744, 'n_tokens_neox': 5528, 'n_words': 1734}