Coverage for thb/actions.py: 24%

Hot-keys on this page

r m x p toggle line displays

j k next/prev highlighted chunk

0 (zero) top of page

1 (one) first highlighted chunk

1# -*- coding: utf-8 -*-

2from __future__ import annotations

4# -- stdlib --

5from collections import defaultdict

6from copy import copy

7from dataclasses import dataclass

8from typing import Any, Dict, List, Optional, Sequence, Set, TYPE_CHECKING, Tuple, Type, Union, cast

9import logging

11# -- third party --

12from typing_extensions import Literal, Protocol

14# -- own --

15from game.base import ActionShootdown, EventArbiter, GameViralContext, InputTransaction, Player

16from game.base import sync_primitive

17from thb.cards.base import Card, CardList, PhysicalCard, Skill, VirtualCard

18from thb.common import PlayerRole

19from thb.inputlets import ActionInputlet, ChoosePeerCardInputlet

20from thb.mode import THBAction, THBEventHandler, THBattle

21from utils.check import CheckFailed, check, check_type

22from utils.misc import BatchList

24# -- typing --

25if TYPE_CHECKING:

26 from thb.characters.base import Character # noqa: F401

29# -- code --

30log = logging.getLogger('THBattle_Actions')

33# ------------------------------------------

34# aux functions

36def ttags(actor):

37 tags = actor.tags

38 tc = tags['turn_count']

39 return tags.setdefault('turn_tags:%s' % tc, defaultdict(int))

42class CardChooser(Protocol):

43 game: THBattle

44 card_usage: str

46 def cond(self, cards: Sequence[Card]) -> bool: ... 46 ↛ exitline 46 didn't return from function 'cond'

49class CharacterChooser(Protocol):

50 game: THBattle

52 def choose_player_target(self, pl: Sequence[Character]) -> Tuple[List[Character], bool]: ... 52 ↛ exitline 52 didn't return from function 'choose_player_target'

55def ask_for_action(initiator: Union[CardChooser, CharacterChooser],

56 actors: List[Character],

57 categories: Sequence[str],

58 candidates: Sequence[Character],

59 timeout: Optional[int] = None,

60 trans: Optional[InputTransaction] = None,

61 ) -> Tuple[Optional[Character], Optional[Tuple[List[Card], List[Character]]]]:

62 # initiator: Action or EH requesting this

63 # actors: players involved

64 # categories: card categories, eg: ['cards', 'showncards']

65 # candidates: players can be selection target, eg: g.players

67 assert categories or candidates

68 assert actors

70 timeout = timeout or 25

72 from thb.cards.base import VirtualCard

74 g = cast(THBattle, initiator.game)

76 ilet = ActionInputlet(initiator, categories, candidates)

78 @ilet.with_post_process

79 def process(actor: Character, rst):

80 usage = getattr(initiator, 'card_usage', 'none')

81 try:

82 check(rst)

84 skills: List[Type[Skill]]

85 rawcards: List[Card]

86 characters: List[Character]

87 params: Dict[str, Any]

89 skills, rawcards, characters, params = rst

90 [check(not c.detached) for c in rawcards]

91 [check(actor.has_skill(s)) for s in skills] # has_skill may be hooked

93 if skills:

94 cards = [skill_wrap(actor, skills, rawcards, params)]

95 usage = cards[0].usage if usage == 'launch' else usage

96 assert usage != 'none', (cards[0], cards[0].usage)

97 else:

98 cards = rawcards

99 usage = 'launch'

100

101 if categories:

102 if len(cards) == 1 and cards[0].is_card(VirtualCard):

103 def walk(c: Card):

104 if not c.is_card(VirtualCard): return

105 if getattr(c, 'no_reveal', False): return

106

107 c = cast(VirtualCard, c)

108

109 g.players.player.reveal(c.associated_cards)

110 for c1 in c.associated_cards:

111 walk(c1)

112

113 walk(cards[0])

114 check(skill_check(cards[0]))

115

116 else:

117 if not getattr(initiator, 'no_reveal', False):

118 g.players.player.reveal(cards)

119

120 check(cast(CardChooser, initiator).cond(cards))

121 assert not (usage == 'none' and rawcards), (skills, rawcards, characters, params) # should not pass check

122 else:

123 cards = []

124

125 if candidates:

126 characters, valid = cast(CharacterChooser, initiator).choose_player_target(characters)

127 check(valid)

128

129 ask_for_action_verify = getattr(initiator, 'ask_for_action_verify', None)

130

131 if ask_for_action_verify:

132 check(ask_for_action_verify(actor, cards, characters))

133

134 return cards, characters, params

135

136 except CheckFailed:

137 return None

138

139 p, rst = g.user_input(actors, ilet, timeout=timeout, type='any', trans=trans)

140 if rst:

141 cards, characters, params = rst

