mirror of
https://github.com/ankitects/anki.git
synced 2025-09-19 06:22:22 -04:00
262 lines
8.8 KiB
Python
262 lines
8.8 KiB
Python
# Copyright: Ankitects Pty Ltd and contributors
|
|
# License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html
|
|
|
|
"""
|
|
This file contains experimental scheduler changes, and is not currently
|
|
used by Anki.
|
|
|
|
It uses the same DB schema as the V2 scheduler, and 'schedVer' remains
|
|
as '2' internally.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import List, Literal, Sequence, Tuple, Union
|
|
|
|
import anki._backend.backend_pb2 as _pb
|
|
from anki.cards import Card, CardId
|
|
from anki.collection import OpChanges
|
|
from anki.consts import *
|
|
from anki.decks import DeckId
|
|
from anki.errors import DBError
|
|
from anki.scheduler.base import CongratsInfo
|
|
from anki.scheduler.legacy import SchedulerBaseWithLegacy
|
|
from anki.types import assert_exhaustive
|
|
from anki.utils import intTime
|
|
|
|
QueuedCards = _pb.QueuedCards
|
|
SchedulingState = _pb.SchedulingState
|
|
NextStates = _pb.NextCardStates
|
|
CardAnswer = _pb.CardAnswer
|
|
|
|
|
|
class Scheduler(SchedulerBaseWithLegacy):
|
|
version = 3
|
|
|
|
# don't rely on this, it will likely be removed in the future
|
|
reps = 0
|
|
|
|
# Fetching the next card
|
|
##########################################################################
|
|
|
|
def get_queued_cards(
|
|
self,
|
|
*,
|
|
fetch_limit: int = 1,
|
|
intraday_learning_only: bool = False,
|
|
) -> Union[QueuedCards, CongratsInfo]:
|
|
"Returns one or more card ids, or the congratulations screen info."
|
|
info = self.col._backend.get_queued_cards(
|
|
fetch_limit=fetch_limit, intraday_learning_only=intraday_learning_only
|
|
)
|
|
kind = info.WhichOneof("value")
|
|
if kind == "queued_cards":
|
|
return info.queued_cards
|
|
elif kind == "congrats_info":
|
|
return info.congrats_info
|
|
else:
|
|
assert_exhaustive(kind)
|
|
assert False
|
|
|
|
def next_states(self, card_id: CardId) -> NextStates:
|
|
"New states corresponding to each answer button press."
|
|
return self.col._backend.get_next_card_states(card_id)
|
|
|
|
def describe_next_states(self, next_states: NextStates) -> Sequence[str]:
|
|
"Labels for each of the answer buttons."
|
|
return self.col._backend.describe_next_states(next_states)
|
|
|
|
# Answering a card
|
|
##########################################################################
|
|
|
|
def build_answer(
|
|
self, *, card: Card, states: NextStates, rating: CardAnswer.Rating.V
|
|
) -> CardAnswer:
|
|
"Build input for answer_card()."
|
|
if rating == CardAnswer.AGAIN:
|
|
new_state = states.again
|
|
elif rating == CardAnswer.HARD:
|
|
new_state = states.hard
|
|
elif rating == CardAnswer.GOOD:
|
|
new_state = states.good
|
|
elif rating == CardAnswer.EASY:
|
|
new_state = states.easy
|
|
else:
|
|
assert False, "invalid rating"
|
|
|
|
return CardAnswer(
|
|
card_id=card.id,
|
|
current_state=states.current,
|
|
new_state=new_state,
|
|
rating=rating,
|
|
answered_at_millis=intTime(1000),
|
|
milliseconds_taken=card.timeTaken(),
|
|
)
|
|
|
|
def answer_card(self, input: CardAnswer) -> OpChanges:
|
|
"Update card to provided state, and remove it from queue."
|
|
self.reps += 1
|
|
return self.col._backend.answer_card(input=input)
|
|
|
|
def state_is_leech(self, new_state: SchedulingState) -> bool:
|
|
"True if new state marks the card as a leech."
|
|
return self.col._backend.state_is_leech(new_state)
|
|
|
|
# Fetching the next card (legacy API)
|
|
##########################################################################
|
|
|
|
def reset(self) -> None:
|
|
# backend automatically resets queues as operations are performed
|
|
pass
|
|
|
|
def getCard(self) -> Optional[Card]:
|
|
"""Fetch the next card from the queue. None if finished."""
|
|
response = self.get_queued_cards()
|
|
if isinstance(response, QueuedCards):
|
|
backend_card = response.cards[0].card
|
|
card = Card(self.col)
|
|
card._load_from_backend_card(backend_card)
|
|
card.startTimer()
|
|
return card
|
|
else:
|
|
return None
|
|
|
|
def _is_finished(self) -> bool:
|
|
"Don't use this, it is a stop-gap until this code is refactored."
|
|
info = self.get_queued_cards()
|
|
return isinstance(info, CongratsInfo)
|
|
|
|
def counts(self, card: Optional[Card] = None) -> Tuple[int, int, int]:
|
|
info = self.get_queued_cards()
|
|
if isinstance(info, CongratsInfo):
|
|
counts = [0, 0, 0]
|
|
else:
|
|
counts = [info.new_count, info.learning_count, info.review_count]
|
|
|
|
return tuple(counts) # type: ignore
|
|
|
|
@property
|
|
def newCount(self) -> int:
|
|
return self.counts()[0]
|
|
|
|
@property
|
|
def lrnCount(self) -> int:
|
|
return self.counts()[1]
|
|
|
|
@property
|
|
def reviewCount(self) -> int:
|
|
return self.counts()[2]
|
|
|
|
def countIdx(self, card: Card) -> int:
|
|
if card.queue in (QUEUE_TYPE_DAY_LEARN_RELEARN, QUEUE_TYPE_PREVIEW):
|
|
return QUEUE_TYPE_LRN
|
|
return card.queue
|
|
|
|
def answerButtons(self, card: Card) -> int:
|
|
return 4
|
|
|
|
def nextIvlStr(self, card: Card, ease: int, short: bool = False) -> str:
|
|
"Return the next interval for CARD as a string."
|
|
states = self.col._backend.get_next_card_states(card.id)
|
|
return self.col._backend.describe_next_states(states)[ease - 1]
|
|
|
|
# Answering a card (legacy API)
|
|
##########################################################################
|
|
|
|
def answerCard(self, card: Card, ease: Literal[1, 2, 3, 4]) -> OpChanges:
|
|
if ease == BUTTON_ONE:
|
|
rating = CardAnswer.AGAIN
|
|
elif ease == BUTTON_TWO:
|
|
rating = CardAnswer.HARD
|
|
elif ease == BUTTON_THREE:
|
|
rating = CardAnswer.GOOD
|
|
elif ease == BUTTON_FOUR:
|
|
rating = CardAnswer.EASY
|
|
else:
|
|
assert False, "invalid ease"
|
|
|
|
changes = self.answer_card(
|
|
self.build_answer(
|
|
card=card, states=self.next_states(card_id=card.id), rating=rating
|
|
)
|
|
)
|
|
|
|
# tests assume card will be mutated, so we need to reload it
|
|
card.load()
|
|
|
|
return changes
|
|
|
|
# Next times (legacy API)
|
|
##########################################################################
|
|
# fixme: move these into tests_schedv2 in the future
|
|
|
|
def _interval_for_state(self, state: _pb.SchedulingState) -> int:
|
|
kind = state.WhichOneof("value")
|
|
if kind == "normal":
|
|
return self._interval_for_normal_state(state.normal)
|
|
elif kind == "filtered":
|
|
return self._interval_for_filtered_state(state.filtered)
|
|
else:
|
|
assert_exhaustive(kind)
|
|
return 0 # unreachable
|
|
|
|
def _interval_for_normal_state(self, normal: _pb.SchedulingState.Normal) -> int:
|
|
kind = normal.WhichOneof("value")
|
|
if kind == "new":
|
|
return 0
|
|
elif kind == "review":
|
|
return normal.review.scheduled_days * 86400
|
|
elif kind == "learning":
|
|
return normal.learning.scheduled_secs
|
|
elif kind == "relearning":
|
|
return normal.relearning.learning.scheduled_secs
|
|
else:
|
|
assert_exhaustive(kind)
|
|
return 0 # unreachable
|
|
|
|
def _interval_for_filtered_state(
|
|
self, filtered: _pb.SchedulingState.Filtered
|
|
) -> int:
|
|
kind = filtered.WhichOneof("value")
|
|
if kind == "preview":
|
|
return filtered.preview.scheduled_secs
|
|
elif kind == "rescheduling":
|
|
return self._interval_for_normal_state(filtered.rescheduling.original_state)
|
|
else:
|
|
assert_exhaustive(kind)
|
|
return 0 # unreachable
|
|
|
|
def nextIvl(self, card: Card, ease: int) -> Any:
|
|
"Don't use this - it is only required by tests, and will be moved in the future."
|
|
states = self.col._backend.get_next_card_states(card.id)
|
|
if ease == BUTTON_ONE:
|
|
new_state = states.again
|
|
elif ease == BUTTON_TWO:
|
|
new_state = states.hard
|
|
elif ease == BUTTON_THREE:
|
|
new_state = states.good
|
|
elif ease == BUTTON_FOUR:
|
|
new_state = states.easy
|
|
else:
|
|
assert False, "invalid ease"
|
|
|
|
return self._interval_for_state(new_state)
|
|
|
|
# Other legacy
|
|
###################
|
|
|
|
# called by col.decks.active(), which add-ons are using
|
|
@property
|
|
def active_decks(self) -> List[DeckId]:
|
|
try:
|
|
return self.col.db.list("select id from active_decks")
|
|
except DBError:
|
|
return []
|
|
|
|
# used by custom study; will likely be rolled into a separate routine
|
|
# in the future
|
|
def totalNewForCurrentDeck(self) -> int:
|
|
return self.col.db.scalar(
|
|
f"""
|
|
select count() from cards where queue={QUEUE_TYPE_NEW} and did in (select id from active_decks)"""
|
|
)
|