python类choices()的实例源码

def check_predict():
    Best = namedtuple("Best", ["U", "values", "min_value"])
    values = [((0, 0), 0) for _ in range(N_TOP_PER_CHUNK)]
    bests = []
    bests.append(Best(U_sinh, list(values), 0))
    bests.append(Best(U_rr, list(values), 0))
    user_indices = random.choices(range(len(users)), k=N_PER_CHUNK)
    repo_indices = random.choices(range(len(repos)), k=N_PER_CHUNK)
    for index in range(N_PER_CHUNK):
        pos = (user_indices[index], repo_indices[index])
        if pos in nonzero_positions:    # Link already exists.
            continue
        for i in range(len(bests)):
            new = bests[i].U[pos[0], :] @ V[:, pos[1]]
            if new > bests[i].min_value:
                bests[i].values.append((pos, new))
                bests[i].values.sort(key=lambda p: p[1], reverse=True)
                bests[i].values.pop()
                bests[i] = bests[i]._replace(min_value=bests[i].values[-1][1])
    for i in range(len(bests)):
        assert len(bests[i].values) == N_TOP_PER_CHUNK
    return bests[0].values, bests[1].values

def _random_depths(self):
        # Pick a random depth
        middle = (self._max_depth // 2) + 1
        weight = self._max_depth + 1
        population = range(weight)
        cum_weights = list()
        cumulative = 0.8
        for d in population:
            # first value 0 (zero) will start with default weight
            if d <= middle:
                cumulative += d
                cum_weights.append(cumulative)
            else:
                cumulative += weight - d
                cum_weights.append(cumulative)
        return choices(population, cum_weights=cum_weights, k=self.total_items)

def test_wildcard_dns_record(self):
        global wildcard_dns_record
        ip_dic = {}
        genrandstr = lambda i: ''.join(random.choices(string.ascii_lowercase + string.digits, k=i))
        tasks = [asyncio.ensure_future(self.resolver.query(genrandstr(20) + '.' + self.domain, 'A')) for _ in range(6)]
        reqs = asyncio.gather(*tasks)
        result = self.loop.run_until_complete(reqs)
        for r in result:
            if ip_dic.get(r.ip[0]):
                ip_dic[r.ip[0]] += 1
                if ip_dic[r.ip[0]] > 3:
                    wildcard_dns_record = r.ip[0]
                    print(f'[*] Found wildcard dns record:{wildcard_dns_record}')
                    return
            else:
                ip_dic[r.ip[0]] = 1

def run(self):
        import random
        import string

        # Create a dir contains suplat files
        working_path = os.getcwd()
        suplat_dir = os.path.join(working_path,
                                  'SUPLAT_{:}'.format(self.comment))
        if not os.path.exists(suplat_dir):
            os.makedirs(suplat_dir)
        else:
            shutil.rmtree(suplat_dir)
            os.makedirs(suplat_dir)

        for hnf in self.hnfs:
            rd_suffix = ''.join(random.choices(string.ascii_uppercase
                                               + string.digits, k=4))
            sl_origin = hnf.to_general_cell()
            sl = sl_origin.get_shaped_cell()

            out = GeneralIO(sl)

            ofname = "SUPLAT_{:}_{:}.{:}".format(self.v, rd_suffix, self.outmode)
            lastpath = os.path.join(suplat_dir, ofname)
            out.write_file(lastpath)

def issue_key():
    # issue api token
    run = True
    while run:
        # issue key
        key = "".join(
            random.choices(
                string.ascii_letters +
                string.digits,
                k=32))
        g.user.api_key = key
        try:
            db.session.add(g.user)
            db.session.commit()
            run = False
        except IntegrityError:  # check for uniqueness
            continue
    return g.user.api_key

def create_test_subvoat(self, count=0):
        api_token = self.get_api_token()

        for x in range(count):

            body  = ''.join(random.choices(string.ascii_letters, k=100))
            title = ''.join(random.choices(string.ascii_letters, k=10))

            result = requests.post('%s/create_subvoat' % (self.base_address), {'subvoat_name':'test_%s' % (title[0:5]),
                                                                              'username':'test_username', 
                                                                              'api_token':api_token,
                                                                              'body':body,
                                                                              'title':title})

            self.check(result)

        #return result.json()['result']


    # test subvoat. 
    # try to create thread without subvoat. 
    # try to create subvoat that already exists
    # try to create subvoat with name length out of parameters (see config.json)
    # try to create subvoat with incorrect api_token
    # try to create subvoat with fake user

def test_pareto_front_attrs():
    amax = 1.0
    amin = 0.7
    bmax = 1.3
    bmin = 1.0

    front_fitness = [(1, 1), (0.9, 1.1), (0.8, 1.2), (0.7, 1.3)]

    models = [Model(a, b) for a, b in front_fitness]
    models.extend([Model(a + 0.01 * abs(random.random()) + 0.01, b + 0.01 *
                         abs(random.random()) + 0.01) for a, b in random.choices(front_fitness, k=50)])
    models = list(filter(lambda m: m.b <= bmax, models))

    front = pareto_front(models, "a", "b")

    for m in front:
        assert amin <= m.a <= amax
        assert bmin <= m.b <= bmax

def new_chat_member_event(chat, member):
    logger.info('New chat member %s joined group', member['first_name'])
    text = format_text('''
    {greet}, {name}!

    Guruhga xush kelibsiz!

    Ushbu guruh o'zbek dasturchilari uchun ochilgan bo'lib, bu yerda guruh a'zolar bir-birlari bilan tajriba almashishlari, savol-javob qilishlari va shu sohadagi foydali narsalar (texnologiyalar, yangiliklar) ni o'zaro ulashishlari maqsad qilingan.

    {name}, {wish}. {emoticon}
    ''')
    greetings = (
        'Assalomu alaykum', 'Salom', 'Guruhimizning yangi a\'zosi',
    )
    greet = random.choice(greetings)
    wishes = (
        'guruhda faol bo\'lasiz degan umiddaman',
        'ishlaringizga omad',
        'yana bir bor hush kelibsiz',
    )
    wish = random.choices(wishes)
    emoticons = (
        '??', '??', '??', '??', '??', '??'
    )
    emoticon = random.choice(emoticons)

    if not await user_exists(chat.bot.pg_pool, member):
        await insert_user(chat.bot.pg_pool, member)

    await chat.send_text(
        text.format(name=member['first_name'], greet=greet, wish=wish, emoticon=emoticon))

def softmax_policy(self):
        action = choices(list(range(self.action_size)), weights=self.softmax)[0]
        return action

    # actor

def softmax_policy(self):
        action = choices(list(range(self.action_size)), weights=self.softmax)[0]
        return action

def randstring(n=1, source=None):
    """Generate a random string of length 'n' from 'source'."""
    if not source:
        source = string.ascii_letters + string.digits
    if n <= 0:
        raise ValueError("Length of sequence must be greater than 0.", n)
    return ''.join(random.choices(source, k=n))

def generate_session_id():
    return ''.join([random.choices(string.digits + string.ascii_letters)[0] for x in range(32)])

def check_predict(_):
    print("#", end="")
    sys.stdout.flush()
    max_b = 0
    user_indices = random.choices(range(len(users)), k=N_PER_CHUNK)
    repo_indices = random.choices(range(len(repos)), k=N_PER_CHUNK)
    for index in range(N_PER_CHUNK):
        if B[user_indices[index], repo_indices[index]] == 1.0:  # Link already exists.
            continue
        new_b = U[user_indices[index], :] @ V[:, repo_indices[index]]
        if new_b > max_b:
            max_b = new_b
            user_id = users[user_indices[index]]
            repo_id = repos[repo_indices[index]]
            best_pred = (user_id, user_id2name[user_id],
                         repo_id, repo_id2name[repo_id],
                         new_b)
    # print(best_pred)
    return best_pred

def check_predict():
    Best = namedtuple("Best", ["U", "values", "min_value"])
    values = [((0, 0), 0) for _ in range(N_TOP_PER_CHUNK)]
    bests = []
    bests.append(Best(U_sinh, list(values), 0))
    bests.append(Best(U_rr, list(values), 0))
    user_indices = random.choices(range(len(users)), k=N_PER_CHUNK)
    repo_indices = random.choices(range(len(repos)), k=N_PER_CHUNK)
    for index in range(N_PER_CHUNK):
        pos = (user_indices[index], repo_indices[index])
        if pos in nonzero_positions:    # Link already exists.
            continue
        for i in range(len(bests)):
            new = bests[i].U[pos[0], :] @ V[:, pos[1]]
            if new > bests[i].min_value:
                bests[i].values.append((pos, new))
                bests[i].values.sort(key=lambda p: p[1], reverse=True)
                bests[i].values.pop()
                bests[i] = bests[i]._replace(min_value=bests[i].values[-1][1])
    for i in range(len(bests)):
        assert len(bests[i].values) == N_TOP_PER_CHUNK
    return bests[0].values, bests[1].values

def get_records_for_test_db(src_db_path, num_rec):
    with sqlite3.connect(str(src_db_path)) as src_conn:
        src_conn.row_factory = sqlite3.Row
        query_yield = src_conn.execute('''SELECT * FROM moz_places''')
        # all_records = [DBRecord(record) for record in enumerate(query_yield)]
        all_records = query_yield.fetchall()
    chosen_records = random.choices(all_records, k=num_rec)
    chosen_records = [DBRecord(*record) for record in chosen_records]
    return chosen_records

def random_dataframe():
    """Generates a DataFrame with five random walk columns and a tag column"""
    arr = np.cumsum(np.random.randn(50, 5), axis=1)
    letters = itertools.combinations(string.ascii_uppercase, 3)
    columns = [''.join(triplet) for triplet in random.choices(list(letters), k=5)]
    tags = [chr(i + 65) for i in np.random.randint(0, 5, 50)]
    ix = pd.date_range(end=pd.Timestamp.utcnow(), periods=50, freq='90min')

    df = pd.DataFrame(arr, columns=columns)
    df['tag'] = tags
    df.index = ix
    return df

def random_string():
    return ''.join(random.choices(string.ascii_lowercase, k=random.randint(4, 10)))

def _gen_rand_str(size, chars=string.ascii_lowercase + string.digits):
        return ''.join(choices(chars, k=size))

def choice(self, ctx, *choices):
        """Picks randomly from the choices provided."""
        await ctx.send(random.choice(choices))

def choices(self, ctx, qty : int, *choices):
        """Picks `{qty}` times from the choices provided, with replacement."""
        await ctx.send(', '.join(random.choices(choices, k=qty)))

def sample(self, ctx, qty : int, *choices):
        """Picks `{qty}` times from the choices provided, without replacement."""
        count, remainder = divmod(qty, len(choices)) # Number of times through the choices, and number of choices afterward
        results = []
        for _ in range(count):
            results.extend(random.sample(choices, k=len(choices)))
        if remainder:
            results.extend(random.sample(choices, k=remainder))
        await ctx.send(', '.join(results))

def choices(population, weights=None, k=1):
    """
    Replacement for `random.choices()`, which is only available in Python 3.6+.
    TODO: drop once Python 3.6 is required by the sdk.
    """
    if weights and sum(weights) != 1:
        # Normalize the weights if needed, as numpy.random.choice requires so.
        weights = [float(i)/sum(weights) for i in weights]

    return numpy.random.choice(population, size=k, p=weights)

def random_string(length=10, lower=True, upper=False, digits=False):
        pool = ''
        if lower:
            pool += string.ascii_lowercase
        if upper:
            pool += string.ascii_uppercase
        if digits:
            pool += string.digits
        return ''.join(random.choices(pool, k=length))

def pick(elem, counters):
        weights = [(elem in v) * 0.5 + 0.5 for v in counters.values()]
        return random.choices(list(counters), weights)[0]

def _default_flip(self, ctx):
        """Flip called with no arguments"""
        side = random.choices(SIDES, WEIGHTS)[0]
        file = discord.File(f'data/images/coins/{side}.png', 'coin.png')

        embed = (discord.Embed(colour=ctx.bot.colour, description=f'...flipped **{side}**')
                 .set_author(name=ctx.author.display_name, icon_url=ctx.author.avatar_url)
                 .set_image(url='attachment://coin.png')
                 )

        await ctx.send(file=file, embed=embed)

def _flip_image(self, num_sides):
        images = {
            Side.heads: self._heads_image,
            Side.tails: self._tails_image,
        }
        stats = collections.Counter()

        root = num_sides ** 0.5
        height, width = round(root), int(math.ceil(root))

        sides = (random.choices(SIDES, WEIGHTS)[0] for _ in range(num_sides))

        size = self.image_size
        image = Image.new('RGBA', (width * size, height * size))

        for i, side in enumerate(sides):
            y, x = divmod(i, width)
            image.paste(images[side], (x * size, y * size))
            stats[side] += 1

        message = ' and '.join(pluralize(**{str(side)[:-1]: n}) for side, n in stats.items())

        f = io.BytesIO()
        image.save(f, 'png')
        f.seek(0)

        return message, discord.File(f, filename='flipcoins.png')

def add_mod():
    # adds a user to the list of moderators
    username = request.form["username"]
    user = User.query.filter(func.lower(User.username)
                             == func.lower(username)).first()
    if not user:
        return bad_request("user not found")
    user.form_mod = True
    run = True
    while run:
        # issue key
        key = "".join(
            random.choices(
                string.ascii_letters +
                string.digits,
                k=32))
        user.api_key = key
        try:
            db.session.add(user)
            db.session.commit()
            run = False
        except IntegrityError:  # check for uniqueness
            continue
    db.session.add(user)
    db.session.commit()
    url = url_for('settings', _external=True)
    subj = f"invitation to moderate {g.settings.site_title}"
    body = f"**gadzooks!** u/{g.user.username} has added you as a moderator of {g.settings.site_title}"
    body += f"\n\nclick [here]({url}) to view the site. mod tools will be visible at the top of the page."
    send_message(username, subj, body)
    return jsonify(status="OK"), 200

def username_from_repo(self, repo):
        """Generate a username for a git repo url

        e.g. minrk-binder-example-abc123
        from https://github.com/minrk/binder-example.git
        """
        # start with url path
        print
        if '://' not in repo and _ssh_repo_pat.match(repo):
            # ssh url
            path = repo.split(':', 1)[1]
        else:
            path = urlparse(repo).path

        prefix = path.strip('/').replace('/', '-').lower()

        if prefix.endswith('.git'):
            # strip trailing .git
            prefix = prefix[:-4]

        if len(prefix) > 32:
            # if it's long, truncate
            prefix = '{}-{}'.format(prefix[:15], prefix[-15:])

        # add a random suffix to avoid collisions for users on the same image
        return '{}-{}'.format(prefix, ''.join(random.choices(SUFFIX_CHARS, k=SUFFIX_LENGTH)))

def expectation_tv(self, data_list):
    N, F = read_data(data_list, self.nmix, self.ndim)
    nfiles = F.shape[0]
    nframes = N.sum()
    LU = np.zeros((self.nmix, self.tv_dim * (self.tv_dim + 1) // 2))
    RU = np.zeros((self.tv_dim, self.nmix * self.ndim))
    LLK = 0.
    T_invS = self.Tm / self.Sigma
    T_iS_Tt = self.comp_T_invS_Tt()
    parts = 2500  # modify this based on your HW resources (e.g., memory)
    nbatch = int(nfiles / parts + 0.99999)
    for batch in range(nbatch):
      start = batch * parts
      fin = min((batch + 1) * parts, nfiles)
      length = fin - start
      N1 = N[start:fin]
      F1 = F[start:fin]
      L1 = N1.dot(T_iS_Tt)
      B1 = F1.dot(T_invS.T)
      Ex = np.empty((length, self.tv_dim))
      Exx = np.empty((length, self.tv_dim * (self.tv_dim + 1) // 2))
      llk = np.zeros((length, 1))
      for ix in range(length):
        L = np.zeros((self.tv_dim, self.tv_dim))
        L[self.itril] = L1[ix]
        L += np.tril(L, -1).T + self.Im
        Cxx = inv(L)
        B = B1[ix][:, np.newaxis]
        this_Ex = Cxx.dot(B)
        llk[ix] = self.res_llk(this_Ex, B)
        Ex[ix] = this_Ex.T
        Exx[ix] = (Cxx + this_Ex.dot(this_Ex.T))[self.itril]
      RU += Ex.T.dot(F1)
      LU += N1.T.dot(Exx)
      LLK += llk.sum()
    self.Tm = None
    tmp_string = ''.join(random.choices(string.ascii_uppercase + string.digits, k=16))
    tmpfile = self.tmpdir + 'tmat_' + tmp_string + '.h5'
    h5write(tmpfile, LU, 'LU')
    return RU, LLK, nframes

def id3_version(self):
     return random.choices(
         [v for v, _ in self.VERSION_WEIGHTS],
         cum_weights=[w for _, w in self.VERSION_WEIGHTS])[0]