142

143 if len(cards) == 1 and cards[0].is_card(VirtualCard):

144 g.deck.register_vcard(cards[0])

145

146 if not cards and not characters:

147 return p, None

148

149 [c.detach() for c in VirtualCard.unwrap(cards)]

150

151 return p, (cards, characters)

152 else:

153 return None, None

154

155

156def user_choose_cards(initiator: CardChooser,

157 actor: Character,

158 categories: Sequence[str],

159 timeout: Optional[int] = None,

160 trans: Optional[InputTransaction] = None,

161 ) -> Optional[List[Card]]:

162 check_type([str, ...], categories)

163

164 _, rst = ask_for_action(initiator, [actor], categories, (), timeout=timeout, trans=trans)

165 if not rst:

166 return None

167

168 return rst[0] # cards

169

170

171def user_choose_players(initiator: CharacterChooser,

172 actor: Character,

173 candidates: List[Character],

174 timeout: Optional[int] = None,

175 trans: Optional[InputTransaction] = None,

176 ) -> Optional[List[Character]]:

177 _, rst = ask_for_action(initiator, [actor], (), candidates, timeout=timeout, trans=trans)

178 if not rst:

179 return None

180

181 return rst[1] # players

182

183

184def random_choose_card(g: THBattle, cardlists: Sequence[Sequence]):

185 from itertools import chain

186 allcards = list(chain.from_iterable(cardlists))

187 if not allcards:

188 return None

189

190 c = g.random.choice(allcards)

191 v = sync_primitive(c.sync_id, g.players.player)

192 c = g.deck.lookup(v)

193 assert c

194 c.detach()

195 return c

196

197

198def skill_wrap(actor: Character, skills: List[Type[Skill]], cards: List[Card], params: Dict[str, Any]):

199 assert skills

200 for skill_cls in skills:

201 card = skill_cls.wrap(cards, actor, params)

202 cards = [card]

203

204 return cards[0]

205

206

207def skill_check(wrapped):

208 from thb.cards.base import Skill

209

210 try:

211 check(wrapped.check())

212 for c in wrapped.associated_cards:

213 if c.is_card(Skill):

214 check(c.character is wrapped.character)

215 check(skill_check(c))

216 else:

217 check(c.resides_in.owner is wrapped.character)

218

219 return True

220

221 except CheckFailed:

222 return False

223

224

225COMMON_EVENT_HANDLERS: Set[Type[THBEventHandler]] = set()

226

227

228def register_eh(cls):

229 COMMON_EVENT_HANDLERS.add(cls)

230 return cls

231

232

233# ------------------------------------------

234class GenericAction(THBAction):

235 pass

236

237

238class UserAction(THBAction): # card/character skill actions

239 target_list: List[Character]

240 associated_card: Card

241

242

243@dataclass

244class CardMigration:

245 trans: MigrateCardsTransaction

246 cards: Sequence[Card]

247 to: CardList

248 unwrap: bool = True # will cards be unwrapped?

249 direction: Literal['front', 'back'] = 'front'

250

251

252@dataclass

253class CardMovement:

254 trans: MigrateCardsTransaction

255 card: Card

256 fr: CardList

257 to: CardList

258 direction: Literal['front', 'back'] = 'front'

259

260

261class MigrateCardsTransaction(GameViralContext):

262 game: THBattle

263 action: THBAction

264 migrations: List[CardMigration]

265 movements: List[CardMovement]

266

267 def __init__(self, action: Optional[THBAction] = None):

268 g = self.game

269 self.action = cast(THBAction, action or g.action_stack[-1])

270

271 if self.action is g.action_stack[-1]:

272 # Ensure no card movements in EventHandlers!

273 assert self.action is g.hybrid_stack[-1], (self.action, g.hybrid_stack[-1])

274

275 self.cancelled = False

276 self.migrations = []

277 self.movements = []

278

279 def __repr__(self):

280 return 'MCT'

281

282 def add_step(self, m: CardMigration) -> None:

283 self.migrations.append(m)

284

285 def __enter__(self):

286 return self

287

288 def __exit__(self, *excinfo):

289 if not self.cancelled:

290 self.commit()

291 else:

292 log.debug('migrate_cards cancelled: %s', self.movements)

293

294 def commit(self):

295 g = self.game

296

297 seen: Set[Card] = set()

298

299 for m in self.migrations:

300 cards = VirtualCard.unwrap(m.cards, include_vcards=True)

301 n = 0

302 for c in cards:

303 if c in seen:

304 continue

305

306 if isinstance(c, VirtualCard):

307 if not m.unwrap:

308 assert m.to.owner, 'Invalid move!'

309 sp = m.to.owner.special

310 for ac in c.associated_cards:

311 assert ac not in seen

312 if not isinstance(ac, VirtualCard):

