How to use can_run method in lisa

Best Python code snippet using lisa_python

_game_skill.py

Source:_game_skill.py

...23    def target(self, source, target):24        """target returns if skill target is 'self' or 'target'.25        """26        return source if self._target == "self" else target27    def can_run(self, source):28        """can_run checks if the skill is available to be executed.29        """30        colors = source.counter_colors_available[self.color_str]31        return colors > self.threshold32    def start_match(self, source=None):33        """start_match sets up the skill to be used in a match.34        """35        pass36    def end_match(self, source=None):37        """end_match cleans up anything related with the skill at the end of38        the match.39        """40        pass41    def action(self, source, target):42        """can_run checks if the skill is available to be executed.43        """44        pass45    def clean_up(self, source, target):46        """clean_up proceeds to reverse any action triggered by the skill.47        """48        pass49    def clean_up_event(self, source, target, **kwargs):50        """cleaun_up_event sends the clean up event to the handler.51        """52        GEvent.handler_event(53            GEvent.SKILL, action=self.clean_up(source, target), args=(), **kwargs54        )55class GameSkillBlowColor(GameSkill):56    def __init__(self, color, blow_color, **kwargs):57        super(GameSkillBlowColor, self).__init__("blow-color", color, 10, **kwargs)58        self.blow_color = blow_color59    def action(self, source, target):60        """can_run checks if the skill is available to be executed.61        """62        if self.can_run(source):63            source.counter_colors_available[self.color_str] -= self.threshold64            GEvent.board_event(65                GEvent.SKILL, action="blow-color", args=(self.blow_color,)66            )67class GameSkillBlowEmpty(GameSkill):68    def __init__(self, color, **kwargs):69        super(GameSkillBlowEmpty, self).__init__("blow-color", color, 250, **kwargs)70    def action(self, source, target):71        """can_run checks if the skill is available to be executed.72        """73        if self.can_run(source):74            source.counter_colors_available[self.color_str] -= self.threshold75            GEvent.board_event(GEvent.SKILL, action="blow-empty", args=())76class GameSkillCopyColor(GameSkill):77    def __init__(self, color, from_color, to_color, **kwargs):78        super(GameSkillCopyColor, self).__init__("copy-color", color, 10, **kwargs)79        self.from_color = from_color80        self.to_color = to_color81    def action(self, source, target):82        """can_run checks if the skill is available to be executed.83        """84        if self.can_run(source):85            source.counter_colors_available[self.color_str] -= self.threshold86            GEvent.board_event(87                GEvent.SKILL, action="copy-color", args=(self.from_color, self.to_color)88            )89class GameSkillColorToEmpty(GameSkill):90    def __init__(self, color, from_color, **kwargs):91        super(GameSkillColorToEmpty, self).__init__(92            "color-to-empty", color, 10, **kwargs93        )94        self.from_color = from_color95    def action(self, source, target):96        """can_run checks if the skill is available to be executed.97        """98        if self.can_run(source):99            source.counter_colors_available[self.color_str] -= self.threshold100            GEvent.board_event(101                GEvent.SKILL, action="color-to-empty", args=(self.from_color,)102            )103class GameSkillHeal(GameSkill):104    def __init__(self, color, **kwargs):105        super(GameSkillHeal, self).__init__("heal", color, 25, **kwargs)106        self.heal_value = kwargs.get("heal", 10)107        # self.image = pygame.image.load(os.path.join("apps/sirtet/images", "heal.jpg"))108        self.image = load_image("apps/sirtet/images", "heal.jpg")109    def action(self, source, target):110        """can_run checks if the skill is available to be executed.111        """112        if self.can_run(source):113            print(f"heal {target} fpr {self.heal_value}")114            target.health += self.heal_value115            source.counter_colors_available[self.color_str] -= self.threshold116class GameSkillGreatHeal(GameSkillHeal):117    def __init__(self, color, **kwargs):118        super(GameSkillGreatHeal, self).__init__(color, heal=50, **kwargs)119        self.name = "great-heal"120        self.threshold = 100121        self.image = load_image("apps/sirtet/images", "great_heal.jpg")122class GameSkillMegaHeal(GameSkill):123    def __init__(self, color, **kwargs):124        super(GameSkillMegaHeal, self).__init__(color, heal=500, **kwargs)125        self.name = "mega-heal"126        self.threshold = 250127        self.image = load_image("apps/sirtet/images", "mega_heal.jpg")128class GameSkillDefenseUp(GameSkill):129    def __init__(self, color, **kwargs):130        super(GameSkillDefenseUp, self).__init__(131            "defense-up", color, 50, expire=5, **kwargs132        )133        self.defense_value = kwargs.get("defense", 1)134        self.image = load_image("apps/sirtet/images", "defense_up.jpg")135    def action(self, source, target):136        """can_run checks if the skill is available to be executed.137        """138        if self.can_run(source):139            target._defense.add_buff(self.defense_value)140            source.counter_colors_available[self.color_str] -= self.threshold141            self.clean_up_event(source, target, expire={"lines": self.expire})142    def clean_up(self, source, target):143        """clean_up proceeds to reverse any action triggered by the skill.144        """145        def _clean_up():146            target._defense.del_buff(self.defense_value)147        return _clean_up148class GameSkillDamageUp(GameSkill):149    def __init__(self, color, **kwargs):150        super(GameSkillDamageUp, self).__init__(151            "damage-up", color, 100, expire=5, **kwargs152        )153        self.damage_value = kwargs.get("damage", 1)154        self.image = load_image("apps/sirtet/images", "damage_up.jpg")155    def action(self, source, target):156        """can_run checks if the skill is available to be executed.157        """158        if self.can_run(source):159            target._damage.add_buff(self.damage_value)160            source.counter_colors_available[self.color_str] -= self.threshold161            self.clean_up_event(source, target, expire={"lines": self.expire})162    def clean_up(self, source, target):163        """clean_up proceeds to reverse any action triggered by the skill.164        """165        def _clean_up():166            target._damage.del_buff(self.damage_value)167        return _clean_up168class GameSkillMindUp(GameSkill):169    def __init__(self, color, **kwargs):170        super(GameSkillMindUp, self).__init__(171            "skill-up", color, 100, expire=5, **kwargs172        )173        self.mind_value = kwargs.get("mind", 1)174    def action(self, source, target):175        """can_run checks if the skill is available to be executed.176        """177        if self.can_run(source):178            target._skill.add_buff(self.mind_value)179            source.counter_colors_available[self.color_str] -= self.threshold180            self.clean_up_event(source, target, expire={"lines": self.expire})181    def clean_up(self, source, target):182        """clean_up proceeds to reverse any action triggered by the skill.183        """184        def _clean_up():185            target._skill.del_buff(self.mind_value)186        return _clean_up187class GameSkillDamageBuffUp(GameSkill):188    def __init__(self, color, **kwargs):189        super(GameSkillDamageBuffUp, self).__init__(190            "damage-buff-up", color, 50, expire=5, **kwargs191        )192        self.damage_value = kwargs.get("damage", 1)193    def action(self, source, target):194        """can_run checks if the skill is available to be executed.195        """196        if self.can_run(source):197            target.damage_buffs.append(self.damage_value)198            source.counter_colors_available[self.color_str] -= self.threshold199            self.clean_up_event(source, target, expire={"lines": self.expire})200    def clean_up(self, source, target):201        """clean_up proceeds to reverse any action triggered by the skill.202        """203        def _clean_up():204            target.damage_buffs.remove(self.damage_value)205        return _clean_up206class GameSkillDefenseBuffUp(GameSkill):207    def __init__(self, color, **kwargs):208        super(GameSkillDefenseBuffUp, self).__init__(209            "defense-buff-up", color, 50, expire=5, **kwargs210        )211        self.defense_value = kwargs.get("defense", 1)212    def action(self, source, target):213        """can_run checks if the skill is available to be executed.214        """215        if self.can_run(source):216            target.defense_buffs.append(self.defense_value)217            source.counter_colors_available[self.color_str] -= self.threshold218            self.clean_up_event(source, target, expire={"lines": self.expire})219    def clean_up(self, source, target):220        """clean_up proceeds to reverse any action triggered by the skill.221        """222        def _clean_up():223            target.defense_buffs.remove(self.defense_value)224        return _clean_up225class GameSkillMindBuffUp(GameSkill):226    def __init__(self, color, **kwargs):227        super(GameSkillMindBuffUp, self).__init__(228            "skill-buff-up", color, 50, expire=5, **kwargs229        )230        self.mind_value = kwargs.get("mind", 1)231    def action(self, source, target):232        """can_run checks if the skill is available to be executed.233        """234        if self.can_run(source):235            target.mind_buffs.append(self.mind_value)236            source.counter_colors_available[self.color_str] -= self.threshold237            self.clean_up_event(source, target, expire={"lines": self.expire})238    def clean_up(self, source, target):239        """clean_up proceeds to reverse any action triggered by the skill.240        """241        def _clean_up():242            target.mind_buffs.remove(self.mind_value)243        return _clean_up244class GameSkillRawDamage(GameSkill):245    def __init__(self, color, **kwargs):246        super(GameSkillRawDamage, self).__init__("raw-damage", color, 25, **kwargs)247        self._target = "target"248        self.damage_value = kwargs.get("damage", 10)249        # self.image = pygame.image.load(250        #     os.path.join("apps/sirtet/images", "raw_damage.jpg")251        # )252        self.image = load_image("apps/sirtet/images", "raw_damage.jpg")253    def action(self, source, target):254        """can_run checks if the skill is available to be executed.255        """256        if self.can_run(source):257            target.health -= self.damage_value...

