Best Python code snippet using playwright-python
immersivefx.py
Source:immersivefx.py
1import sys2import threading3from time import sleep, time4import numpy as np5from devices import WLED, Serial, DualShock6class Core:7 name = 'ImmersiveFX Core' # override this in your fxmode8 # those need to be set by their respective fxmodes, as a list containing all applying values9 target_versions = None # 'dev' works best for builtin fxmodes, external stuff should name actual versions though10 target_platforms = None # check https://docs.python.org/3/library/sys.html#sys.platform or use 'all' if it applies11 def __init__(self, core_version, config, launch_arguments, *args, **kwargs):12 """13 fancy little base class, make your fxmode inherit from it to spare yourself unnecessary work14 or don't, I'm a comment not a cop.15 """16 self.launch_arguments = launch_arguments17 self.check_target(core_version)18 self.config = config19 self.devices = {}20 self.parse_devices(config)21 self.data_thread = None22 self.device_classes = {23 'wled': WLED,24 'serial': Serial,25 'dualshock': DualShock,26 }27 self.raw_data = np.zeros([1, 3]) # empty, but valid staring point28 self.frame_sleep = 1000 / self.config.get('fps', 30)29 self.splash()30 def start_threads(self):31 """32 initializes and starts data and device threads, call this in your fxmode, usually at the end33 """34 self.data_thread = threading.Thread(35 target=self.data_loop,36 args=(),37 kwargs={},38 )39 if self.launch_arguments.single_threaded:40 while True:41 start = time()42 self.data_loop()43 for device, device_config in self.devices.items():44 self.device_loop(device)45 duration = (time() - start) * 100046 print(f'Cycle took {round(duration, 2)}ms')47 else:48 self.data_thread.start()49 self.start_device_threads()50 def start_device_threads(self):51 for device, device_config in self.devices.items():52 thread = device_config.get('thread')53 if thread:54 if not thread.ident:55 thread.start()56 def parse_devices(self, config):57 """58 reads config.json and configures everything according to it. This method only takes care of the basics,59 so you may wanna extend it if your fxmode takes/requires additional settings.60 See fxmodes/screenfx/main.py for a reference implementation61 :return:62 """63 # These keys need to be set per device, everything else has some kinda default instead64 required_keys = {65 'wled': {'ip', 'leds'},66 'serial': {'path', 'leds'},67 'dualshock': {'device_num'},68 }69 # Here we'll put the final configurations70 final_devices = {}71 devices = config.get('devices')72 if not devices:73 print('You didn\'t define devices in your config, which renders this program kinda useless')74 exit()75 for name, device in devices.items():76 device_enabled = device.get('enabled', True)77 if device_enabled:78 device_type = device.get('type')79 if device_type:80 if device_type not in required_keys.keys():81 print(f'WARNING: device {name} has an invalid type, must be {required_keys.keys()}, skipping it')82 else:83 if not (required_keys.get(device_type) - device.keys()):84 base_config = {85 'type': device_type,86 'enabled': device_enabled,87 'brightness': device.get('brightness', 1),88 'flip': device.get('flip', False),89 'color_temperature': device.get('color_temperature'),90 'saturation': device.get('saturation', 1),91 'thread': threading.Thread(),92 }93 if device_type == 'wled':94 device_config = {95 'ip': device.get('ip'),96 'port': device.get('port', 21324),97 'leds': device.get('leds'),98 **base_config,99 }100 if device_type == 'serial':101 baud = device.get('baud', 115200)102 device_config = {103 'path': device.get('path'),104 'baud': baud,105 'leds': device.get('leds'),106 **base_config,107 }108 if device_type == 'dualshock':109 device_config = {110 'device_num': device.get('device_num'),111 **base_config,112 }113 device_config['thread'] = threading.Thread(114 target=self.device_loop,115 args=[name],116 kwargs={},117 )118 final_devices[name] = device_config119 else:120 print(f'WARNING: device {name} lacks one of these keys: '121 f'{required_keys.get(device_type)} skipping it.')122 else:123 print(f'WARNING: you didn\'t define the device type for "{name}", skipping it.')124 if not final_devices:125 print('ERROR: There\'s no device with complete configuration, please check your config!')126 print('Exiting now...')127 exit(1)128 self.devices = final_devices129 def check_target(self, core_version):130 """131 checks if the user's ImmersiveFX Core version and platform match the fxmode requirements132 can be overridden with --no-version-check and --no-platform-check, expect errors in this case though133 """134 if not self.launch_arguments.no_version_check:135 match = None136 core_major, core_minor, *_ = core_version.split('.')137 for version in self.target_versions:138 major, minor, *_ = version.split('.')139 if (major, minor) == (core_major, core_minor):140 match = True141 if not match:142 print('Your ImmersiveFX Core is on version %(core)s but it needs to be on %(targets)s' % {143 'core': '.'.join([core_major, core_minor]),144 'targets': ', '.join(self.target_versions),145 })146 print('for this FXMode to work!')147 exit(1)148 if not self.launch_arguments.no_platform_check:149 if sys.platform not in self.target_platforms and 'all' not in self.target_platforms:150 print('your platform is %(platform)s but it needs to be %(targets)s.' % {151 'platform': sys.platform,152 'targets': ', '.join(self.target_platforms),153 })154 exit()155 def splash(self):156 """157 Override this in your fxmode and print whatever you want. Preferably some kinda logo of course158 """159 print('***************************************************')160 print('* NO SPLASH SCREEN SPECIFIED! *')161 print('* you\'re seeing this because the fxmode creator *')162 print('* didn\'t override the splash() method. *')163 print('* No big deal but kinda boring, huh? *')164 print('***************************************************')165 def data_processing(self, *args, **kwargs):166 """167 Override this in your fxmode and continuously set self.raw_data168 """169 pass170 def data_loop(self, *args, **kwargs):171 """172 Manages data cycle timing, for handling preferably use data_processing173 """174 if self.launch_arguments.single_threaded:175 self.data_processing()176 else:177 while True:178 start = time()179 self.data_processing()180 duration = (time() - start) * 1000181 if duration > self.frame_sleep:182 if not self.launch_arguments.no_performance_warnings:183 print('WARNING: data cycle took longer than frame time!')184 print(f'frame time: {round(self.frame_sleep, 2)}ms, cycle time: {round(duration, 2)}ms')185 print('If this happens repeatedly, consider lowering the fps.')186 else:187 sleep((self.frame_sleep - duration) / 1000)188 def device_processing(self, device, device_instance):189 """190 Override this in your fxmode and process raw data so you end up with a list of rgb arrays191 :param device: device info as per config, provided so that custom fxmode config keys can be respected192 :param device_instance: device class instance, containing further info193 :return: a 2d numpy array; a list of rgb lists194 """195 return np.zeros([1, 3])196 def device_loop(self, device_name):197 """198 Thread manager for a device. Creates a class instance for it and runs a continuous loop to send data199 :param device_name: key to fetch device_config200 """201 device = self.devices.get(device_name)202 device_type = device.get('type')203 device_class = self.device_classes.get(device_type)204 def run_loop(instance):205 data = np.array(self.device_processing(device, instance))206 data = (instance.apply_enhancements(207 data * instance.brightness * instance.color_temperature208 )).astype(int)209 if instance.flip:210 data = np.flip(data, axis=0)211 instance.loop(data)212 if device_class:213 device_instance = device_class(device, device_name)214 if self.launch_arguments.single_threaded:215 if not device_instance.enabled:216 return None217 run_loop(device_instance)218 else:219 while True:220 if not device_instance.enabled:221 break222 start = time()223 run_loop(device_instance)224 duration = (time() - start) * 1000225 if duration > self.frame_sleep:226 if not self.launch_arguments.no_performance_warnings:227 print(f'WARNING: device "{device_instance.name}" cycle took longer than frame time!')228 print(f'frame time: {round(self.frame_sleep, 2)}ms, cycle time: {round(duration, 2)}ms')229 print('If this happens repeatedly, consider lowering the fps.')230 else:...
test_include_launch_description.py
Source:test_include_launch_description.py
...72 # Result should only contain the launch description as there are no launch arguments.73 assert action2.visit(lc2) == [ld2]74 assert lc2.locals.current_launch_file_directory == os.path.dirname(this_file)75 assert action2.get_asyncio_future() is None76def test_include_launch_description_launch_arguments():77 """Test the interactions between declared launch arguments and IncludeLaunchDescription."""78 # test that arguments are set when given, even if they are not declared79 ld1 = LaunchDescription([])80 action1 = IncludeLaunchDescription(81 LaunchDescriptionSource(ld1),82 launch_arguments={'foo': 'FOO'}.items(),83 )84 assert len(action1.launch_arguments) == 185 lc1 = LaunchContext()86 result1 = action1.visit(lc1)87 assert len(result1) == 288 assert isinstance(result1[0], SetLaunchConfiguration)89 assert perform_substitutions(lc1, result1[0].name) == 'foo'90 assert perform_substitutions(lc1, result1[0].value) == 'FOO'...
test_headful.py
Source:test_headful.py
1# Copyright (c) Microsoft Corporation.2#3# Licensed under the Apache License, Version 2.0 (the "License")4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7# http:#www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14import pytest15from flaky import flaky16async def test_should_have_default_url_when_launching_browser(17 browser_type, launch_arguments, tmpdir18):19 browser_context = await browser_type.launch_persistent_context(20 tmpdir, **{**launch_arguments, "headless": False}21 )22 urls = [page.url for page in browser_context.pages]23 assert urls == ["about:blank"]24 await browser_context.close()25async def test_headless_should_be_able_to_read_cookies_written_by_headful(26 browser_type, launch_arguments, server, tmpdir, is_chromium, is_win27):28 if is_chromium and is_win:29 pytest.skip("see https://github.com/microsoft/playwright/issues/717")30 return31 # Write a cookie in headful chrome32 headful_context = await browser_type.launch_persistent_context(33 tmpdir, **{**launch_arguments, "headless": False}34 )35 headful_page = await headful_context.new_page()36 await headful_page.goto(server.EMPTY_PAGE)37 await headful_page.evaluate(38 """() => document.cookie = 'foo=true; expires=Fri, 31 Dec 9999 23:59:59 GMT'"""39 )40 await headful_context.close()41 # Read the cookie from headless chrome42 headless_context = await browser_type.launch_persistent_context(43 tmpdir, **{**launch_arguments, "headless": True}44 )45 headless_page = await headless_context.new_page()46 await headless_page.goto(server.EMPTY_PAGE)47 cookie = await headless_page.evaluate("() => document.cookie")48 await headless_context.close()49 # This might throw. See https://github.com/GoogleChrome/puppeteer/issues/277850 assert cookie == "foo=true"51async def test_should_close_browser_with_beforeunload_page(52 browser_type, launch_arguments, server, tmpdir53):54 browser_context = await browser_type.launch_persistent_context(55 tmpdir, **{**launch_arguments, "headless": False}56 )57 page = await browser_context.new_page()58 await page.goto(server.PREFIX + "/beforeunload.html")59 # We have to interact with a page so that 'beforeunload' handlers60 # fire.61 await page.click("body")62 await browser_context.close()63async def test_should_not_crash_when_creating_second_context(64 browser_type, launch_arguments, server65):66 browser = await browser_type.launch(**{**launch_arguments, "headless": False})67 browser_context = await browser.new_context()68 await browser_context.new_page()69 await browser_context.close()70 browser_context = await browser.new_context()71 await browser_context.new_page()72 await browser_context.close()73 await browser.close()74async def test_should_click_background_tab(browser_type, launch_arguments, server):75 browser = await browser_type.launch(**{**launch_arguments, "headless": False})76 page = await browser.new_page()77 await page.set_content(78 f'<button>Hello</button><a target=_blank href="{server.EMPTY_PAGE}">empty.html</a>'79 )80 await page.click("a")81 await page.click("button")82 await browser.close()83async def test_should_close_browser_after_context_menu_was_triggered(84 browser_type, launch_arguments, server85):86 browser = await browser_type.launch(**{**launch_arguments, "headless": False})87 page = await browser.new_page()88 await page.goto(server.PREFIX + "/grid.html")89 await page.click("body", button="right")90 await browser.close()91async def test_should_not_block_third_party_cookies(92 browser_type, launch_arguments, server, is_chromium, is_firefox93):94 browser = await browser_type.launch(**{**launch_arguments, "headless": False})95 page = await browser.new_page()96 await page.goto(server.EMPTY_PAGE)97 await page.evaluate(98 """src => {99 let fulfill;100 const promise = new Promise(x => fulfill = x);101 const iframe = document.createElement('iframe');102 document.body.appendChild(iframe);103 iframe.onload = fulfill;104 iframe.src = src;105 return promise;106 }""",107 server.CROSS_PROCESS_PREFIX + "/grid.html",108 )109 document_cookie = await page.frames[1].evaluate(110 """() => {111 document.cookie = 'username=John Doe';112 return document.cookie;113 }"""114 )115 await page.wait_for_timeout(2000)116 allows_third_party = is_chromium or is_firefox117 assert document_cookie == ("username=John Doe" if allows_third_party else "")118 cookies = await page.context.cookies(server.CROSS_PROCESS_PREFIX + "/grid.html")119 if allows_third_party:120 assert cookies == [121 {122 "domain": "127.0.0.1",123 "expires": -1,124 "httpOnly": False,125 "name": "username",126 "path": "/",127 "sameSite": "None",128 "secure": False,129 "value": "John Doe",130 }131 ]132 else:133 assert cookies == []134 await browser.close()135@pytest.mark.skip_browser("webkit")136async def test_should_not_override_viewport_size_when_passed_null(137 browser_type, launch_arguments, server138):139 # Our WebKit embedder does not respect window features.140 browser = await browser_type.launch(**{**launch_arguments, "headless": False})141 context = await browser.new_context(no_viewport=True)142 page = await context.new_page()143 await page.goto(server.EMPTY_PAGE)144 async with page.expect_popup() as popup_info:145 await page.evaluate(146 """() => {147 const win = window.open(window.location.href, 'Title', 'toolbar=no,location=no,directories=no,status=no,menubar=no,scrollbars=yes,resizable=yes,width=600,height=300,top=0,left=0');148 win.resizeTo(500, 450);149 }"""150 )151 popup = await popup_info.value152 await popup.wait_for_load_state()153 await popup.wait_for_function(154 """() => window.outerWidth === 500 && window.outerHeight === 450"""155 )156 await context.close()157 await browser.close()158@flaky159async def test_page_bring_to_front_should_work(browser_type, launch_arguments):160 browser = await browser_type.launch(**{**launch_arguments, "headless": False})161 page1 = await browser.new_page()162 await page1.set_content("Page1")163 page2 = await browser.new_page()164 await page2.set_content("Page2")165 await page1.bring_to_front()166 assert await page1.evaluate("document.visibilityState") == "visible"167 assert await page2.evaluate("document.visibilityState") == "visible"168 await page2.bring_to_front()169 assert await page1.evaluate("document.visibilityState") == "visible"170 assert await page2.evaluate("document.visibilityState") == "visible"...
runner_interpreter.py
Source:runner_interpreter.py
1"""Transform runner parameters to data usable for runtime execution"""2import os3import shlex4import stat5from lutris.util import system6from lutris.util.linux import LINUX_SYSTEM7from lutris.util.log import logger8def get_mangohud_conf(system_config):9 """Return correct launch arguments and environment variables for Mangohud."""10 mango_args = []11 mangohud = system_config.get("mangohud") or ""12 if mangohud and system.find_executable("mangohud"):13 if mangohud == "gl32":14 mango_args = ["mangohud.x86"]15 else:16 mango_args = ["mangohud"]17 return mango_args, {"MANGOHUD": "1", "MANGOHUD_DLSYM": "1"}18def get_launch_parameters(runner, gameplay_info):19 system_config = runner.system_config20 launch_arguments = gameplay_info["command"]21 env = {22 "DISABLE_LAYER_AMD_SWITCHABLE_GRAPHICS_1": "1"23 }24 # Steam compatibility25 if os.environ.get("SteamAppId"):26 logger.info("Game launched from steam (AppId: %s)", os.environ["SteamAppId"])27 env["LC_ALL"] = ""28 # Optimus29 optimus = system_config.get("optimus")30 if optimus == "primusrun" and system.find_executable("primusrun"):31 launch_arguments.insert(0, "primusrun")32 elif optimus == "optirun" and system.find_executable("optirun"):33 launch_arguments.insert(0, "virtualgl")34 launch_arguments.insert(0, "-b")35 launch_arguments.insert(0, "optirun")36 elif optimus == "pvkrun" and system.find_executable("pvkrun"):37 launch_arguments.insert(0, "pvkrun")38 mango_args, mango_env = get_mangohud_conf(system_config)39 if mango_args:40 launch_arguments = mango_args + launch_arguments41 env.update(mango_env)42 # Libstrangle43 fps_limit = system_config.get("fps_limit") or ""44 if fps_limit:45 strangle_cmd = system.find_executable("strangle")46 if strangle_cmd:47 launch_arguments = [strangle_cmd, fps_limit] + launch_arguments48 else:49 logger.warning("libstrangle is not available on this system, FPS limiter disabled")50 prefix_command = system_config.get("prefix_command") or ""51 if prefix_command:52 launch_arguments = (shlex.split(os.path.expandvars(prefix_command)) + launch_arguments)53 single_cpu = system_config.get("single_cpu") or False54 if single_cpu:55 limit_cpu_count = system_config.get("limit_cpu_count")56 if limit_cpu_count and limit_cpu_count.isnumeric():57 limit_cpu_count = int(limit_cpu_count)58 else:59 limit_cpu_count = 160 limit_cpu_count = max(1, limit_cpu_count)61 logger.info("The game will run on %d CPU core(s)", limit_cpu_count)62 launch_arguments.insert(0, "0-%d" % (limit_cpu_count - 1))63 launch_arguments.insert(0, "-c")64 launch_arguments.insert(0, "taskset")65 env.update(runner.get_env())66 env.update(gameplay_info.get("env") or {})67 # Set environment variables dependent on gameplay info68 # LD_PRELOAD69 ld_preload = gameplay_info.get("ld_preload")70 if ld_preload:71 env["LD_PRELOAD"] = ld_preload72 # LD_LIBRARY_PATH73 game_ld_library_path = gameplay_info.get("ld_library_path")74 if game_ld_library_path:75 ld_library_path = env.get("LD_LIBRARY_PATH")76 env["LD_LIBRARY_PATH"] = os.pathsep.join(filter(None, [77 game_ld_library_path, ld_library_path]))78 # Feral gamemode79 gamemode = system_config.get("gamemode") and LINUX_SYSTEM.gamemode_available()80 if gamemode:81 launch_arguments.insert(0, "gamemoderun")82 # Gamescope83 gamescope = system_config.get("gamescope") and system.find_executable("gamescope")84 if gamescope:85 launch_arguments = get_gamescope_args(launch_arguments, system_config)86 return launch_arguments, env87def get_gamescope_args(launch_arguments, system_config):88 """Insert gamescope at the start of the launch arguments"""89 launch_arguments.insert(0, "--")90 launch_arguments.insert(0, "-f")91 if system_config.get("gamescope_output_res"):92 output_width, output_height = system_config["gamescope_output_res"].lower().split("x")93 launch_arguments.insert(0, output_height)94 launch_arguments.insert(0, "-H")95 launch_arguments.insert(0, output_width)96 launch_arguments.insert(0, "-W")97 if system_config.get("gamescope_game_res"):98 game_width, game_height = system_config["gamescope_game_res"].lower().split("x")99 launch_arguments.insert(0, game_height)100 launch_arguments.insert(0, "-h")101 launch_arguments.insert(0, game_width)102 launch_arguments.insert(0, "-w")103 launch_arguments.insert(0, "gamescope")104 return launch_arguments105def export_bash_script(runner, gameplay_info, script_path):106 """Convert runner configuration into a bash script"""107 if getattr(runner, 'prelaunch', None) is not None:108 runner.prelaunch()109 command, env = get_launch_parameters(runner, gameplay_info)110 # Override TERM otherwise the script might not run111 env["TERM"] = "xterm"112 script_content = "#!/bin/bash\n\n\n"113 script_content += "# Environment variables\n"114 for name, value in env.items():115 script_content += 'export %s="%s"\n' % (name, value)116 script_content += "\n# Command\n"117 script_content += " ".join([shlex.quote(c) for c in command])118 with open(script_path, "w", encoding='utf-8') as script_file:119 script_file.write(script_content)...
LambdaTest’s Playwright tutorial will give you a broader idea about the Playwright automation framework, its unique features, and use cases with examples to exceed your understanding of Playwright testing. This tutorial will give A to Z guidance, from installing the Playwright framework to some best practices and advanced concepts.
Get 100 minutes of automation test minutes FREE!!