313 self.movements.append(CardMovement(

314 trans=self, card=ac, fr=ac.resides_in, to=sp, direction='front'

315 ))

316 ac.move_to(sp)

317 seen.add(ac)

318 else:

319 c.detach()

320 c.unwrapped = True

321 continue

322

323 self.movements.append(CardMovement(

324 trans=self, card=c, fr=c.resides_in, to=m.to, direction=m.direction

325 ))

326 c.move_to(m.to)

327 seen.add(c)

328 n += 1

329

330 if m.direction == 'back':

331 m.to.rotate(n)

332

333 g.emit_event('post_card_migration', self)

334

335

336def migrate_cards(cards: Sequence[Card],

337 to: CardList,

338 unwrap: bool = False,

339 direction: Literal['front', 'back'] = 'front',

340 trans: Optional[MigrateCardsTransaction] = None,

341 ):

342 '''

343 cards: cards to move around

344 to: destination card list

345 unwrap: tear down VirtualCard wrapping, preserve PhysicalCard only

346 trans: associated MigrateCardsTransaction

347 '''

348 if not trans:

349 with MigrateCardsTransaction() as t:

350 migrate_cards(cards, to, unwrap, direction, t)

351 return not t.cancelled

352

353 assert to is not None

354

355 if to.owner and to.owner.dead:

356 # do not migrate cards to dead character

357 trans.cancelled = True

358 return

359

360 trans.add_step(CardMigration(

361 trans=trans,

362 cards=cards,

363 to=to,

364 unwrap=unwrap,

365 direction=direction,

366 ))

367

368

369class PostCardMigrationHandler(EventArbiter):

370 game: THBattle

371 handlers: Sequence[THBEventHandler]

372

373 interested = ['post_card_migration']

374

375 def handle(self, evt_type, arg):

376 if evt_type != 'post_card_migration': return arg

377

378 g = self.game

379 act = arg.action

380 tgt = act.target or act.source or g.players[0]

381

382 n = len(g.players)

383 try:

384 idx = g.players.index(tgt) - n

385 except ValueError:

386 # Character died or switched out or whatever

387 return arg

388 for i in range(idx, idx + n):

389 for eh in self.handlers:

390 g.dispatcher.handle_single_event(eh, g.players[i], arg)

391

392 return arg

393

394

395def detach_cards(cards: Sequence[Card], trans=None):

396 if not trans:

397 with MigrateCardsTransaction() as t:

398 return detach_cards(cards, t)

399

400 for c in VirtualCard.unwrap(cards):

401 c.detach()

402

403 g = trans.game

404 g.emit_event('detach_cards', (trans, cards))

405

406

407class DeadDropCards(GenericAction):

408 def apply_action(self):

409 tgt = self.target

410 g = self.game

411

412 others = g.players.exclude(tgt)

413 for cl in tgt.lists:

414 if not cl: continue

415 others.reveal(list(cl))

416 g.process_action(DropCards(tgt, tgt, cl))

417 assert not cl

418

419 return True

420

421

422class PlayerDeath(GenericAction):

423 # FIXME: should be `CharacterDeath`

424 def apply_action(self) -> bool:

425 tgt = self.target

426 tgt.dead = True

427 tgt.skills[:] = []

428 tgt.tags.clear()

429 g = self.game

430 g.process_action(DeadDropCards(tgt, tgt))

431 return True

432

433

434class PlayerRevive(GenericAction):

435 def __init__(self, source, target, hp):

436 self.source = source

437 self.target = target

438 self.hp = hp

439

440 def apply_action(self):

441 tgt = self.target

442 assert tgt.dead

443

444 tgt.dead = False

445 tgt.maxlife = tgt.__class__.maxlife

446 tgt.skills = list(tgt.__class__.skills)

447

448 tgt.life = min(tgt.maxlife, self.hp)

449 return True

450

451 def is_valid(self):

452 return self.target.dead

453

454

455class TryRevive(GenericAction):

456 def __init__(self, target: Character, dmgact: BaseDamage):

457 self.source = self.target = target

458 self.dmgact = dmgact

459 self.revived_by: Optional[Character] = None

460 if target.dead:

461 log.error('TryRevive buggy condition, __init__')

462 return

463

464 def apply_action(self) -> bool:

465 tgt = self.target

466

467 if tgt.dead:

468 log.error('TryRevive buggy condition, apply')

469 import traceback

470 traceback.print_stack()

471 return False

472

473 g = self.game

474 from thb.cards.basic import AskForHeal

475 for p in g.players.rotate_to(tgt):

476 while True:

477 if p.dead:

478 break

479

480 if g.process_action(AskForHeal(tgt, p)):

481 if tgt.life > 0:

482 self.revived_by = p

483 return True

484 continue

485

486 break

487

488 return tgt.life > 0

489

490 def is_valid(self):

491 tgt = self.target

