Best Python code snippet using playwright-python
worldstate.py
Source:worldstate.py
1# -*- coding: utf-8 -*-2from sdtp.alias_table import AliasTable3from sdtp.lkp_table import lkp_table4from sdtp.lp_table import lp_table5import logging6from sqlalchemy import Integer7import sys8import threading9import time10class WorldState(threading.Thread):11 def __init__ ( self, controller ):12 super ( self.__class__, self ).__init__ ( )13 self.controller = controller14 self.keep_running = True15 self.logger = logging.getLogger(__name__)16 self.online_players = []17 self.online_players_count = 10000018 self.online_players_changed = False19 self.latest_day = 020 self.latest_hour = 021 self.server_empty = False22 self.zombies_alive = -123 # Thread control24 def run ( self ):25 self.logger.info("Start.")26 self.register_callbacks ( )27 while ( self.keep_running ):28 time.sleep ( 0.1 )29 self.deregister_callbacks ( )30 self.logger.debug("world_state.run returning" )31 def stop ( self ):32 self.keep_running = False33 self.logger.info("Stop.")34 # Component-specific35 ####################36 def register_callbacks ( self ):37 self.controller.dispatcher.register_callback(38 "gt command output", self.log_time_of_day)39 self.controller.dispatcher.register_callback(40 "lkp output", self.parse_lkp_output)41 self.controller.dispatcher.register_callback(42 "lp output", self.update_lkp_table )43 self.controller.dispatcher.register_callback(44 "lp output", self.update_lp_table )45 self.controller.dispatcher.register_callback(46 "le/lp output footer", self.update_online_players_count )47 self.controller.dispatcher.register_callback(48 "mem output", self.update_zombies_alive)49 self.controller.dispatcher.register_callback(50 "player joined", self.player_connected)51 self.controller.dispatcher.register_callback(52 "player left", self.player_disconnected)53 def deregister_callbacks ( self ):54 self.controller.dispatcher.deregister_callback(55 "gt command output", self.log_time_of_day)56 self.controller.dispatcher.deregister_callback(57 "lkp output", self.parse_lkp_output)58 self.controller.dispatcher.deregister_callback(59 "lp output", self.update_lkp_table )60 self.controller.dispatcher.deregister_callback (61 "lp output", self.update_lp_table )62 self.controller.dispatcher.deregister_callback (63 "le/lp output footer", self.update_online_players_count )64 self.controller.dispatcher.deregister_callback(65 "mem output", self.update_zombies_alive)66 self.controller.dispatcher.deregister_callback(67 "player joined", self.player_connected)68 self.controller.dispatcher.deregister_callback(69 "player left", self.player_disconnected)70 def player_connected(self, match_group):71 self.logger.info("Player {} connected.".format(match_group[7]))72 self.online_players_changed = True73 74 def player_disconnected(self, match_group):75 self.logger.info("Player {} disconnected.".format(match_group[7]))76 self.online_players_changed = True77 78 def update_lkp_table ( self, match_group ):79 self.logger.debug("update_lkp_table for {}.".format(match_group[1]))80 self.logger.debug("({})".format (match_group))81 this_steamid = int ( match_group [ 15 ] ),82 results = self.controller.database.blocking_consult(83 lkp_table, [ ( lkp_table.steamid, "==", match_group [ 15 ] ) ])84 self.logger.debug("results = {}.".format ( results ) )85 if len ( results ) == 0:86 self.logger.info("New lkp entry: {}".format(match_group[1]))87 self.controller.database.blocking_add(88 lkp_table,89 [ lkp_table (90 player_id = match_group [ 0 ],91 name = match_group [ 1 ],92 longitude = match_group [ 2 ],93 height = match_group [ 3 ],94 latitude = match_group [ 4 ],95 rotation_height = match_group [ 5 ],96 rotation_longitude = match_group [ 6 ],97 rotation_latitude = match_group [ 7 ],98 remote = match_group [ 8 ],99 health = match_group [ 9 ],100 deaths = match_group [ 10 ],101 zombies = match_group [ 11 ],102 players = match_group [ 12 ],103 score = match_group [ 13 ],104 level = match_group [ 14 ],105 steamid = match_group [ 15 ],106 ip = match_group [ 16 ],107 ping = match_group [ 17 ] ) ] )108 else:109 self.logger.debug("Update lkp entry." )110 entry = results [ 0 ]111 self.logger.debug("entry before: {}".format(entry))112 entry [ "name" ] = match_group [ 1 ]113 entry [ "longitude" ] = match_group [ 2 ]114 entry [ "height" ] = match_group [ 3 ]115 entry [ "latitude" ] = match_group [ 4 ]116 entry [ "rotation_height" ] = match_group [ 5 ]117 entry [ "rotation_longitude" ] = match_group [ 6 ]118 entry [ "rotation_latitude" ] = match_group [ 7 ]119 entry [ "remote" ] = match_group [ 8 ]120 entry [ "health" ] = match_group [ 9 ]121 entry [ "deaths" ] = match_group [ 10 ]122 entry [ "zombies" ] = match_group [ 11 ]123 entry [ "players" ] = match_group [ 12 ]124 entry [ "score" ] = match_group [ 13 ]125 entry [ "level" ] = match_group [ 14 ]126 entry["steamid"] = match_group[15]127 entry [ "ip" ] = match_group [ 16 ]128 entry [ "ping" ] = match_group [ 17 ]129 self.logger.debug("entry after: {}".format(entry))130 self.controller.database.blocking_update(131 lkp_table,132 entry )133 self.logger.debug("added entry." )134 self.logger.debug("returning." )135 def update_lp_table ( self, match_group ):136 self.logger.debug("({})".format (match_group))137 if self.online_players_changed:138 self.controller.database.blocking_delete(139 lp_table, [])140 self.online_players_changed = False141 this_steamid = int ( match_group [ 15 ] )142 results = self.controller.database.blocking_consult (143 lp_table, [ ( lp_table.steamid, "==", match_group [ 15 ] ) ])144 self.logger.debug("results = {}.".format ( results ) )145 if len ( results ) == 0:146 self.logger.debug("New entry." )147 self.controller.database.blocking_add(148 lp_table,149 [ lp_table (150 player_id = match_group [ 0 ],151 name = match_group [ 1 ],152 longitude = match_group [ 2 ],153 height = match_group [ 3 ],154 latitude = match_group [ 4 ],155 rotation_height = match_group [ 5 ],156 rotation_longitude = match_group [ 6 ],157 rotation_latitude = match_group [ 7 ],158 remote = match_group [ 8 ],159 health = match_group [ 9 ],160 deaths = match_group [ 10 ],161 zombies = match_group [ 11 ],162 players = match_group [ 12 ],163 score = match_group [ 13 ],164 level = match_group [ 14 ],165 steamid = match_group [ 15 ],166 ip = match_group [ 16 ],167 ping = match_group [ 17 ] ) ])168 else:169 self.logger.debug("Update lp entry." )170 entry = results [ 0 ]171 self.logger.debug("entry before: {}".format(entry))172 entry [ "name" ] = match_group [ 1 ]173 entry [ "longitude" ] = match_group [ 2 ]174 entry [ "height" ] = match_group [ 3 ]175 entry [ "latitude" ] = match_group [ 4 ]176 entry [ "rotation_height" ] = match_group [ 5 ]177 entry [ "rotation_longitude" ] = match_group [ 6 ]178 entry [ "rotation_latitude" ] = match_group [ 7 ]179 entry [ "remote" ] = match_group [ 8 ]180 entry [ "health" ] = match_group [ 9 ]181 entry [ "deaths" ] = match_group [ 10 ]182 entry [ "zombies" ] = match_group [ 11 ]183 entry [ "players" ] = match_group [ 12 ]184 entry [ "score" ] = match_group [ 13 ]185 entry [ "level" ] = match_group [ 14 ]186 entry["steamid"] = match_group[15]187 entry [ "ip" ] = match_group [ 16 ]188 entry [ "ping" ] = match_group [ 17 ]189 self.logger.debug("Entry after: {}".format(entry))190 self.controller.database.blocking_update(191 lp_table,192 entry)193 self.logger.debug("Updated entry." )194 self.logger.debug("returning." )195 def update_online_players_count(self, match_group):196 self.logger.debug(match_group)197 count = int ( match_group [ 0 ] )198 self.online_players_count = count199 if count == 0:200 self.controller.database.blocking_delete(lp_table, [])201 self.server_empty = True202 return203 else:204 self.server_empty = False205 206 answer = self.controller.database.blocking_consult(lp_table, [])207 if len(answer) != self.online_players_count:208 self.logger.debug(209 "Number of online players does not match number of DB entries.")210 self.online_players_changed = True211 def get_online_players(self):212 answer = self.controller.database.blocking_consult(213 lp_table,214 [])215 if len(answer) != self.online_players_count:216 self.logger.warning(217 "Number of online players {} does not match number of DB " \218 "entries {}.".format(self.online_players_count, len(answer)))219 self.online_players = answer220 self.logger.debug("{} players online now.".format(len(self.online_players)))221 return self.online_players222 def log_time_of_day(self, match_group):223 self.day = int(match_group[0])224 self.hour = int(match_group[1])225 self.minute = int(match_group[2])226 if self.day != self.latest_day:227 self.latest_day = self.day228 self.controller.dispatcher.call_registered_callbacks(229 "new day", [self.day, self.hour, self.minute])230 if self.hour != self.latest_hour:231 self.latest_hour = self.hour232 self.logger.info("Day {} {:02}:{:02}".format(233 self.day, self.hour, self.minute))234 self.controller.dispatcher.call_registered_callbacks(235 "new hour", [self.day, self.hour, self.minute])236 def parse_lkp_output(self, match_groups):237 name = match_groups[0]238 player_id = match_groups[1]239 steamid = match_groups[2]240 ip = match_groups[4]241 db = self.controller.database.blocking_consult(242 lkp_table, [(lkp_table.steamid, "==", steamid)])243 if len(db) != 0:244 self.logger.debug("lkp output for {} ignored: already on db.".format(245 name))246 return247 self.controller.database.blocking_add(248 lkp_table, [lkp_table(name = name,249 player_id = player_id,250 steamid = steamid,251 ip = ip)])252 self.logger.info("Added lkp_table entry for {}.".format(name))253 def update_zombies_alive(self, match_groups):254 self.zombies_alive = int(match_groups[7])255 256 # API257 258 def get_player_string(self, name_or_alias):259 self.logger.debug("Trying to get player for name or alias '{}'.".format(260 name_or_alias))261 db = self.controller.database.blocking_consult(262 lkp_table, [(lkp_table.name, "==", name_or_alias)])263 if len(db) == 1:264 self.logger.debug("Player {} found.".format(db[0]["name"]))265 return db[0]266 if len(db) > 1:267 self.logger.error("Multiple players with name {} on db.".format(268 name_or_alias))269 return None270 self.logger.info("No player with name {} on db.".format(name_or_alias))271 db = self.controller.database.blocking_consult(272 AliasTable, [(AliasTable.alias, "==", name_or_alias)])273 if len(db) > 1:274 self.logger.error("Multiple players with alias {} on db.".format(275 name_or_alias))276 return None277 if len(db) == 0:278 self.logger.warning("No player with alias {} on db.".format(279 name_or_alias))280 return None281 player = self.controller.database.blocking_consult(282 lkp_table, [(lkp_table.steamid, "==", db[0]["steamid"])])[0]283 self.logger.info("Found player {} with alias {}.".format(284 player["name"], name_or_alias))285 return player286 def get_player_steamid(self, steamid):287 self.logger.debug("Trying to find db entry for steamid {}.".format(288 steamid))289 db = self.controller.database.blocking_consult(290 lkp_table, [(lkp_table.steamid, "==", steamid)])291 if len(db) == 1:292 self.logger.debug("DB entry for steamid {} found: {}.".format(293 steamid, db[0]["name"]))294 return db[0]295 else:296 self.logger.warning(297 "Couldn't find single db entry for steamid {}.".format(298 steamid))...
help_provider.py
Source:help_provider.py
1# -*- coding: utf-8 -*-2# Copyright 2012 Google Inc. All Rights Reserved.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8# http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15"""Module defining help types and providers for gsutil commands."""16from __future__ import absolute_import17import collections18from gslib.exception import CommandException19ALL_HELP_TYPES = ['command_help', 'additional_help']20# Constants enforced by SanityCheck21MAX_HELP_NAME_LEN = 1622MIN_ONE_LINE_SUMMARY_LEN = 1023MAX_ONE_LINE_SUMMARY_LEN = 80 - MAX_HELP_NAME_LEN24DESCRIPTION_PREFIX = """25<B>DESCRIPTION</B>"""26SYNOPSIS_PREFIX = """27<B>SYNOPSIS</B>"""28class HelpProvider(object):29 """Interface for providing help."""30 # Each subclass of HelpProvider define a property named 'help_spec' that is31 # an instance of the following class.32 HelpSpec = collections.namedtuple('HelpSpec', [33 # Name of command or auxiliary help info for which this help applies.34 'help_name',35 # List of help name aliases.36 'help_name_aliases',37 # Type of help.38 'help_type',39 # One line summary of this help.40 'help_one_line_summary',41 # The full help text.42 'help_text',43 # Help text for subcommands of the command's help being specified.44 'subcommand_help_text',45 ])46 # Each subclass must override this with an instance of HelpSpec.47 help_spec = None48# This is a static helper instead of a class method because the help loader49# (gslib.commands.help._LoadHelpMaps()) operates on classes not instances.50def SanityCheck(help_provider, help_name_map):51 """Helper for checking that a HelpProvider has minimally adequate content."""52 # Sanity check the content.53 assert (len(help_provider.help_spec.help_name) > 154 and len(help_provider.help_spec.help_name) < MAX_HELP_NAME_LEN)55 for hna in help_provider.help_spec.help_name_aliases:56 assert hna57 one_line_summary_len = len(help_provider.help_spec.help_one_line_summary)58 assert (one_line_summary_len > MIN_ONE_LINE_SUMMARY_LEN59 and one_line_summary_len < MAX_ONE_LINE_SUMMARY_LEN)60 assert len(help_provider.help_spec.help_text) > 1061 # Ensure there are no dupe help names or aliases across commands.62 name_check_list = [help_provider.help_spec.help_name]63 name_check_list.extend(help_provider.help_spec.help_name_aliases)64 for name_or_alias in name_check_list:65 if help_name_map.has_key(name_or_alias):66 raise CommandException(67 'Duplicate help name/alias "%s" found while loading help from %s. '68 'That name/alias was already taken by %s' % (69 name_or_alias, help_provider.__module__,70 help_name_map[name_or_alias].__module__))71def CreateHelpText(synopsis, description):72 """Helper for adding help text headers given synopsis and description."""...
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!!