scheduler.py

ABC-win32-v3.1 一个P2P软源代码

import sys
import wx

from operator import attrgetter
from threading import Event
from threading import Timer
#from traceback import print_exc
#from cStringIO import StringIO

from ABC.Scheduler.action import ActionHandler
from ABC.Scheduler.addtorrents import AddTorrents
from ABC.Scheduler.ratemanager import RateManager

from Utility.constants import * #IGNORE:W0611


wxEVT_INVOKE = wx.NewEventType()

def EVT_INVOKE(win, func):
    win.Connect(-1, -1, wxEVT_INVOKE, func)
    
def DELEVT_INVOKE(win):
    win.Disconnect(-1, -1, wxEVT_INVOKE)

class InvokeEvent(wx.PyEvent):
    def __init__(self, func, args, kwargs):
        wx.PyEvent.__init__(self)
        self.SetEventType(wxEVT_INVOKE)
        self.func = func
        self.args = args
        self.kwargs = kwargs


################################################################
#
# Class: ABCScheduler
#
# Determine which torrents need to run, update global stats,
# and deal with loading, moving, and removing torrents.
#
################################################################
class ABCScheduler(wx.EvtHandler):
    def __init__(self, utility):
        wx.EvtHandler.__init__(self)
        
        self.utility = utility
        self.utility.queue = self
        
        self.utility.actionhandler = ActionHandler(self.utility)
        self.ratemanager = RateManager(self)
        self.addtorrents = AddTorrents(self)

        self.timers = {}
        
        self.flag = Event()

        self.totals = { 'up' : 0.0, 
                        'down' : 0.0, 
                        'connections': 0 }
        self.totals_kb = { 'up': 0.0,
                           'down': 0.0 }

        self.UpdateRunningTorrentCounters()

        EVT_INVOKE(self, self.onInvoke)
        
    def postInitTasks(self):
        # Read old list from torrent.lst
        ####################################
        self.addtorrents.readTorrentList()

        # Wait until after creating the list and adding torrents
        # to start CyclicalTasks in the scheduler
        self.CyclicalTasks()
        self.InfrequentCyclicalTasks(False)
      
    # Update the counters for torrents in a single unified place
    def CalculateTorrentCounters(self):
        torrents_active = self.utility.torrents["active"].keys()

        paused = {}
        seeding = {}
        downloading = {}
                                                   
        for ABCTorrentTemp in torrents_active:
            # Torrent is active
            if (ABCTorrentTemp.status.value == STATUS_HASHCHECK):
                activevalues = [ STATUS_ACTIVE, STATUS_PAUSE, STATUS_SUPERSEED ]
                # Torrent is doing a hash check
                # (Count towards counters if it was active before the the check,
                #  otherwise don't)
                if not ABCTorrentTemp.actions.oldstatus in activevalues:
                    continue
            
            if ABCTorrentTemp.status.value == STATUS_PAUSE:
                paused[ABCTorrentTemp] = 1
            elif ABCTorrentTemp.status.completed:
                seeding[ABCTorrentTemp] = 1
            else:
                downloading[ABCTorrentTemp] = 1

        self.utility.torrents["pause"] = paused
        self.utility.torrents["seeding"] = seeding
        self.utility.torrents["downloading"] = downloading
    
    def getProcCount(self):
        if self.utility.config.Read('trigwhenfinishseed', "boolean"):
            return len(self.utility.torrents["active"])
        else:
            return len(self.utility.torrents["active"]) - len(self.utility.torrents["seeding"])
        
    def UpdateRunningTorrentCounters(self):
        self.CalculateTorrentCounters()
            
        statusfunc = self.utility.frame.abc_sb.SetStatusText      
        statusfunc((" " + self.utility.lang.get('abbrev_loaded') + " %u " % len(self.utility.torrents["all"])), 1)
        statusfunc((" " + self.utility.lang.get('abbrev_running') + " %u " % len(self.utility.torrents["active"])), 2)
        statusfunc((" " + self.utility.lang.get('abbrev_downloading') + " %u " % len(self.utility.torrents["downloading"])), 3)
        statusfunc((" " + self.utility.lang.get('abbrev_seeding') + " %u " % len(self.utility.torrents["seeding"])), 4)
        statusfunc((" " + self.utility.lang.get('abbrev_pause') + " %u " % len(self.utility.torrents["pause"])), 5)
        
        try:
            if hasattr(self.utility, "bottomline2"):
                self.utility.bottomline2.updateCounters()
        except wx.PyDeadObjectError:
            pass

    def getDownUpConnections(self):
        # Ask UD/DL speed of all threads
        ########################################
        totalupload     = 0.0
        totaldownload   = 0.0
        totalconnections = 0

        for ABCTorrentTemp in self.utility.torrents["active"].keys():
            if ABCTorrentTemp.status.value != STATUS_PAUSE:
                downrate = ABCTorrentTemp.getColumnValue(COL_DLSPEED)
                uprate = ABCTorrentTemp.getColumnValue(COL_ULSPEED)
                
                ABCTorrentTemp.connection.rate["up"] = (uprate / 1024.0)
                ABCTorrentTemp.connection.rate["down"] = (downrate / 1024.0)
                
                totaldownload += downrate
                totalupload += uprate
                totalconnections += ABCTorrentTemp.getColumnValue(COL_CONNECTIONS)
                
        self.totals['up'] = totalupload
        self.totals_kb['up'] = (totalupload / 1024.0)
        
        self.totals['down'] = totaldownload
        self.totals_kb['down'] = (totaldownload / 1024.0)
        
        self.totals['connections'] = totalconnections
        
    def updateTrayAndStatusBar(self):
        maxuprate = self.ratemanager.MaxRate("up")
        if maxuprate == 0:
            upspeed = self.utility.speed_format(self.totals['up'], truncate = 1)
            upratecap = "oo"
        else:
            upspeed = self.utility.size_format(self.totals['up'], truncate = 1, stopearly = "KB", applylabel = False)
            upratecap = self.utility.speed_format((maxuprate * 1024), truncate = 0, stopearly = "KB")
        uploadspeed = upspeed + " / " + upratecap

        maxdownrate = self.ratemanager.MaxRate("down")
        if maxdownrate == 0:
            downspeed = self.utility.speed_format(self.totals['down'], truncate = 1)
            downratecap = "oo"
        else:
            downspeed = self.utility.size_format(self.totals['down'], truncate = 1, stopearly = "KB", applylabel = False)
            downratecap = self.utility.speed_format((maxdownrate * 1024), truncate = 0, stopearly = "KB")
        downloadspeed = downspeed + " / " + downratecap
        
        try:
            # update value in minimize icon
            ###########################################
            if self.utility.frame.tbicon is not None and self.utility.frame.tbicon.IsIconInstalled():
                icontext = "ABC" + "\n\n" + \
                           self.utility.lang.get('totaldlspeed') + " " + downloadspeed + "\n" + \
                           self.utility.lang.get('totalulspeed') + " " + uploadspeed + " "
                self.utility.frame.tbicon.SetIcon(self.utility.icon, icontext)

            # update in status bar
            ##########################################
            if self.utility.frame.abc_sb is not None:
                self.utility.frame.abc_sb.SetStatusText(" " + self.utility.lang.get('abbrev_connections') + " " + str(int(self.totals['connections'])), 6)
                self.utility.frame.abc_sb.SetStatusText(" " + self.utility.lang.get('abbrev_down') + " " + downloadspeed, 7)
                self.utility.frame.abc_sb.SetStatusText(" " + self.utility.lang.get('abbrev_up') + " " + uploadspeed, 8)
        except wx.PyDeadObjectError:
            pass
                                
    def CyclicalTasks(self):       
        self.getDownUpConnections()
            
        self.updateTrayAndStatusBar()

        self.ratemanager.RunTasks()
   
        try:
            # Run postponed deleting events
            while self.utility.window.postponedevents:
                ev = self.utility.window.postponedevents.pop(0)
                #print "POSTPONED EVENT : ", ev[0]
                ev[0](ev[1])
            self.utility.list.Enable()
        except wx.PyDeadObjectError:
            pass

        # Try invoking the scheduler
        # (just in case we need to start more stuff:
        #  should return almost immediately otherwise)
        self.Scheduler()

        # Start Timer
        ##########################################
        self.timers['frequent'] = Timer(2, self.CyclicalTasks)
        self.timers['frequent'].start()
            
    def InfrequentCyclicalTasks(self, update = True):
        if update:           
            try:
                if self.timers['infrequent'] is not None:
                    self.timers['infrequent'].cancel()
            except:
                pass
        
            self.updateTorrentList()

        self.timers['infrequent'] = Timer(300, self.InfrequentCyclicalTasks)
        self.timers['infrequent'].start()

               
    def onInvoke(self, event):
        if not self.flag.isSet():
            event.func(*event.args, **event.kwargs)

    def invokeLater(self, func, args = None, kwargs = None):
        if args is None:
            args = []
        if kwargs is None:
            kwargs = {}