From bd1b0b5e3276a77d9eeb08c81bc88c440d0f8738 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Manuel=20Cort=C3=A9z?= Date: Tue, 13 Jan 2015 12:31:37 -0600 Subject: [PATCH] Some stream improvements --- src/controller/buffersController.py | 4 ++-- src/controller/mainController.py | 36 ++++++++++++++++++++++------- src/logger.py | 4 ++-- src/main.py | 1 + src/mysc/thread_utils.py | 9 ++++---- src/sessionmanager/session.py | 33 ++++++++++++++++++++++---- src/twitter/buffers/stream.py | 1 - src/wxUI/dialogs/message.py | 4 ++-- 8 files changed, 67 insertions(+), 25 deletions(-) diff --git a/src/controller/buffersController.py b/src/controller/buffersController.py index 0ee031dc..60bf2839 100644 --- a/src/controller/buffersController.py +++ b/src/controller/buffersController.py @@ -211,9 +211,9 @@ class baseBufferController(bufferController): message = messages.reply(self.session, _(u"Reply"), _(u"Reply to %s") % (screen_name,), "@%s" % (screen_name,), users) if message.message.get_response() == widgetUtils.OK: if message.image == None: - call_threaded(self.session.twitter.api_call, call_name="update_status", _sound="reply_send.ogg", in_reply_to_status_id=id, status=message.message.get_text()) + call_threaded(self.session.api_call, call_name="update_status", _sound="reply_send.ogg", in_reply_to_status_id=id, status=message.message.get_text()) else: - call_threaded(self.session.twitter.api_call, call_name="update_status_with_media", _sound="reply_send.ogg", in_reply_to_status_id=id, status=message.message.get_text(), media=message.file) + call_threaded(self.session.api_call, call_name="update_status_with_media", _sound="reply_send.ogg", in_reply_to_status_id=id, status=message.message.get_text(), media=message.file) def direct_message(self, *args, **kwargs): tweet = self.get_tweet() diff --git a/src/controller/mainController.py b/src/controller/mainController.py index cf1343e2..cb27b3e9 100644 --- a/src/controller/mainController.py +++ b/src/controller/mainController.py @@ -2,7 +2,7 @@ from wxUI import (view, dialogs, commonMessageDialogs) import buffersController import messages -from sessionmanager import session +from sessionmanager import session as session_ from pubsub import pub import sound import output @@ -13,9 +13,12 @@ import config import widgetUtils import platform from extra import SoundsTutorial +import logging if platform.system() == "Windows": import keystrokeEditor +log = logging.getLogger("mainController") + class Controller(object): """ Main Controller for TWBlue. It manages the main window and sessions.""" @@ -65,6 +68,7 @@ class Controller(object): def bind_other_events(self): pub.subscribe(self.editing_keystroke, "editing_keystroke") + pub.subscribe(self.manage_stream_errors, "stream-error") widgetUtils.connect_event(self.view, widgetUtils.MENU, self.search, menuitem=self.view.menuitem_search) widgetUtils.connect_event(self.view, widgetUtils.MENU, self.learn_sounds, menuitem=self.view.sounds_tutorial) widgetUtils.connect_event(self.view, widgetUtils.MENU, self.exit, menuitem=self.view.close) @@ -87,10 +91,14 @@ class Controller(object): self.do_work() def do_work(self): - for i in session.sessions: - self.create_buffers(session.sessions[i]) - call_threaded(self.start_buffers, session.sessions[i]) + log.debug("Creating buffers for all sessions...") + for i in session_.sessions: + log.debug("Working on session %s" % (i,)) + self.create_buffers(session_.sessions[i]) + call_threaded(self.start_buffers, session_.sessions[i]) sound.player.play("tweet_timeline.ogg") + self.checker_function = RepeatingTimer(60, self.check_connection) + self.checker_function.start() def create_buffers(self, session): session.get_user_info() @@ -222,10 +230,10 @@ class Controller(object): buffer.destroy_status() def exit(self, *args, **kwargs): - for item in session.sessions: - session.sessions[item].settings.write() - session.sessions[item].main_stream.disconnect() - session.sessions[item].timelinesStream.disconnect() + for item in session_.sessions: + session_.sessions[item].settings.write() + session_.sessions[item].main_stream.disconnect() + session_.sessions[item].timelinesStream.disconnect() sound.player.cleaner.cancel() widgetUtils.exit_application() @@ -396,10 +404,22 @@ class Controller(object): print "i've pressed" def start_buffers(self, session): + log.debug("starting buffers... Session %s" % (session,)) for i in self.buffers: + log.debug("Starting %s for %s" % (i.name, session)) if i.session == session and i.needs_init == True: i.start_stream() + log.debug("Starting the streaming endpoint") session.start_streaming() + def manage_stream_errors(self, session): + log.error("An error ocurred with the stream for the %s session. It will be destroyed" % (session,)) + s = sessions_.session[session] + s.listen_stream_error() + + def check_connection(self): + for i in session_.sessions: + session_.sessions[i].check_connection() + def __del__(self): config.app.write() \ No newline at end of file diff --git a/src/logger.py b/src/logger.py index 9c4f21c2..f396f337 100644 --- a/src/logger.py +++ b/src/logger.py @@ -6,8 +6,8 @@ import sys APP_LOG_FILE = 'debug.log' ERROR_LOG_FILE = "error.log" -MESSAGE_FORMAT = "%(asctime)s %(name)s %(levelname)s: %(message)s" -DATE_FORMAT = "%a %b %d, %Y %H:%M:%S" +MESSAGE_FORMAT = u"%(asctime)s %(name)s %(levelname)s: %(message)s" +DATE_FORMAT = u"%b %d, %Y %H:%M:%S" formatter = logging.Formatter(MESSAGE_FORMAT, datefmt=DATE_FORMAT) diff --git a/src/main.py b/src/main.py index f472b14a..254a1abb 100644 --- a/src/main.py +++ b/src/main.py @@ -6,6 +6,7 @@ import commandline import config import sound import output +from logger import logger as logging def setup(): config.setup() diff --git a/src/mysc/thread_utils.py b/src/mysc/thread_utils.py index 29aecf4d..9fd3dc25 100644 --- a/src/mysc/thread_utils.py +++ b/src/mysc/thread_utils.py @@ -11,9 +11,7 @@ def call_threaded(func, *args, **kwargs): try: func(*a, **k) except TwythonRateLimitError: - wx.MessageDialog(None, u"Has superado el límite de llamadas a la API que Twitter permite. Eso puede deberse a muchas líneas temporales abiertas, a abrir y cerrar el programa en un periodo corto de tiempo, o a tener muchas llamadas a la API en tu configuración. TW Blue esperará 5 minutos y volverá a intentarlo.", u"Límite de llamadas a la API Superado", wx.ICON_ERROR|wx.OK).ShowModal() - time.sleep(300) - call_threaded(func, *args, **kwargs) + pass # except: # pass thread = threading.Thread(target=new_func, args=args, kwargs=kwargs) @@ -23,10 +21,11 @@ def call_threaded(func, *args, **kwargs): def stream_threaded(func, *args, **kwargs): def new_func(*a, **k): +# global session try: - func(*a, **k) + func(**k) except: - pub.sendMessage("streamError") + pub.sendMessage("streamError", session=a[0]) thread = threading.Thread(target=new_func, args=args, kwargs=kwargs) thread.daemon = True thread.start() diff --git a/src/sessionmanager/session.py b/src/sessionmanager/session.py index a294a8e9..62651cfd 100644 --- a/src/sessionmanager/session.py +++ b/src/sessionmanager/session.py @@ -10,7 +10,7 @@ import sound from twitter import utils from twython import TwythonError, TwythonRateLimitError, TwythonAuthError from config_utils import Configuration, ConfigurationResetException -from mysc.thread_utils import call_threaded +from mysc.thread_utils import stream_threaded sessions = {} @@ -88,6 +88,7 @@ class Session(object): self.settings = None self.twitter = twitter.twitter.twitter() self.db = {} + self.reconnection_function_active = False def get_configuration(self): @@ -102,7 +103,7 @@ class Session(object): @_require_configuration def login(self): - """ Login into twitter using credentials from settings. + """ Login in to twitter using credentials from settings. if the user account isn't authorised, it needs to call self.authorise() before login.""" if self.settings["twitter"]["user_key"] != None and self.settings["twitter"]["user_secret"] != None: @@ -257,12 +258,34 @@ class Session(object): def start_streaming(self): """ Start the streaming for sending tweets in realtime.""" + self.get_main_stream() + self.get_timelines() + + def get_main_stream(self): self.main_stream = twitter.buffers.stream.streamer(application.app_key, application.app_secret, self.settings["twitter"]["user_key"], self.settings["twitter"]["user_secret"], self) - call_threaded(self.main_stream.user) + stream_threaded(self.main_stream.user, self.session_id) + + def get_timelines(self): self.timelinesStream = twitter.buffers.indibidual.timelinesStreamer(application.app_key, application.app_secret, self.settings["twitter"]["user_key"], self.settings["twitter"]["user_secret"], session=self) ids = "" for i in self.settings["other_buffers"]["timelines"]: ids = ids + "%s, " % (self.db[i+"-timeline"][0]["user"]["id_str"]) # if ids != "": - call_threaded(self.timelinesStream.statuses.filter, follow=ids) - \ No newline at end of file + stream_threaded(self.timelinesStream.statuses.filter, self.session_id, follow=ids) + + def listen_stream_error(self): + if hasattr(self, "main_stream"): + self.main_stream.disconnect() + del self.main_stream + if hasattr(self, "timelinesStream"): + self.timelinesStream.disconnect() + del self.timelinesStream + + def check_connection(self): + if self.reconnection_function_active == True: return + self.reconnection_function_active = True + if not hasattr(self, "main_stream"): + self.get_main_stream() + if not hasattr(self, "timelinesStream"): + self.get_timelines() + self.reconnection_function_active = False diff --git a/src/twitter/buffers/stream.py b/src/twitter/buffers/stream.py index ac680c56..fc503870 100644 --- a/src/twitter/buffers/stream.py +++ b/src/twitter/buffers/stream.py @@ -74,7 +74,6 @@ class streamer(TwythonStreamer): def remove_fav(self, data): if self.session.db["user_name"] == data["source"]["screen_name"]: item = utils.find_item(data["target_object"]["id"], self.session.db["favourites"]) - print item self.session.db["favourites"].pop(item) pub.sendMessage("unfavourite", item=item, user=self.get_user()) diff --git a/src/wxUI/dialogs/message.py b/src/wxUI/dialogs/message.py index ca64e528..d68c3f13 100644 --- a/src/wxUI/dialogs/message.py +++ b/src/wxUI/dialogs/message.py @@ -210,7 +210,7 @@ class viewTweet(widgetUtils.BaseDialog): self.SetClientSize(mainBox.CalcMin()) def set_text(self, text): - self.text.ChangeValue() + self.text.ChangeValue(text) def get_text(self): return self.text.GetValue() @@ -264,7 +264,7 @@ class viewNonTweet(widgetUtils.BaseDialog): self.text.SelectAll() def set_text(self, text): - self.text.ChangeValue() + self.text.ChangeValue(text) def get_text(self): return self.text.GetValue()