492 return not tgt.dead and tgt.maxlife > 0

493

494

495class BaseDamage(GenericAction):

496 def __init__(self, source, target, amount=1):

497 self.source = source

498 self.target = target

499 self.amount = amount

500

501 def apply_action(self):

502 tgt = self.target

503 tgt.life -= self.amount

504 return True

505

506 def is_valid(self):

507 return not self.target.dead

508

509

510class Damage(BaseDamage):

511 pass

512

513

514class LifeLost(BaseDamage):

515 def __init__(self, source, target, amount=1):

516 self.source = None # type: ignore

517 self.target = target

518 self.amount = amount

519

520

521class MaxLifeChange(GenericAction):

522 def __init__(self, source: Character, target: Character, amount):

523 self.source = source

524 self.target = target

525 self.amount = amount

526

527 def apply_action(self):

528 src = self.source

529 tgt = self.target

530 g = self.game

531 tgt.maxlife += self.amount

532

533 if tgt.life > tgt.maxlife:

534 g.process_action(

535 LifeLost(src, tgt, abs(tgt.life - tgt.maxlife))

536 )

537 assert tgt.life == tgt.maxlife

538

539 assert tgt.maxlife or tgt.dead

540

541 return True

542

543 def is_valid(self):

544 return not self.target.dead

545

546

547# ---------------------------------------------------

548

549class DropCards(GenericAction):

550 def __init__(self, source, target, cards):

551 self.source = source

552 self.target = target

553 self.cards = cards

554

555 def apply_action(self):

556 g = self.game

557 target = self.target

558 cards = self.cards

559 assert all(c.resides_in.owner in (target, None) for c in cards), 'WTF?!'

560 migrate_cards(cards, g.deck.droppedcards, unwrap=True)

561

562 return True

563

564 def is_valid(self):

565 return True

566

567

568class UseCard(GenericAction):

569

570 def __init__(self, target, card):

571 self.source = self.target = target

572 self.card = card

573

574 def apply_action(self):

575 g = self.game

576 tgt = self.target

577 c = self.card

578 act = getattr(c, 'use_action', None)

579 if act:

580 return g.process_action(act(tgt, c))

581 else:

582 migrate_cards([c], g.deck.droppedcards, unwrap=True)

583 return True

584

585 def can_fire(self):

586 c = self.card

587 act = getattr(c, 'use_action', None)

588 if act:

589 return act(self.target, self.card).can_fire()

590 else:

591 return True

592

593

594class AskForCard(GenericAction):

595 card_usage = 'drop'

596

597 def __init__(self, source: Character, target: Character, card_cls: Type[PhysicalCard], categories: Sequence[str] = ('cards', 'showncards')):

598 self.source = source

599 self.target = target

600 self.card_cls = card_cls

601 self.categories = categories

602

603 self.card: Optional[Card] = None

604

605 def apply_action(self):

606 target = self.target

607

608 if not self.card: # ask if not already provided

609 cards = user_choose_cards(self, target, self.categories)

610 if not cards or len(cards) != 1:

611 self.card = None

612 return False

613

614 self.card = cards[0]

615

616 return self.process_card(self.card)

617

618 def cond(self, cl):

619 from thb.cards.base import VirtualCard

620 t = self.target

621 return (

622 len(cl) == 1 and

623 cl[0].is_card(self.card_cls) and

624 (cl[0].is_card(VirtualCard) or cl[0].resides_in.owner is t)

625 )

626

627 def process_card(self, card):

628 raise NotImplementedError

629

630

631class ActiveDropCards(GenericAction):

632 card_usage: str

633

634 def __init__(self, source: Character, target: Character, dropn: int) -> None:

635 self.source = source

636 self.target = target

637 self.dropn = dropn

638 self.cards: List[Card] = []

639

640 def apply_action(self) -> bool:

641 tgt = self.target

642 if tgt.dead: return False

643 n = self.dropn

644 if n <= 0: return True

645

646 g = self.game

647 cards = user_choose_cards(self, tgt, ('cards', 'showncards'))

648 if cards:

649 g.process_action(DropCards(tgt, tgt, cards=cards))

650 else:

651 from itertools import chain

652 cards = list(chain(tgt.cards, tgt.showncards))[min(-n, 0):]

653 g.players.player.reveal(cards)

654 g.process_action(DropCards(tgt, tgt, cards=cards))

655

656 self.cards = cards

657 return True

658

659 def cond(self, cards: Sequence[Card]) -> bool:

660 tgt = self.target

661 if not len(cards) == self.dropn:

662 return False

663

664 if not all(c.resides_in in (tgt.cards, tgt.showncards) for c in cards):

665 return False

666

667 from thb.cards.base import Skill

668 if any(c.is_card(Skill) for c in cards):

669 return False

670

671 return True

672

673

