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!!
