# Copyright (C) 2003, 2004 Konstantin Korikov
# This program 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.
#
# This program 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
"""Base application class"""
import os
import Queue
import chestnut_dialer
from chestnut_dialer import _
from chestnut_dialer import debug_msg
import chestnut_dialer.connection
from chestnut_dialer.sig import Signal
import chestnut_dialer.share
import chestnut_dialer.dobj as dobj
class Application:
"""Base application class"""
share = None
accounts = None
config = None
_connection = None
current_account = None
_conn_debug_queue = None
_state = 0
_STATE_INIT = 1
_STATE_SETUP = 2
event_monitor_signal = None
connection_monitor_signal = None
def __init__(self):
self.share = chestnut_dialer.share.Share()
self.share.new_session(os.getpid())
self.accounts = self.share.accounts
self.config = self.share.config
self._conn_debug_queue = Queue.Queue(512)
self.event_monitor_signal = Signal((self._event_monitor,))
self.connection_monitor_signal = Signal()
def run(self):
"Run application"
return chestnut_dialer.EX_OK
def exit(self):
pass
def destroy(self):
self.disconnect()
self.share.remove_session(os.getpid())
self.event_monitor_signal = None
self.connection_monitor_signal = None
self.config = None
self.current_account = None
self.accounts = None
self.share = None
def is_connection_init(self):
return self._connection != None and self._connection.is_active()
def is_connection_setup(self):
return self._connection != None and self._connection.is_connected()
def connect(self, account):
self.disconnect()
self._state = 0
self.current_account = account
def_acc_set_to = account
while def_acc_set_to.get_default_account() != None:
def_acc_set_to = def_acc_set_to.get_default_account()
def_acc_set_to.set_default_account(
self.config.default_account)
acc = self.current_account.copy()
if acc['ask_passwd']:
acc['passwd'], cancel = self.ask_password()
if cancel: return
self._connection = chestnut_dialer.connection.Connection(
acc, self._conn_debug_queue,
pppd_cbcp_with_auto_answer = self.config.pppd_cbcp_with_auto_answer,
write_dns_to_resolv_conf = self.config.write_dns_to_resolv_conf)
self._connection.start()
self.connection_monitor_signal.append(self._connection_monitor)
def_acc_set_to.set_default_account(None)
self._connection_monitor()
def disconnect(self):
s = ""
try:
while 1: s += self._conn_debug_queue.get_nowait()
except Queue.Empty: pass
if s: self.on_connection_debug_msg(s)
if self.is_connection_init():
self.connection_monitor_signal.remove(self._connection_monitor)
self._connection.stop()
self.on_connection_stop(
self._connection.get_stop_reson())
self._state = 0
def _connection_monitor(self):
self._connection.iteration()
msg = [""]
try:
for i in range(0, self._conn_debug_queue.qsize()):
msg[0] += self._conn_debug_queue.get_nowait()
except Queue.Empty: pass
def push_msg():
if msg[0]:
self.on_connection_debug_msg(msg[0])
msg[0] = ""
new_state = 0
if self.is_connection_init(): new_state |= self._STATE_INIT
if self.is_connection_setup(): new_state |= self._STATE_SETUP
if (new_state ^ self._state) & self._STATE_INIT:
if new_state & self._STATE_INIT:
self.on_connection_init()
else:
push_msg()
self.connection_monitor_signal.remove(self._connection_monitor)
self.on_connection_stop(
self._connection.get_stop_reson())
self.current_account.set_default_account(None)
if (new_state ^ self._state) & self._STATE_SETUP:
if new_state & self._STATE_SETUP:
self.on_connection_setup()
elif new_state & self._STATE_INIT:
self.on_connection_init()
self._state = new_state
push_msg()
def _event_monitor(self):
accounts_changed_handled = 0
for e in self.share.get_events():
if isinstance(e, dobj.SimpleEvent):
if e.data[0] == "config-attr-changed":
debug_msg(_("Event: config-attr-changed: %s") % e.data[1], 3)
self.on_config_attr_changed(e.data[1], e.data[2])
elif e.data[0] == "accounts-changed" and not accounts_changed_handled:
debug_msg(_("Event: accounts-changed"), 3)
self.on_accounts_changed()
accounts_changed_handled = 1
def get_connection(self):
return self._connection
def register_monitor(self, func, period_sec):
pass
def unregister_monitor(self):
pass
def on_connection_init(self):
pass
def on_connection_setup(self):
pass
def on_connection_stop(self, reason):
pass
def on_connection_debug_msg(self, msg):
pass
def on_config_attr_changed(self, name, value):
pass
def on_accounts_changed(self):
pass
def ask_password(self):
return ("", 0)
def get_window_title(self):
fmt = ""
data = {
"program_name": chestnut_dialer.program_name,
"program_version": chestnut_dialer.program_version}
if self.is_connection_init():
data.update(self._connection.account)
if self.is_connection_setup():
data.update(self._connection.get_info())
fmt = self.config.wnd_title_fmt_connected
else:
fmt = self.config.wnd_title_fmt_connecting
else:
fmt = self.config.wnd_title_fmt_disconnected
try:
return fmt % data
except: debug_msg(_("invalid format string '%s'") % fmt, 1)
return ""
|