674class DropCardStage(ActiveDropCards):

675 def __init__(self, target):

676 dropn = len(target.cards) + len(target.showncards) - target.life

677 ActiveDropCards.__init__(self, target, target, dropn)

678

679

680class BaseDrawCards(GenericAction):

681 def __init__(self, target, amount=2, back=False):

682 self.source = self.target = target

683 self.amount = amount

684 self.back = back

685

686 def apply_action(self):

687 g = self.game

688 target = self.target

689

690 if self.back:

691 g.deck.getcards(self.amount) # forcing dropped cards to join if cards in deck are insufficient

692 g.deck.cards.rotate(self.amount)

693

694 cards = g.deck.getcards(self.amount)

695

696 target.reveal(cards)

697 migrate_cards(cards, target.cards)

698 self.cards = cards

699 return True

700

701 def is_valid(self):

702 return not self.target.dead

703

704

705class DrawCards(BaseDrawCards):

706 pass

707

708

709class DistributeCards(BaseDrawCards):

710 pass

711

712

713class DrawCardStage(DrawCards):

714 pass

715

716

717class PutBack(GenericAction):

718 def __init__(self, source, cards, direction: Literal['front', 'back'] = 'front'):

719 self.source = self.target = source

720 self.cards = cards

721 self.direction = direction

722

723 def apply_action(self):

724 g = self.game

725 cards = self.cards

726

727 migrate_cards(cards, g.deck.cards, direction=self.direction)

728

729 return True

730

731

732class LaunchCard(GenericAction):

733 def __init__(self, src: Character,

734 target_list: Sequence[Character],

735 card: Card,

736 action: Optional[UserAction] = None,

737 bypass_check=False):

738 self.force_action = action

739 bypass_check = bool(action) or bypass_check

740 self.bypass_check = bypass_check

741 if bypass_check:

742 tl, tl_valid = list(target_list), True

743 else:

744 tl, tl_valid = card.target(src, target_list)

745

746 self.source, self.target_list, self.card, self.tl_valid = src, tl, card, tl_valid

747 self.target = target_list[0] if target_list else src

748

749 def apply_action(self) -> bool:

750 card = self.card

751 target_list = self.target_list

752 if not card: return False

753

754 if not self.force_action and not card.associated_action:

755 return False

756

757 g = self.game

758 src = self.source

759 drop = card.usage == 'drop'

760 try:

761 if drop: # should drop before action

762 g.process_action(DropCards(src, src, cards=[card]))

763

764 elif not getattr(card, 'no_drop', False):

765 detach_cards([card]) # emit events

766

767 else:

768 card.detach()

769

770 _, tl = g.emit_event('choose_target', (self, target_list))

771 assert _ is self

772 card = self.card # card may be touched

773

774 if not tl:

775 return True

776

777 if self.force_action:

778 a = self.force_action

779 else:

780 tgt = tl[0] if tl else src

781 assert card.associated_action

782 a = card.associated_action(source=src, target=tgt)

783 a.target_list = tl

784

785 a.associated_card = card

786 self.card_action = a

787

788 _ = g.emit_event('post_choose_target', (self, tl))

789 card = self.card # card may be touched

790 assert _ == (self, tl)

791

792 g.process_action(a)

793 card = self.card # card may be touched

794

795 return True

796 finally:

797 # card may be touched

798 # e.g. Kyouko Echo will move this card, left a placeholder here.

799 card = self.card

800

801 if not drop and card.detached:

802 # card/skill still in disputed state,

803 # means no actions have done anything to the card/skill,

804 # drop it

805 if not getattr(card, 'no_drop', False) and not card.unwrapped:

806 migrate_cards([card], g.deck.droppedcards, unwrap=True)

807

808 else:

809 from thb.cards.base import VirtualCard

810 for c in VirtualCard.unwrap([card]):

811 if c.detached: c.attach()

812

813 return True

814

815 def is_valid(self) -> bool:

816 if self.bypass_check:

817 return True

818

819 if not self.tl_valid:

820 log.debug('LaunchCard.tl_valid FALSE')

821 return False

822

823 card = self.card

824 if not card:

825 log.debug('LaunchCard.card FALSE')

826 return False

827

828 g = self.game

829 src = self.source

830

831 dist = self.calc_distance(g, src, card)

832 if not all([dist[p] <= 0 for p in self.target_list]):

833 log.debug('LaunchCard: does not fulfill distance constraint')

834 return False

835

836 cls = card.associated_action

837 if not cls:

838 return False

839

840 tl = self.target_list

841 target = tl[0] if tl else src

842 act = cls(source=src, target=target)

843 act.associated_card = card

844 act.target_list = tl

845 try:

846 act.action_shootdown_exception()

847 except Exception:

848 log.debug('LaunchCard card_action.can_fire() FALSE')

849 raise

850

851 return True

852