TaskManager.py

Source:TaskManager.py

1# This module provides a simple task manager for running parallel2# calculations on shared-memory machines.3#4# Written by Konrad Hinsen <hinsen@cnrs-orleans.fr>5# last revision: 2006-6-126#7"""8Parallel task manager for shared-memory multiprocessor machines9@undocumented: Task10"""11import threading12class TaskManager:13    """14    Parallel task manager for shared-memory multiprocessor machines15    This class provides a rather simple way to profit from16    shared-memory multiprocessor machines by running several tasks17    in parallel. The calling program decides how many execution threads18    should run at any given time, and then feeds compute tasks to19    the task manager, who runs them as soon as possible without exceeding20    the maximum number of threads.21    The major limitation of this approach lies in Python's Global22    Interpreter Lock. This effectively means that no more than one23    Python thread can run at the same time. Consequently, parallelization24    can only be achieved if the tasks to be parallelized spend25    significant time in C extension modules that release the Global26    Interpreter Lock.27    """28    def __init__(self, nthreads):29        """30        @param nthreads: the maximum number of compute threads that should31                         run in parallel. Note: This does not include the32                         main thread which generated and feeds the task33                         manager!34        @type nthreads: C{int}35        """36        self.nthreads = nthreads37        self.waiting_tasks = []38        self.running_tasks = []39        self.lock = threading.Lock()40        self.data_lock = threading.Lock()41        self.can_run = threading.Condition(self.lock)42        self.can_submit = threading.Condition(self.lock)43        self.task_available = threading.Condition(self.lock)44        self.scheduler = threading.Thread(target=self._scheduler)45        self.scheduler.start()46    def runTask(self, function, args):47        """48        Run a task as soon as processing capacity becomes available49        @param function: the function that will be executed as the body of50                         the task51        @type function: callable52        @param args: the arguments that will be passed to function when it53                     is called. An additional argument will be added at the54                     end: a lock object that the task can use to get55                     temporarily exclusive access to data shared with other56                     tasks.57        @type args: C{tuple}58        """59        self.can_submit.acquire()60        if len(self.waiting_tasks) >= self.nthreads:61            self.can_submit.wait()62        self.can_submit.release()63        task = Task(self, function, args + (self.data_lock,))64        self.task_available.acquire()65        self.waiting_tasks.append(task)66        self.task_available.notify()67        self.task_available.release()68    def terminate(self):69        """70        Wait until all tasks have finished71        """72        self.task_available.acquire()73        self.waiting_tasks.append(None)74        self.task_available.notify()75        self.task_available.release()76        self.scheduler.join()77        done = 078        while not done:79            self.can_run.acquire()80            if self.running_tasks:81                self.can_run.wait()82            done = len(self.running_tasks) == 083            self.can_run.release()84    def _removeTask(self, task):85        self.can_run.acquire()86        self.running_tasks.remove(task)87        self.can_run.notifyAll()88        self.can_run.release()89        90    def _scheduler(self):91        while 1:92            self.task_available.acquire()93            if not self.waiting_tasks:94                self.task_available.wait()95            self.task_available.release()96            self.can_run.acquire()97            while len(self.running_tasks) >= self.nthreads:98                self.can_run.wait()99            task = self.waiting_tasks[0]100            del self.waiting_tasks[0]101            if task is not None:102                self.running_tasks.append(task)103                task.start()104            self.can_submit.notify()105            self.can_run.release()106            if task is None:107                break108class Task(threading.Thread):109    def __init__(self, manager, function, args):110        self.__task_manager = manager111        self.__function = function112        self.__args = args113        threading.Thread.__init__(self)114    def run(self):115        apply(self.__function, self.__args)116        self.__task_manager._removeTask(self)117# Test code118if __name__ == '__main__':119    import time120    from random import randint121    def dummy(n, results, lock):122        print n, "running"123        time.sleep(randint(1, 5))124        lock.acquire()125        results.append(n)126        lock.release()127        print n, "finished"128    m = TaskManager(2)129    results = []130    for i in range(5):131        m.runTask(dummy, (i, results))132    m.terminate()133    print "All finished"...

