How to use portpicker - 10 common examples

def setUpClass(cls):
    super(GRRSeleniumTest, cls).setUpClass()
    with GRRSeleniumTest._selenium_set_up_lock:
      if not GRRSeleniumTest._selenium_set_up_done:

        port = portpicker.pick_unused_port()
        logging.info("Picked free AdminUI port %d.", port)

        # Start up a server in another thread
        GRRSeleniumTest._server_trd = wsgiapp_testlib.ServerThread(
            port, name="SeleniumServerThread")
        GRRSeleniumTest._server_trd.StartAndWaitUntilServing()
        GRRSeleniumTest._SetUpSelenium(port)

        GRRSeleniumTest._selenium_set_up_done = True

def _StartEmulator(self, timer,
                     net_type, new_process_group, window_scale,
                     with_audio, with_boot_anim):
    """Start emulator or user mode android."""

    if not self.emulator_adb_port:
      self.emulator_adb_port = portpicker.PickUnusedPort()
    if not self.emulator_telnet_port:
      self.emulator_telnet_port = portpicker.PickUnusedPort()
    if not self.device_serial:
      self.device_serial = 'localhost:%s' % self.emulator_adb_port

    emulator_binary = os.path.abspath(
        self.android_platform.GetEmulator(
            self._metadata_pb.emulator_architecture,
            self._metadata_pb.emulator_type))

    pipe_dir = self._TempDir('pipe_trav')
    exec_dir = self._SessionImagesDir()
    self._emulator_env = self._MakeEmulatorEnv(os.environ)

    if (self._metadata_pb.emulator_type in
        [emulator_meta_data_pb2.EmulatorMetaDataPb.QEMU,

def ConnectDevice(self):
    """Connects the device to the adb server.

    Returns:
      True on Success, False otherwise.
    """
    assert self._CanConnect()
    if FLAGS.skip_connect_device:
      return True
    connect_args = [self.android_platform.real_adb,
                    'connect',
                    'localhost:%s' % self.emulator_adb_port]
    logging.info('Connecting adb server to device: %s', connect_args)
    connect_task = None
    if not self.adb_server_port:
      self.adb_server_port = portpicker.PickUnusedPort()
    elif self.adb_server_port < 0 or self.adb_server_port > 65535:
      logging.warn('Invalid adb server port %d, skip connecting',
                   self.adb_server_port)
      return

    try:
      logging.info('Starting: %s', connect_args)
      connect_task = common.SpawnAndWaitWithRetry(
          connect_args,
          proc_output=True,
          exec_env=self._AdbEnv(),
          timeout_seconds=ADB_SHORT_TIMEOUT_SECONDS,
          retries=5)
      logging.info('Done: %s', connect_args)
    except common.SpawnError:
      return False

def _AdbEnv(self):
    """Prepare environment for running adb."""

    env = {}
    if not self.adb_server_port:
      self.adb_server_port = portpicker.PickUnusedPort()
    if not self.device_serial:
      self.device_serial = 'localhost:%s' % self.emulator_adb_port
    env['ANDROID_ADB_SERVER_PORT'] = str(self.adb_server_port)
    env['ANDROID_ADB'] = self.android_platform.real_adb
    env['ANDROID_SERIAL'] = self.device_serial
    env['ADB_LIBUSB'] = '0'
    if self._emulator_env and 'HOME' in self._emulator_env:
      env['HOME'] = self._emulator_env['HOME']
    test_output_dir = os.environ.get('TEST_UNDECLARED_OUTPUTS_DIR')
    if test_output_dir:
      env['TMPDIR'] = test_output_dir
    return env

def _StartEmulator(self, timer,
                     net_type, new_process_group, window_scale,
                     with_audio, with_boot_anim):
    """Start emulator or user mode android."""

    if not self.emulator_adb_port:
      self.emulator_adb_port = portpicker.PickUnusedPort()
    if not self.emulator_telnet_port:
      self.emulator_telnet_port = portpicker.PickUnusedPort()
    if not self.device_serial:
      self.device_serial = 'localhost:%s' % self.emulator_adb_port

    emulator_binary = os.path.abspath(
        self.android_platform.GetEmulator(
            self._metadata_pb.emulator_architecture,
            self._metadata_pb.emulator_type))

    pipe_dir = self._TempDir('pipe_trav')
    exec_dir = self._SessionImagesDir()
    self._emulator_env = self._MakeEmulatorEnv(os.environ)

    if (self._metadata_pb.emulator_type in
        [emulator_meta_data_pb2.EmulatorMetaDataPb.QEMU,
         emulator_meta_data_pb2.EmulatorMetaDataPb.QEMU2]):
      self._PrepareQemuArgs(emulator_binary, net_type, window_scale,

def close(self):
    """Shut down the game and clean up."""
    if hasattr(self, "_controller") and self._controller:
      self._controller.quit()
      self._controller.close()
      self._controller = None
    self._shutdown()
    if hasattr(self, "_port") and self._port:
      if not FLAGS.sc2_port:
        portpicker.return_port(self._port)
      self._port = None
    if hasattr(self, "_tmp_dir") and os.path.exists(self._tmp_dir):
      shutil.rmtree(self._tmp_dir)

def close(self):
    """Shut down the game and clean up."""
    if hasattr(self, "_controller") and self._controller:
      self._controller.quit()
      self._controller.close()
      self._controller = None
    self._shutdown()
    if hasattr(self, "_port") and self._port:
      portpicker.return_port(self._port)
      self._port = None
    if hasattr(self, "_tmp_dir") and os.path.exists(self._tmp_dir):
      shutil.rmtree(self._tmp_dir)

self._renderer_human.close()
      self._renderer_human = None

    # Don't use parallel since it might be broken by an exception.
    if hasattr(self, "_controllers") and self._controllers:
      for c in self._controllers:
        c.quit()
      self._controllers = None
    if hasattr(self, "_sc2_procs") and self._sc2_procs:
      for p in self._sc2_procs:
        p.close()
      self._sc2_procs = None

    if hasattr(self, "_ports") and self._ports:
      for port in self._ports:
        portpicker.return_port(port)
      self._ports = None

self._emulator_cmd = emulator_cmd
    self._http = httplib2.Http()
    self.__running = False

    self._tmp_dir = tempfile.mkdtemp(dir=working_directory)
    self._project_directory = os.path.join(self._tmp_dir, self._project_id)
    p = subprocess.Popen([emulator_cmd,
                          'create',
                          '--project_id=%s' % self._project_id,
                          self._project_directory])
    if p.wait() != 0:
      raise IOError('could not create project in directory: %s'
                    % self._project_directory)

    # Start the emulator and wait for it to start responding to requests.
    port = portpicker.PickUnusedPort()
    self._host = 'http://localhost:%d' % port
    cmd = [self._emulator_cmd, 'start', '--port=%d' % port]
    cmd.extend(_DEFAULT_EMULATOR_OPTIONS)
    if start_options:
      cmd.extend(start_options)
    cmd.append(self._project_directory)
    subprocess.Popen(cmd)
    if not self._WaitForStartup(deadline):
      raise IOError('emulator did not respond within %ds' % deadline)
    endpoint = '%s/v1/projects/%s' % (self._host, self._project_id)
    self.__datastore = connection.Datastore(project_endpoint=endpoint)
    self.__running = True

def human():
  """Run a host which expects one player to connect remotely."""
  run_config = run_configs.get()

  map_inst = maps.get(FLAGS.map)

  if not FLAGS.rgb_screen_size or not FLAGS.rgb_minimap_size:
    logging.info("Use --rgb_screen_size and --rgb_minimap_size if you want rgb "
                 "observations.")

  ports = [FLAGS.config_port + p for p in range(5)]  # tcp + 2 * num_players
  if not all(portpicker.is_port_free(p) for p in ports):
    sys.exit("Need 5 free ports after the config port.")

  proc = None
  ssh_proc = None
  tcp_conn = None
  udp_sock = None
  try:
    proc = run_config.start(extra_ports=ports[1:], timeout_seconds=300,
                            host=FLAGS.host, window_loc=(50, 50))

    tcp_port = ports[0]
    settings = {
        "remote": FLAGS.remote,
        "game_version": proc.version.game_version,
        "realtime": FLAGS.realtime,
        "map_name": map_inst.name,