853 @classmethod

854 def calc_distance(cls, g, src, card):

855 dist = cls.calc_raw_distance(g, src, card)

856 card_dist = getattr(card, 'distance', 1000)

857 for p in dist:

858 dist[p] -= card_dist

859 g.emit_event('post_calcdistance', (src, card, dist))

860 return dist

861

862 @classmethod

863 def calc_raw_distance(cls, g, src, card):

864 dist = cls.calc_base_distance(g, src)

865 g.emit_event('calcdistance', (src, card, dist))

866 return dist

867

868 @classmethod

869 def calc_base_distance(cls, g, src):

870 pl = [p for p in g.players if not p.dead or p is src]

871 loc = pl.index(src)

872 n = len(pl)

873 dist = {

874 p: min(abs(i), n - abs(i))

875 for p, i in zip(pl, range(-loc, -loc + n))

876 }

877 return dist

878

879

880class ActionStageLaunchCard(LaunchCard):

881 pass

882

883

884class BaseActionStage(GenericAction):

885 card_usage = 'launch'

886 one_shot: bool

887 launch_card_cls: Type[LaunchCard]

888

889 def __init__(self, target):

890 self.source = self.source = target

891 self.target = target

892 self.in_user_input = False

893 self._force_break = False

894 self.action_count = 0

895

896 def apply_action(self):

897 g = self.game

898 target = self.target

899 if target.dead: return False

900

901 try:

902 while not target.dead:

903 try:

904 g.emit_event('action_stage_action', target)

905 self.in_user_input = True

906 with InputTransaction('ActionStageAction', [target]) as trans:

907 p, rst = ask_for_action(

908 self, [target], ('cards', 'showncards'), g.players, trans=trans

909 )

910 check(p is target)

911 assert rst

912 finally:

913 self.in_user_input = False

914

915 cards, target_list = rst

916 g.players.reveal(cards)

917 card = cards[0]

918

919 lc = self.launch_card_cls(target, target_list, card)

920 if not g.process_action(lc):

921 if lc.invalid:

922 log.debug('ActionStage: LaunchCard invalid.')

923 check(False)

924

925 self.action_count += 1

926

927 if self.one_shot or self._force_break:

928 break

929

930 except CheckFailed:

931 pass

932

933 return True

934

935 @staticmethod

936 def force_break(g):

937 for a in g.action_stack:

938 if isinstance(a, ActionStage):

939 a._force_break = True

940 log.debug('ActioStage: force_break requested')

941 break

942

943 def cond(self, cl):

944 if not cl: return False

945

946 tgt = self.target

947 if not len(cl) == 1:

948 return False

949

950 c = cl[0]

951 return (

952 c.is_card(Skill) or c.resides_in in (tgt.cards, tgt.showncards)

953 ) and bool(c.associated_action)

954

955 def ask_for_action_verify(self, p, cl, tl):

956 assert len(cl) == 1

957 return self.launch_card_cls(p, tl, cl[0]).can_fire()

958

959 def choose_player_target(self, tl):

960 return tl, True

961

962

963class ActionStage(BaseActionStage):

964 one_shot = False

965 launch_card_cls = ActionStageLaunchCard

966

967

968@register_eh

969class ShuffleHandler(THBEventHandler):

970 interested = ['action_after', 'action_before', 'action_stage_action', 'user_input_start']

971

972 def handle(self, evt_type, arg):

973 g = self.game

974 if evt_type == 'action_stage_action': 974 ↛ 975line 974 didn't jump to line 975, because the condition on line 974 was never true

975 self.do_shuffle(g, g.players)

976

977 elif evt_type in ('action_before', 'action_after') and isinstance(arg, ActionStage): 977 ↛ 978line 977 didn't jump to line 978, because the condition on line 977 was never true

978 self.do_shuffle(g, g.players)

979

980 elif evt_type == 'user_input_start': 980 ↛ 981line 980 didn't jump to line 981, because the condition on line 980 was never true

981 trans, ilet = arg

982 if isinstance(ilet, ChoosePeerCardInputlet):

983 self.do_shuffle(g, [ilet.target])

984

985 return arg

986

987 @staticmethod

988 def do_shuffle(g, pl):

989 log.debug('Shuffling cards')

990

991 for p in pl:

992 if not p.cards: continue

993 if any([c.is_card(VirtualCard) for c in p.cards]):

994 log.warning('VirtualCard in cards of %s, not shuffling.' % repr(p))

995 continue

996

997 g.deck.shuffle(p.cards)

998

999

1000class FatetellStage(GenericAction):

1001 def __init__(self, target):

1002 self.target = target

1003

1004 def apply_action(self):

1005 g = self.game

1006 target = self.target

1007 if target.dead: return False

1008 ft_cards = target.fatetell

1009 while ft_cards:

1010 if target.dead: break

