KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run lisa automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