_game_skill_mind.py

Source:_game_skill_mind.py

...8        self.expire = kwargs.get("expire", 5)9    def action(self, source, target):10        """can_run checks if the skill is available to be executed.11        """12        if self.can_run(source):13            target._skill.add_buff(self.mind_value)14            source.counter_colors_available[self.color_str] -= self.threshold15            self.clean_up_event(source, target, expire={"lines": self.expire})16    def clean_up(self, source, target):17        """clean_up proceeds to reverse any action triggered by the skill.18        """19        def _clean_up():20            target._skill.del_buff(self.mind_value)21        return _clean_up22class GameSkillMindBuffUp(GameSkill):23    def __init__(self, **kwargs):24        super(GameSkillMindBuffUp, self).__init__(25            "skill-buff-up", MIND_COLOR, 50, **kwargs26        )27        self.mind_value = kwargs.get("mind", 1)28        self.expire = kwargs.get("expire", 5)29    def action(self, source, target):30        """can_run checks if the skill is available to be executed.31        """32        if self.can_run(source):33            target.mind_buffs.append(self.mind_value)34            source.counter_colors_available[self.color_str] -= self.threshold35            self.clean_up_event(source, target, expire={"lines": self.expire})36    def clean_up(self, source, target):37        """clean_up proceeds to reverse any action triggered by the skill.38        """39        def _clean_up():40            target.mind_buffs.remove(self.mind_value)41        return _clean_up42class GameSkillBlowColor(GameSkill):43    def __init__(self, blow_color, **kwargs):44        super(GameSkillBlowColor, self).__init__("blow-color", MIND_COLOR, 10, **kwargs)45        self.blow_color = blow_color46    def action(self, source, target):47        """can_run checks if the skill is available to be executed.48        """49        if self.can_run(source):50            source.counter_colors_available[self.color_str] -= self.threshold51            GEvent.board_event(52                GEvent.SKILL, action="blow-color", args=(self.blow_color,)53            )54class GameSkillBlowEmpty(GameSkill):55    def __init__(self, **kwargs):56        super(GameSkillBlowEmpty, self).__init__("blow-color", MIND_COLOR, 10, **kwargs)57    def action(self, source, target):58        """can_run checks if the skill is available to be executed.59        """60        if self.can_run(source):61            source.counter_colors_available[self.color_str] -= self.threshold62            GEvent.board_event(GEvent.SKILL, action="blow-empty", args=())63class GameSkillCopyColor(GameSkill):64    def __init__(self, from_color, to_color, **kwargs):65        super(GameSkillCopyColor, self).__init__("copy-color", MIND_COLOR, 10, **kwargs)66        self.from_color = from_color67        self.to_color = to_color68    def action(self, source, target):69        """can_run checks if the skill is available to be executed.70        """71        if self.can_run(source):72            source.counter_colors_available[self.color_str] -= self.threshold73            GEvent.board_event(74                GEvent.SKILL, action="copy-color", args=(self.from_color, self.to_color)75            )76class GameSkillColorToEmpty(GameSkill):77    def __init__(self, from_color, **kwargs):78        super(GameSkillColorToEmpty, self).__init__(79            "color-to-empty", MIND_COLOR, 10, **kwargs80        )81        self.from_color = from_color82    def action(self, source, target):83        """can_run checks if the skill is available to be executed.84        """85        if self.can_run(source):86            source.counter_colors_available[self.color_str] -= self.threshold87            GEvent.board_event(88                GEvent.SKILL, action="color-to-empty", args=(self.from_color,)89            )90class GameSkillHeal(GameSkill):91    def __init__(self, **kwargs):92        super(GameSkillHeal, self).__init__("heal", MIND_COLOR, 25, **kwargs)93        self.heal_value = kwargs.get("heal", 10)94    def action(self, source, target):95        """can_run checks if the skill is available to be executed.96        """97        if self.can_run(source):98            print(f"heal {target} fpr {self.heal_value}")99            target.health += self.heal_value100            source.counter_colors_available[self.color_str] -= self.threshold101class GameSkillGreatHeal(GameSkillHeal):102    def __init__(self, **kwargs):103        super(GameSkillGreatHeal, self).__init__(heal=50, **kwargs)104        self.name = "great-heal"105        self.threshold = 100106class GameSkillMegaHeal(GameSkill):107    def __init__(self, **kwargs):108        super(GameSkillMegaHeal, self).__init__(heal=500, **kwargs)109        self.name = "mega-heal"...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.