1011 card = ft_cards[-1] # what comes last, launches first.

1012 g.process_action(LaunchFatetellCard(target, card))

1013

1014 return True

1015

1016

1017class BaseFatetell(GenericAction):

1018 type: str

1019

1020 def __init__(self, target, cond):

1021 self.source = target

1022 self.target = target

1023 self.cond = cond

1024 self.initiator = self.game.hybrid_stack[-1]

1025 self.card_manipulator = self

1026

1027 def apply_action(self):

1028 g = self.game

1029 card, = g.deck.getcards(1)

1030 g.players.reveal(card)

1031 self.card = card

1032 detach_cards([card])

1033 g.emit_event(self.type, self)

1034 return self.succeeded

1035

1036 def set_card(self, card, card_manipulator):

1037 self.card = card

1038 self.card_manipulator = card_manipulator

1039

1040 @property

1041 def succeeded(self):

1042 # This is necessary, for ui

1043 return self.cond(self.card)

1044

1045

1046class Fatetell(BaseFatetell):

1047 type = 'fatetell'

1048

1049

1050class TurnOverCard(BaseFatetell):

1051 type = 'turnover'

1052

1053

1054class FatetellMalleateHandler(EventArbiter):

1055 interested = ['fatetell']

1056 game: THBattle

1057

1058 def handle(self, evt_type, data):

1059 if evt_type != 'fatetell': return data

1060

1061 g = self.game

1062 try:

1063 tgt = PlayerTurn.get_current(g).target

1064 except IndexError:

1065 for a in g.action_stack:

1066 if isinstance(a, LaunchCard):

1067 tgt = a.source

1068 break

1069 else:

1070 # No one's turn

1071 # Only observed in `character_debut` events

1072 tgt = g.players[0]

1073

1074 n = len(g.players)

1075 idx = g.players.index(tgt) - n

1076 for i in range(idx, idx + n):

1077 for eh in self.handlers:

1078 g.dispatcher.handle_single_event(eh, g.players[i], data)

1079

1080 return data

1081

1082

1083class FatetellAction(GenericAction):

1084 fatetell_target = None

1085

1086 def apply_action(self):

1087 assert self.fatetell_target is not None, 'Should specify fatetell_target!'

1088

1089 ft = Fatetell(self.fatetell_target, self.fatetell_cond)

1090 g = self.game

1091 g.process_action(ft)

1092

1093 if not ft.cancelled:

1094 rst = self.fatetell_action(ft)

1095

1096 c = ft.card

1097 if (isinstance(c, PhysicalCard) and c.detached) or isinstance(c, VirtualCard):

1098 with MigrateCardsTransaction(ft.card_manipulator) as trans:

1099 migrate_cards([c], g.deck.droppedcards, unwrap=True, trans=trans)

1100

1101 return rst

1102

1103 return False

1104

1105 def fatetell_action(self, ft):

1106 raise Exception('Implement fatetell_action!')

1107

1108 def fatetell_cond(self, card):

1109 raise Exception('Implement fatetell_cond!')

1110

1111 def fatetell_postprocess(self):

1112 pass

1113

1114

1115class LaunchFatetellCard(GenericAction):

1116 def __init__(self, target, card):

1117 self.source = target

1118 self.target = target

1119 self.card = card

1120

1121 def apply_action(self):

1122 g = self.game

1123 target = self.target

1124 card = self.card

1125 act = card.delayed_action

1126 assert act

1127 a = act(source=target, target=target)

1128 a.associated_card = card

1129 self.card_action = a

1130 g.process_action(a)

1131 a.fatetell_postprocess()

1132 return True

1133

1134

1135class ForEach(UserAction):

1136 action_cls: Type[THBAction]

1137 include_dead = False

1138

1139 def prepare(self):

1140 pass

1141

1142 def cleanup(self):

1143 pass

1144

1145 def __init__(self, source, target):

1146 self.source = source

1147 self.target = target

1148

1149 def apply_action(self):

1150 tl = self.target_list

1151 source = self.source

1152 card = self.associated_card

1153 g = self.game

1154

1155 try:

1156 self.prepare()

1157 for t in tl:

1158 if t.dead and not self.include_dead:

1159 continue

1160 a = self.action_cls(source, t)

1161 a.associated_card = card

1162 a._['for_each'] = self

1163 g.process_action(a)

1164

1165 finally:

1166 self.cleanup()

1167

1168 return True

1169

1170 @classmethod

1171 def get_actual_action(self, act: THBAction):

1172 assert isinstance(act, THBAction)

1173 return act._.get('for_each')

1174

1175 @classmethod

1176 def is_group_effect(self, act: THBAction):

1177 return getattr(self.get_actual_action(act), 'group_effect', False)

1178

1179

1180class PrepareStage(GenericAction):

1181 def __init__(self, target):

1182 self.source = target

