# Coverage for thb/actions.py : 24%

## 895 statements 253 run 642 missing 5 excluded 9 partial

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

2 from __future__ import annotations

3

4# -- stdlib --

5 from collections import defaultdict

6 from copy import copy

7 from dataclasses import dataclass

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

9 import logging

10

11# -- third party --

12 from typing_extensions import Literal, Protocol

13

14# -- own --

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

16 from game.base import sync_primitive

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

18 from thb.common import PlayerRole

19 from thb.inputlets import ActionInputlet, ChoosePeerCardInputlet

20 from thb.mode import THBAction, THBEventHandler, THBattle

21 from utils.check import CheckFailed, check, check_type

22 from utils.misc import BatchList

23

24# -- typing --

25 if TYPE_CHECKING:

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

27

28

29# -- code --

30 log = logging.getLogger('THBattle_Actions')

31

32

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

34# aux functions

35

36 def ttags(actor):

37 tags = actor.tags

38 tc = tags['turn_count']

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

40

41

42 class CardChooser(Protocol):

43 game: THBattle

44 card_usage: str

45

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

47

48

49 class CharacterChooser(Protocol):

50 game: THBattle

51

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

53

54

55 def 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

66

67 assert categories or candidates

68 assert actors

69

70 timeout = timeout or 25

71

72 from thb.cards.base import VirtualCard

73

74 g = cast(THBattle, initiator.game)

75

76 ilet = ActionInputlet(initiator, categories, candidates)

77

78 @ilet.with_post_process

79 def process(actor: Character, rst):

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

81 try:

82 check(rst)

83

84 skills: List[Type[Skill]]

85 rawcards: List[Card]

86 characters: List[Character]

87 params: Dict[str, Any]

88

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

92

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

156 def 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

171 def 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

184 def 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

198 def 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

207 def 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

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

226

227

228 def register_eh(cls):

229 COMMON_EVENT_HANDLERS.add(cls)

230 return cls

231

232

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

234 class GenericAction(THBAction):

235 pass

236

237

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

239 target_list: List[Character]

240 associated_card: Card

241

242

243@dataclass

244 class 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

253 class CardMovement:

254 trans: MigrateCardsTransaction

255 card: Card

256 fr: CardList

257 to: CardList

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

259

260

261 class 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

336 def 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

369 class 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

395 def 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

407 class 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

422 class 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

434 class 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

455 class 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

495 class 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

510 class Damage(BaseDamage):

511 pass

512

513

514 class 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

521 class 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

549 class 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

568 class 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

594 class 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

631 class 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

674 class 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

680 class 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

705 class DrawCards(BaseDrawCards):

706 pass

707

708

709 class DistributeCards(BaseDrawCards):

710 pass

711

712

713 class DrawCardStage(DrawCards):

714 pass

715

716

717 class 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

732 class 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

880 class ActionStageLaunchCard(LaunchCard):

881 pass

882

883

884 class 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

963 class ActionStage(BaseActionStage):

964 one_shot = False

965 launch_card_cls = ActionStageLaunchCard

966

967

968@register_eh

969 class 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 ↛ 975 line 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 ↛ 978 line 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 ↛ 981 line 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

1000 class 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

1017 class 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

1046 class Fatetell(BaseFatetell):

1047 type = 'fatetell'

1048

1049

1050 class TurnOverCard(BaseFatetell):

1051 type = 'turnover'

1052

1053

1054 class 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

1083 class 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

1115 class 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

1135 class 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

1180 class 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

1189 class 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

1198 class 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 ↛ 1220 line 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 ↛ 1228 line 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

1231 class 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

1240 class 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

1256 class 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

1305 class 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

1325 class DyingHandler(THBEventHandler):

1326 interested = ['action_after']

1327

1328 def handle(self, evt_type, act):

1329 if not evt_type == 'action_after': return act 1329 ↛ exit line 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 ↛ 1332 line 1330 didn't jump to line 1332, because the condition on line 1330 was never false

1331

1332 src = act.source

1333 tgt = act.target

1334 if tgt.dead or tgt.life > 0: return act

1335 if tgt.tags['in_tryrevive']:

1336 # nested TryRevive, just return

1337 # will trigger when Eirin uses Diamond Exinwan to heal herself

1338 return act

1339

1340 try:

1341 tgt.tags['in_tryrevive'] = True

1342 g = self.game

1343 if g.process_action(TryRevive(tgt, dmgact=act)):

1344 return act

1345 finally:

1346 tgt.tags['in_tryrevive'] = False

1347

1348 g.process_action(PlayerDeath(src, tgt))

1349

1350 return act

1351

1352

1353 class ShowCards(GenericAction):

1354 def __init__(self, source, cards, to: Optional[Sequence[Character]] = None):

1355 self.source = self.target = source

1356 self.cards = cards

1357 self.to: Optional[BatchList[Character]] = BatchList(to) if to is not None else None

1358

1359 def apply_action(self):

1360 if not self.cards:

1361 return False

1362

1363 g = self.game

1364 cards = self.cards

1365 to = self.to

1366 to = g.players if to is None else to

1367 to.reveal(cards)

1368 g.emit_event('showcards', (self.target, [copy(c) for c in cards], to))

1369 # g.user_input(

1370 # [p for p in g.players if not p.dead],

1371 # ChooseOptionInputlet(self, (True,)),

1372 # type='all', timeout=1,

1373 # ) # just a delay

1374 return True

1375

1376

1377 class ActionLimitExceeded(ActionShootdown):

1378 pass

1379

1380

1381 class VitalityLimitExceeded(ActionLimitExceeded):

1382 pass