Download features.py from xcx0902/QOJDoudizhuModel: direct link, hf CLI and curl.
- Browser
- Download file 6.02 kB
-
https://huggingface.co/xcx0902/QOJDoudizhuModel/resolve/main/features.py
- Command line
-
hf download hf://xcx0902/QOJDoudizhuModel/features.py
-
curl -L -o features.py https://huggingface.co/xcx0902/QOJDoudizhuModel/resolve/main/features.py
6.02 kB
| """Suit-invariant public features; privileged labels live outside this module.""" | |
| import numpy as np | |
| import torch | |
| from qojbot import rules as R | |
| from qojbot.adapter import latest_log | |
| KINDS = ('pass', 'single', 'pair', 'trio', 'trio1', 'trio2', 'bomb', | |
| 'rocket', 'four2', 'four22', 'straight', 'pairs', 'plane', | |
| 'plane1', 'plane2', 'bid') | |
| ROLES = ('landlord', 'landlord_down', 'landlord_up') | |
| HISTORY = 96 | |
| ACTION_DIM = 30 + len(KINDS) + 4 | |
| HISTORY_DIM = 15 + len(KINDS) + 3 + 4 | |
| SCALAR_DIM = 15 | |
| def role_seats(landlord): | |
| return (landlord, (landlord + 1) % 3, (landlord + 2) % 3) | |
| def action_features(hand, moves): | |
| result = np.zeros((len(moves), ACTION_DIM), dtype=np.float32) | |
| own = np.asarray(R.counts(hand)) | |
| for index, move in enumerate(moves): | |
| cards = np.asarray(R.counts(move.code)) | |
| result[index, :15] = cards / 4 | |
| result[index, 15:30] = (own - cards) / 4 | |
| result[index, 30 + KINDS.index(move.kind)] = 1 | |
| result[index, -4:] = ((move.rank + 1) / 15, move.length / 12, | |
| move.size / 20, float(move.size == R.size(hand))) | |
| return torch.from_numpy(result) | |
| def encode_public(obs, state, moves): | |
| """Only Observation and actor's state are accepted, never engine hands.""" | |
| seats = role_seats(obs.landlord) | |
| role = seats.index(obs.seat) | |
| played = np.zeros((3, 15), dtype=np.float32) | |
| history = [] | |
| for event in latest_log(state): | |
| kind = event.get('kind') | |
| if kind not in ('play', 'pass', 'bid'): | |
| continue | |
| row = np.zeros(HISTORY_DIM, dtype=np.float32) | |
| actor = seats.index(event['seat']) | |
| if kind == 'play': | |
| cards = np.asarray(R.counts(R.from_cards(event['cards']))) | |
| played[actor] += cards | |
| row[:15] = cards / 4 | |
| pattern = event['pattern'] | |
| move_kind = pattern['type'] | |
| row[-4:-1] = ((pattern['rank'] + 1) / 15, | |
| pattern.get('len', 1) / 12, len(event['cards']) / 20) | |
| else: | |
| move_kind = kind | |
| if kind == 'bid': | |
| row[-1] = event['value'] / 3 | |
| row[15 + KINDS.index(move_kind)] = 1 | |
| row[15 + len(KINDS) + actor] = 1 | |
| history.append(row) | |
| last = np.zeros(15) if obs.last is None else R.counts(obs.last.code) | |
| # Online last patterns have code=0; recover actual cards from public history. | |
| if obs.last is not None: | |
| for event in reversed(latest_log(state)): | |
| if event.get('kind') == 'play': | |
| last = R.counts(R.from_cards(event['cards'])) | |
| break | |
| ranks = np.stack((R.counts(obs.hand), R.counts(obs.unseen), | |
| R.counts(obs.known_landlord), last, *played), axis=1) / 4 | |
| last_actor = [float(obs.last is not None and obs.last_seat == seat) for seat in seats] | |
| scalars = [obs.sizes[seat] / 20 for seat in seats] | |
| scalars += [obs.plays[seat] / 20 for seat in seats] | |
| scalars += [obs.bombs / 14, obs.passes / 2, float(obs.last is None)] | |
| scalars += last_actor + [float(state.get('bid', 1)) / 3, | |
| float(role != 0 and obs.last_seat == obs.teammate), 1.] | |
| rows = np.zeros((HISTORY, HISTORY_DIM), dtype=np.float32) | |
| mask = np.zeros(HISTORY, dtype=np.bool_) | |
| recent = history[-HISTORY:] | |
| if recent: | |
| rows[:len(recent)] = recent | |
| mask[:len(recent)] = True | |
| return {'ranks': torch.tensor(ranks, dtype=torch.float32), | |
| 'scalars': torch.tensor(scalars, dtype=torch.float32), | |
| 'history': torch.from_numpy(rows), 'mask': torch.from_numpy(mask), | |
| 'role': role, 'actions': action_features(obs.hand, moves)} | |
| def legal_bids(state): | |
| current = state.get('bid', 0) or 0 | |
| if type(current) is not int or current not in range(4): | |
| raise ValueError('invalid current bid') | |
| bids = list(range(current + 1, 4)) | |
| if not state.get('must_bid') or not bids: | |
| bids.insert(0, 0) | |
| return bids | |
| def encode_bid(state, bids): | |
| if state.get('seat') not in (0, 1, 2) or state.get('turn', state['seat']) != state['seat']: | |
| raise ValueError('invalid bidding turn') | |
| hand = R.from_cards(state['hand']) | |
| if R.size(hand) != 17: | |
| raise ValueError('bidding requires 17 cards') | |
| ranks = torch.zeros(15, 7) | |
| ranks[:, 0] = torch.tensor(R.counts(hand)) / 4 | |
| ranks[:, 1] = torch.tensor(R.counts(R.DECK - hand)) / 4 | |
| scalars = torch.zeros(SCALAR_DIM) | |
| scalars[:3] = 17 / 20 | |
| scalars[12] = (state.get('bid', 0) or 0) / 3 | |
| scalars[13] = float(bool(state.get('must_bid'))) | |
| history = torch.zeros(HISTORY, HISTORY_DIM) | |
| mask = torch.zeros(HISTORY, dtype=torch.bool) | |
| events = [event for event in latest_log(state) if event.get('kind') == 'bid'][-HISTORY:] | |
| for index, event in enumerate(events): | |
| history[index, 15 + KINDS.index('bid')] = 1 | |
| history[index, 15 + len(KINDS) + (event['seat'] - state['seat']) % 3] = 1 | |
| history[index, -1] = event['value'] / 3 | |
| mask[index] = True | |
| actions = torch.zeros(len(bids), ACTION_DIM) | |
| actions[:, 15:30] = ranks[:, 0] | |
| actions[:, 30 + KINDS.index('bid')] = 1 | |
| actions[:, -1] = torch.tensor(bids) / 3 | |
| return {'ranks': ranks, 'scalars': scalars, 'history': history, | |
| 'mask': mask, 'role': 3, 'actions': actions} | |
| def collate(samples, device): | |
| result = {key: torch.stack([sample[key] for sample in samples]).to( | |
| device=device, dtype=torch.bool if key == 'mask' else torch.float32) | |
| for key in ('ranks', 'scalars', 'history', 'mask')} | |
| result['role'] = torch.tensor([sample['role'] for sample in samples], | |
| device=device, dtype=torch.long) | |
| result['actions'] = torch.cat([sample['actions'] for sample in samples]).to( | |
| device=device, dtype=torch.float32) | |
| result['owner'] = torch.repeat_interleave( | |
| torch.arange(len(samples)), torch.tensor([len(sample['actions']) for sample in samples])).to(device) | |
| return result | |