1183 self.target = target

1184

1185 def apply_action(self):

1186 return True

1187

1188

1189class FinalizeStage(GenericAction):

1190 def __init__(self, target):

1191 self.source = target

1192 self.target = target

1193

1194 def apply_action(self):

1195 return True

1196

1197

1198class PlayerTurn(GenericAction):

1199 def __init__(self, target: Character):

1200 self.source = self.target = target

1201 self.pending_stages = [

1202 PrepareStage,

1203 FatetellStage,

1204 DrawCardStage,

1205 ActionStage,

1206 DropCardStage,

1207 FinalizeStage,

1208 ]

1209

1210 def apply_action(self) -> bool:

1211 g = self.game

1212 p = self.target

1213 p.tags['turn_count'] += 1

1214

1215 while self.pending_stages: 1215 ↛ 1220line 1215 didn't jump to line 1220, because the condition on line 1215 was never false

1216 stage = self.pending_stages.pop(0)

1217 self.current_stage = cs = stage(p)

1218 g.process_action(cs)

1219

1220 return True

1221

1222 @staticmethod

1223 def get_current(g: THBattle) -> PlayerTurn:

1224 for act in g.action_stack: 1224 ↛ 1228line 1224 didn't jump to line 1228, because the loop on line 1224 didn't complete

1225 if isinstance(act, PlayerTurn):

1226 return act

1227

1228 raise IndexError('Could not find current turn!')

1229

1230

1231class DummyAction(GenericAction):

1232 def __init__(self, source, target, result=True):

1233 self.source, self.target, self.result = \

1234 source, target, result

1235

1236 def apply_action(self):

1237 return self.result

1238

1239

1240class RevealRole(THBAction):

1241

1242 def __init__(self, p: Player, role: PlayerRole, to: Sequence[Player]):

1243 self.player = p

1244 self.role = role

1245 self.to = to

1246

1247 def apply_action(self) -> bool:

1248 for p in self.to:

1249 p.reveal(self.role)

1250 return True

1251

1252 def can_be_seen_by(self, p: Player) -> bool:

1253 return p in self.to

1254

1255

1256class Pindian(UserAction):

1257 no_reveal = True

1258 card_usage = 'pindian'

1259

1260 def __init__(self, source, target):

1261 self.source = source

1262 self.target = target

1263

1264 def apply_action(self) -> bool:

1265 src = self.source

1266 tgt = self.target

1267 g = self.game

1268

1269 pl = BatchList([tgt, src])

1270 pindian_card: Dict[Character, Card] = {}

1271

1272 with InputTransaction('Pindian', pl) as trans:

1273 for p in pl:

1274 cards = user_choose_cards(self, p, ('cards', 'showncards'), trans=trans)

1275 if cards:

1276 card = cards[0]

1277 else:

1278 card = random_choose_card(g, [p.cards, p.showncards])

1279

1280 pindian_card[p] = card

1281 detach_cards([card])

1282 g.emit_event('pindian_card_chosen', (p, card))

1283

1284 g.players.player.reveal([pindian_card[src], pindian_card[tgt]])

1285 g.emit_event('pindian_card_revealed', self) # for ui.

1286 migrate_cards([pindian_card[src], pindian_card[tgt]], g.deck.droppedcards, unwrap=True)

1287

1288 return pindian_card[src].number > pindian_card[tgt].number

1289

1290 @staticmethod

1291 def cond(cl):

1292 return len(cl) == 1 and \

1293 (not cl[0].is_card(Skill)) and \

1294 cl[0].resides_in.type in ('cards', 'showncards')

1295

1296 def is_valid(self):

1297 src = self.source

1298 tgt = self.target

1299 if src.dead or tgt.dead: return False

1300 if not (src.cards or src.showncards): return False

1301 if not (tgt.cards or tgt.showncards): return False

1302 return True

1303

1304

1305class Reforge(GenericAction):

1306 card_usage = 'reforge'

1307

1308 def __init__(self, source, target, card):

1309 self.source = source

1310 self.target = target

1311 self.card = card

1312

1313 def apply_action(self):

1314 g = self.game

1315 migrate_cards([self.card], g.deck.droppedcards, True)

1316 g.process_action(DrawCards(self.source, 1))

1317

1318 return True

1319

1320 def is_valid(self):

1321 return DrawCards(self.source, 1).can_fire()

1322

1323

1324@register_eh

1325class DyingHandler(THBEventHandler):

1326 interested = ['action_after']

1327

1328 def handle(self, evt_type, act):

1329 if not evt_type == 'action_after': return act 1329 ↛ exitline 1329 didn't return from function 'handle', because the return on line 1329 wasn't executed

1330 if not isinstance(act, BaseDamage): return act 1330 ↛ 1332line 1330 didn't jump to line 1332, because the condition on line 1330 was never false