# Copyright: Ankitects Pty Ltd and contributors # License: GNU AGPL, version 3 or later; http://www.gnu.org/licenses/agpl.html from __future__ import annotations import itertools import random import time from heapq import * from operator import itemgetter from typing import Any, Dict, List, Optional, Tuple, Union import anki from anki import hooks from anki.cards import Card from anki.consts import * from anki.schedv2 import Scheduler as V2 from anki.utils import ids2str, intTime # queue types: 0=new/cram, 1=lrn, 2=rev, 3=day lrn, -1=suspended, -2=buried # revlog types: 0=lrn, 1=rev, 2=relrn, 3=cram # positive revlog intervals are in days (rev), negative in seconds (lrn) class Scheduler(V2): name = "std" haveCustomStudy = True _spreadRev = True _burySiblingsOnAnswer = True def __init__( # pylint: disable=super-init-not-called self, col: anki.storage._Collection ) -> None: self.col = col.weakref() self.queueLimit = 50 self.reportLimit = 1000 self.dynReportLimit = 99999 self.reps = 0 self.lrnCount = 0 self.revCount = 0 self.newCount = 0 self.today: Optional[int] = None self._haveQueues = False self._updateCutoff() def answerCard(self, card: Card, ease: int) -> None: self.col.log() assert 1 <= ease <= 4 self.col.markReview(card) if self._burySiblingsOnAnswer: self._burySiblings(card) card.reps += 1 # former is for logging new cards, latter also covers filt. decks card.wasNew = card.type == CARD_TYPE_NEW # type: ignore wasNewQ = card.queue == QUEUE_TYPE_NEW if wasNewQ: # came from the new queue, move to learning card.queue = QUEUE_TYPE_LRN # if it was a new card, it's now a learning card if card.type == CARD_TYPE_NEW: card.type = CARD_TYPE_LRN # init reps to graduation card.left = self._startingLeft(card) # dynamic? if card.odid and card.type == CARD_TYPE_REV: if self._resched(card): # reviews get their ivl boosted on first sight card.ivl = self._dynIvlBoost(card) card.odue = self.today + card.ivl self._updateStats(card, "new") if card.queue in (QUEUE_TYPE_LRN, QUEUE_TYPE_DAY_LEARN_RELEARN): self._answerLrnCard(card, ease) if not wasNewQ: self._updateStats(card, "lrn") elif card.queue == QUEUE_TYPE_REV: self._answerRevCard(card, ease) self._updateStats(card, "rev") else: raise Exception("Invalid queue") self._updateStats(card, "time", card.timeTaken()) card.mod = intTime() card.usn = self.col.usn() card.flushSched() def counts(self, card: Optional[Card] = None) -> Tuple[int, int, int]: counts = [self.newCount, self.lrnCount, self.revCount] if card: idx = self.countIdx(card) if idx == QUEUE_TYPE_LRN: counts[QUEUE_TYPE_LRN] += card.left // 1000 else: counts[idx] += 1 new, lrn, rev = counts return (new, lrn, rev) def countIdx(self, card: Card) -> int: if card.queue == QUEUE_TYPE_DAY_LEARN_RELEARN: return QUEUE_TYPE_LRN return card.queue def answerButtons(self, card: Card) -> int: if card.odue: # normal review in dyn deck? if card.odid and card.queue == QUEUE_TYPE_REV: return 4 conf = self._lrnConf(card) if card.type in (CARD_TYPE_NEW, CARD_TYPE_LRN) or len(conf["delays"]) > 1: return 3 return 2 elif card.queue == QUEUE_TYPE_REV: return 4 else: return 3 def unburyCards(self) -> None: "Unbury cards." self.col.conf["lastUnburied"] = self.today self.col.log( self.col.db.list( f"select id from cards where queue = {QUEUE_TYPE_SIBLING_BURIED}" ) ) self.col.db.execute( f"update cards set queue=type where queue = {QUEUE_TYPE_SIBLING_BURIED}" ) def unburyCardsForDeck(self) -> None: # type: ignore[override] sids = self._deckLimit() self.col.log( self.col.db.list( f"select id from cards where queue = {QUEUE_TYPE_SIBLING_BURIED} and did in %s" % sids ) ) self.col.db.execute( f"update cards set mod=?,usn=?,queue=type where queue = {QUEUE_TYPE_SIBLING_BURIED} and did in %s" % sids, intTime(), self.col.usn(), ) # Deck list ########################################################################## def deckDueList(self) -> List[List[Any]]: "Returns [deckname, did, rev, lrn, new]" self._checkDay() self.col.decks.checkIntegrity() decks = self.col.decks.all() decks.sort(key=itemgetter("name")) lims: Dict[str, List[int]] = {} data = [] def parent(name): parts = name.split("::") if len(parts) < 2: return None parts = parts[:-1] return "::".join(parts) for deck in decks: p = parent(deck["name"]) # new nlim = self._deckNewLimitSingle(deck) if p: nlim = min(nlim, lims[p][0]) new = self._newForDeck(deck["id"], nlim) # learning lrn = self._lrnForDeck(deck["id"]) # reviews rlim = self._deckRevLimitSingle(deck) if p: rlim = min(rlim, lims[p][1]) rev = self._revForDeck(deck["id"], rlim) # save to list data.append([deck["name"], deck["id"], rev, lrn, new]) # add deck as a parent lims[deck["name"]] = [nlim, rlim] return data def _groupChildrenMain(self, grps: List[List[Any]]) -> Any: tree = [] # group and recurse def key(grp): return grp[0][0] for (head, tail) in itertools.groupby(grps, key=key): tail = list(tail) # type: ignore did = None rev = 0 new = 0 lrn = 0 children = [] for c in tail: if len(c[0]) == 1: # current node did = c[1] rev += c[2] lrn += c[3] new += c[4] else: # set new string to tail c[0] = c[0][1:] children.append(c) children = self._groupChildrenMain(children) # tally up children counts for ch in children: rev += ch[2] lrn += ch[3] new += ch[4] # limit the counts to the deck's limits conf = self.col.decks.confForDid(did) deck = self.col.decks.get(did) if not conf["dyn"]: rev = max(0, min(rev, conf["rev"]["perDay"] - deck["revToday"][1])) new = max(0, min(new, self._deckNewLimitSingle(deck))) tree.append((head, did, rev, lrn, new, children)) return tuple(tree) # Getting the next card ########################################################################## def _getCard(self) -> Optional[Card]: "Return the next due card id, or None." # learning card due? c = self._getLrnCard() if c: return c # new first, or time for one? if self._timeForNewCard(): c = self._getNewCard() if c: return c # card due for review? c = self._getRevCard() if c: return c # day learning card due? c = self._getLrnDayCard() if c: return c # new cards left? c = self._getNewCard() if c: return c # collapse or finish return self._getLrnCard(collapse=True) # Learning queues ########################################################################## def _resetLrnCount(self) -> None: # sub-day self.lrnCount = ( self.col.db.scalar( f""" select sum(left/1000) from (select left from cards where did in %s and queue = {QUEUE_TYPE_LRN} and due < ? limit %d)""" % (self._deckLimit(), self.reportLimit), self.dayCutoff, ) or 0 ) # day self.lrnCount += self.col.db.scalar( f""" select count() from cards where did in %s and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit %d""" % (self._deckLimit(), self.reportLimit), self.today, ) def _resetLrn(self) -> None: self._resetLrnCount() self._lrnQueue: List[Any] = [] self._lrnDayQueue: List[Any] = [] self._lrnDids = self.col.decks.active()[:] # sub-day learning def _fillLrn(self) -> Union[bool, List[Any]]: if not self.lrnCount: return False if self._lrnQueue: return True self._lrnQueue = self.col.db.all( f""" select due, id from cards where did in %s and queue = {QUEUE_TYPE_LRN} and due < ? limit %d""" % (self._deckLimit(), self.reportLimit), self.dayCutoff, ) # as it arrives sorted by did first, we need to sort it self._lrnQueue.sort() return self._lrnQueue def _getLrnCard(self, collapse: bool = False) -> Optional[Card]: if self._fillLrn(): cutoff = time.time() if collapse: cutoff += self.col.conf["collapseTime"] if self._lrnQueue[0][0] < cutoff: id = heappop(self._lrnQueue)[1] card = self.col.getCard(id) self.lrnCount -= card.left // 1000 return card return None def _answerLrnCard(self, card: Card, ease: int) -> None: # ease 1=no, 2=yes, 3=remove conf = self._lrnConf(card) if card.odid and not card.wasNew: # type: ignore type = REVLOG_CRAM elif card.type == CARD_TYPE_REV: type = REVLOG_RELRN else: type = REVLOG_LRN leaving = False # lrnCount was decremented once when card was fetched lastLeft = card.left # immediate graduate? if ease == BUTTON_THREE: self._rescheduleAsRev(card, conf, True) leaving = True # graduation time? elif ease == BUTTON_TWO and (card.left % 1000) - 1 <= 0: self._rescheduleAsRev(card, conf, False) leaving = True else: # one step towards graduation if ease == BUTTON_TWO: # decrement real left count and recalculate left today left = (card.left % 1000) - 1 card.left = self._leftToday(conf["delays"], left) * 1000 + left # failed else: card.left = self._startingLeft(card) resched = self._resched(card) if "mult" in conf and resched: # review that's lapsed card.ivl = max(1, conf["minInt"], card.ivl * conf["mult"]) else: # new card; no ivl adjustment pass if resched and card.odid: card.odue = self.today + 1 delay = self._delayForGrade(conf, card.left) if card.due < time.time(): # not collapsed; add some randomness delay *= int(random.uniform(1, 1.25)) card.due = int(time.time() + delay) # due today? if card.due < self.dayCutoff: self.lrnCount += card.left // 1000 # if the queue is not empty and there's nothing else to do, make # sure we don't put it at the head of the queue and end up showing # it twice in a row card.queue = QUEUE_TYPE_LRN if self._lrnQueue and not self.revCount and not self.newCount: smallestDue = self._lrnQueue[0][0] card.due = max(card.due, smallestDue + 1) heappush(self._lrnQueue, (card.due, card.id)) else: # the card is due in one or more days, so we need to use the # day learn queue ahead = ((card.due - self.dayCutoff) // 86400) + 1 card.due = self.today + ahead card.queue = QUEUE_TYPE_DAY_LEARN_RELEARN self._logLrn(card, ease, conf, leaving, type, lastLeft) def _lrnConf(self, card: Card) -> Dict[str, Any]: if card.type == CARD_TYPE_REV: return self._lapseConf(card) else: return self._newConf(card) def _rescheduleAsRev(self, card: Card, conf: Dict[str, Any], early: bool) -> None: lapse = card.type == CARD_TYPE_REV if lapse: if self._resched(card): card.due = max(self.today + 1, card.odue) else: card.due = card.odue card.odue = 0 else: self._rescheduleNew(card, conf, early) card.queue = QUEUE_TYPE_REV card.type = CARD_TYPE_REV # if we were dynamic, graduating means moving back to the old deck resched = self._resched(card) if card.odid: card.did = card.odid card.odue = 0 card.odid = 0 # if rescheduling is off, it needs to be set back to a new card if not resched and not lapse: card.queue = card.type = CARD_TYPE_NEW card.due = self.col.nextID("pos") def _startingLeft(self, card: Card) -> int: if card.type == CARD_TYPE_REV: conf = self._lapseConf(card) else: conf = self._lrnConf(card) tot = len(conf["delays"]) tod = self._leftToday(conf["delays"], tot) return tot + tod * 1000 def _graduatingIvl( self, card: Card, conf: Dict[str, Any], early: bool, adj: bool = True ) -> int: if card.type == CARD_TYPE_REV: # lapsed card being relearnt if card.odid: if conf["resched"]: return self._dynIvlBoost(card) return card.ivl if not early: # graduate ideal = conf["ints"][0] else: # early remove ideal = conf["ints"][1] if adj: return self._adjRevIvl(card, ideal) else: return ideal def _rescheduleNew(self, card: Card, conf: Dict[str, Any], early: bool) -> None: "Reschedule a new card that's graduated for the first time." card.ivl = self._graduatingIvl(card, conf, early) card.due = self.today + card.ivl card.factor = conf["initialFactor"] def _logLrn( self, card: Card, ease: int, conf: Dict[str, Any], leaving: bool, type: int, lastLeft: int, ) -> None: lastIvl = -(self._delayForGrade(conf, lastLeft)) ivl = card.ivl if leaving else -(self._delayForGrade(conf, card.left)) def log(): self.col.db.execute( "insert into revlog values (?,?,?,?,?,?,?,?,?)", int(time.time() * 1000), card.id, self.col.usn(), ease, ivl, lastIvl, card.factor, card.timeTaken(), type, ) try: log() except: # duplicate pk; retry in 10ms time.sleep(0.01) log() def removeLrn(self, ids: Optional[List[int]] = None) -> None: "Remove cards from the learning queues." if ids: extra = " and id in " + ids2str(ids) else: # benchmarks indicate it's about 10x faster to search all decks # with the index than scan the table extra = " and did in " + ids2str(self.col.decks.allIds()) # review cards in relearning self.col.db.execute( f""" update cards set due = odue, queue = {QUEUE_TYPE_REV}, mod = %d, usn = %d, odue = 0 where queue in ({QUEUE_TYPE_LRN},{QUEUE_TYPE_DAY_LEARN_RELEARN}) and type = {CARD_TYPE_REV} %s """ % (intTime(), self.col.usn(), extra) ) # new cards in learning self.forgetCards( self.col.db.list( f"select id from cards where queue in ({QUEUE_TYPE_LRN},{QUEUE_TYPE_DAY_LEARN_RELEARN}) %s" % extra ) ) def _lrnForDeck(self, did: int) -> int: cnt = ( self.col.db.scalar( f""" select sum(left/1000) from (select left from cards where did = ? and queue = {QUEUE_TYPE_LRN} and due < ? limit ?)""", did, intTime() + self.col.conf["collapseTime"], self.reportLimit, ) or 0 ) return cnt + self.col.db.scalar( f""" select count() from (select 1 from cards where did = ? and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit ?)""", did, self.today, self.reportLimit, ) # Reviews ########################################################################## def _deckRevLimit(self, did: int) -> int: return self._deckNewLimit(did, self._deckRevLimitSingle) def _deckRevLimitSingle(self, d: Dict[str, Any]) -> int: # type: ignore[override] if d["dyn"]: return self.reportLimit c = self.col.decks.confForDid(d["id"]) limit = max(0, c["rev"]["perDay"] - d["revToday"][1]) return hooks.scheduler_review_limit_for_single_deck(limit, d) def _revForDeck(self, did: int, lim: int) -> int: # type: ignore[override] lim = min(lim, self.reportLimit) return self.col.db.scalar( f""" select count() from (select 1 from cards where did = ? and queue = {QUEUE_TYPE_REV} and due <= ? limit ?)""", did, self.today, lim, ) def _resetRevCount(self) -> None: def cntFn(did, lim): return self.col.db.scalar( f""" select count() from (select id from cards where did = ? and queue = {QUEUE_TYPE_REV} and due <= ? limit %d)""" % lim, did, self.today, ) self.revCount = self._walkingCount(self._deckRevLimitSingle, cntFn) def _resetRev(self) -> None: self._resetRevCount() self._revQueue: List[Any] = [] self._revDids = self.col.decks.active()[:] def _fillRev(self) -> Optional[bool]: if self._revQueue: return True if not self.revCount: return False while self._revDids: did = self._revDids[0] lim = min(self.queueLimit, self._deckRevLimit(did)) if lim: # fill the queue with the current did self._revQueue = self.col.db.list( f""" select id from cards where did = ? and queue = {QUEUE_TYPE_REV} and due <= ? limit ?""", did, self.today, lim, ) if self._revQueue: # ordering if self.col.decks.get(did)["dyn"]: # dynamic decks need due order preserved self._revQueue.reverse() else: # random order for regular reviews r = random.Random() r.seed(self.today) r.shuffle(self._revQueue) # is the current did empty? if len(self._revQueue) < lim: self._revDids.pop(0) return True # nothing left in the deck; move to next self._revDids.pop(0) if self.revCount: # if we didn't get a card but the count is non-zero, # we need to check again for any cards that were # removed from the queue but not buried self._resetRev() return self._fillRev() return None # Answering a review card ########################################################################## def _answerRevCard(self, card: Card, ease: int) -> None: delay: int = 0 if ease == BUTTON_ONE: delay = self._rescheduleLapse(card) else: self._rescheduleRev(card, ease) self._logRev(card, ease, delay, REVLOG_REV) def _rescheduleLapse(self, card: Card) -> int: conf = self._lapseConf(card) card.lastIvl = card.ivl if self._resched(card): card.lapses += 1 card.ivl = self._nextLapseIvl(card, conf) card.factor = max(1300, card.factor - 200) card.due = self.today + card.ivl # if it's a filtered deck, update odue as well if card.odid: card.odue = card.due # if suspended as a leech, nothing to do delay: int = 0 if self._checkLeech(card, conf) and card.queue == QUEUE_TYPE_SUSPENDED: return delay # if no relearning steps, nothing to do if not conf["delays"]: return delay # record rev due date for later if not card.odue: card.odue = card.due delay = self._delayForGrade(conf, 0) card.due = int(delay + time.time()) card.left = self._startingLeft(card) # queue 1 if card.due < self.dayCutoff: self.lrnCount += card.left // 1000 card.queue = QUEUE_TYPE_LRN heappush(self._lrnQueue, (card.due, card.id)) else: # day learn queue ahead = ((card.due - self.dayCutoff) // 86400) + 1 card.due = self.today + ahead card.queue = QUEUE_TYPE_DAY_LEARN_RELEARN return delay def _nextLapseIvl(self, card: Card, conf: Dict[str, Any]) -> int: return max(conf["minInt"], int(card.ivl * conf["mult"])) def _rescheduleRev(self, card: Card, ease: int) -> None: # type: ignore[override] # update interval card.lastIvl = card.ivl if self._resched(card): self._updateRevIvl(card, ease) # then the rest card.factor = max(1300, card.factor + [-150, 0, 150][ease - 2]) card.due = self.today + card.ivl else: card.due = card.odue if card.odid: card.did = card.odid card.odid = 0 card.odue = 0 # Interval management ########################################################################## def _nextRevIvl(self, card: Card, ease: int) -> int: # type: ignore[override] "Ideal next interval for CARD, given EASE." delay = self._daysLate(card) conf = self._revConf(card) fct = card.factor / 1000 ivl2 = self._constrainedIvl((card.ivl + delay // 4) * 1.2, conf, card.ivl) ivl3 = self._constrainedIvl((card.ivl + delay // 2) * fct, conf, ivl2) ivl4 = self._constrainedIvl( (card.ivl + delay) * fct * conf["ease4"], conf, ivl3 ) if ease == BUTTON_TWO: interval = ivl2 elif ease == BUTTON_THREE: interval = ivl3 elif ease == BUTTON_FOUR: interval = ivl4 # interval capped? return min(interval, conf["maxIvl"]) def _constrainedIvl(self, ivl: float, conf: Dict[str, Any], prev: int) -> int: # type: ignore[override] "Integer interval after interval factor and prev+1 constraints applied." new = ivl * conf.get("ivlFct", 1) return int(max(new, prev + 1)) def _updateRevIvl(self, card: Card, ease: int) -> None: idealIvl = self._nextRevIvl(card, ease) card.ivl = min( max(self._adjRevIvl(card, idealIvl), card.ivl + 1), self._revConf(card)["maxIvl"], ) def _adjRevIvl(self, card: Card, idealIvl: int) -> int: if self._spreadRev: idealIvl = self._fuzzedIvl(idealIvl) return idealIvl # Dynamic deck handling ########################################################################## def rebuildDyn(self, did: Optional[int] = None) -> Optional[List[int]]: # type: ignore[override] "Rebuild a dynamic deck." did = did or self.col.decks.selected() deck = self.col.decks.get(did) assert deck["dyn"] # move any existing cards back first, then fill self.emptyDyn(did) ids = self._fillDyn(deck) if not ids: return None # and change to our new deck self.col.decks.select(did) return ids def _fillDyn(self, deck: Dict[str, Any]) -> List[int]: # type: ignore[override] search, limit, order = deck["terms"][0] orderlimit = self._dynOrder(order, limit) if search.strip(): search = "(%s)" % search search = "%s -is:suspended -is:buried -deck:filtered -is:learn" % search try: ids = self.col.findCards(search, order=orderlimit) except: ids = [] return ids # move the cards over self.col.log(deck["id"], ids) self._moveToDyn(deck["id"], ids) return ids def emptyDyn(self, did: Optional[int], lim: Optional[str] = None) -> None: if not lim: lim = "did = %s" % did self.col.log(self.col.db.list("select id from cards where %s" % lim)) # move out of cram queue self.col.db.execute( f""" update cards set did = odid, queue = (case when type = {CARD_TYPE_LRN} then {QUEUE_TYPE_NEW} else type end), type = (case when type = {CARD_TYPE_LRN} then {CARD_TYPE_NEW} else type end), due = odue, odue = 0, odid = 0, usn = ? where %s""" % lim, self.col.usn(), ) def _moveToDyn(self, did: int, ids: List[int]) -> None: # type: ignore[override] deck = self.col.decks.get(did) data = [] t = intTime() u = self.col.usn() for c, id in enumerate(ids): # start at -100000 so that reviews are all due data.append((did, -100000 + c, u, id)) # due reviews stay in the review queue. careful: can't use # "odid or did", as sqlite converts to boolean queue = f""" (case when type={CARD_TYPE_REV} and (case when odue then odue <= %d else due <= %d end) then {QUEUE_TYPE_REV} else {QUEUE_TYPE_NEW} end)""" queue %= (self.today, self.today) self.col.db.executemany( """ update cards set odid = (case when odid then odid else did end), odue = (case when odue then odue else due end), did = ?, queue = %s, due = ?, usn = ? where id = ?""" % queue, data, ) def _dynIvlBoost(self, card: Card) -> int: assert card.odid and card.type == CARD_TYPE_REV assert card.factor elapsed = card.ivl - (card.odue - self.today) factor = ((card.factor / 1000) + 1.2) / 2 ivl = int(max(card.ivl, elapsed * factor, 1)) conf = self._revConf(card) return min(conf["maxIvl"], ivl) # Leeches ########################################################################## def _checkLeech(self, card: Card, conf: Dict[str, Any]) -> bool: "Leech handler. True if card was a leech." lf = conf["leechFails"] if not lf: return False # if over threshold or every half threshold reps after that if card.lapses >= lf and (card.lapses - lf) % (max(lf // 2, 1)) == 0: # add a leech tag f = card.note() f.addTag("leech") f.flush() # handle a = conf["leechAction"] if a == LEECH_SUSPEND: # if it has an old due, remove it from cram/relearning if card.odue: card.due = card.odue if card.odid: card.did = card.odid card.odue = card.odid = 0 card.queue = QUEUE_TYPE_SUSPENDED # notify UI hooks.card_did_leech(card) return True else: return False # Tools ########################################################################## def _newConf(self, card: Card) -> Dict[str, Any]: conf = self._cardConf(card) # normal deck if not card.odid: return conf["new"] # dynamic deck; override some attributes, use original deck for others oconf = self.col.decks.confForDid(card.odid) delays = conf["delays"] or oconf["new"]["delays"] return dict( # original deck ints=oconf["new"]["ints"], initialFactor=oconf["new"]["initialFactor"], bury=oconf["new"].get("bury", True), # overrides delays=delays, separate=conf["separate"], order=NEW_CARDS_DUE, perDay=self.reportLimit, ) def _lapseConf(self, card: Card) -> Dict[str, Any]: conf = self._cardConf(card) # normal deck if not card.odid: return conf["lapse"] # dynamic deck; override some attributes, use original deck for others oconf = self.col.decks.confForDid(card.odid) delays = conf["delays"] or oconf["lapse"]["delays"] return dict( # original deck minInt=oconf["lapse"]["minInt"], leechFails=oconf["lapse"]["leechFails"], leechAction=oconf["lapse"]["leechAction"], mult=oconf["lapse"]["mult"], # overrides delays=delays, resched=conf["resched"], ) def _resched(self, card: Card) -> bool: conf = self._cardConf(card) if not conf["dyn"]: return True return conf["resched"] # Daily cutoff ########################################################################## def _updateCutoff(self) -> None: oldToday = self.today # days since col created self.today = int((time.time() - self.col.crt) // 86400) # end of day cutoff self.dayCutoff = self.col.crt + (self.today + 1) * 86400 if oldToday != self.today: self.col.log(self.today, self.dayCutoff) # update all daily counts, but don't save decks to prevent needless # conflicts. we'll save on card answer instead def update(g): for t in "new", "rev", "lrn", "time": key = t + "Today" if g[key][0] != self.today: g[key] = [self.today, 0] for deck in self.col.decks.all(): update(deck) # unbury if the day has rolled over unburied = self.col.conf.get("lastUnburied", 0) if unburied < self.today: self.unburyCards() # Deck finished state ########################################################################## def haveBuried(self) -> bool: sdids = self._deckLimit() cnt = self.col.db.scalar( f"select 1 from cards where queue = {QUEUE_TYPE_SIBLING_BURIED} and did in %s limit 1" % sdids ) return not not cnt # Next time reports ########################################################################## def nextIvl(self, card: Card, ease: int) -> float: "Return the next interval for CARD, in seconds." if card.queue in (QUEUE_TYPE_NEW, QUEUE_TYPE_LRN, QUEUE_TYPE_DAY_LEARN_RELEARN): return self._nextLrnIvl(card, ease) elif ease == BUTTON_ONE: # lapsed conf = self._lapseConf(card) if conf["delays"]: return conf["delays"][0] * 60 return self._nextLapseIvl(card, conf) * 86400 else: # review return self._nextRevIvl(card, ease) * 86400 # this isn't easily extracted from the learn code def _nextLrnIvl(self, card: Card, ease: int) -> float: if card.queue == 0: card.left = self._startingLeft(card) conf = self._lrnConf(card) if ease == BUTTON_ONE: # fail return self._delayForGrade(conf, len(conf["delays"])) elif ease == BUTTON_THREE: # early removal if not self._resched(card): return 0 return self._graduatingIvl(card, conf, True, adj=False) * 86400 else: left = card.left % 1000 - 1 if left <= 0: # graduate if not self._resched(card): return 0 return self._graduatingIvl(card, conf, False, adj=False) * 86400 else: return self._delayForGrade(conf, left) # Suspending ########################################################################## def suspendCards(self, ids: List[int]) -> None: "Suspend cards." self.col.log(ids) self.remFromDyn(ids) self.removeLrn(ids) self.col.db.execute( f"update cards set queue={QUEUE_TYPE_SUSPENDED},mod=?,usn=? where id in " + ids2str(ids), intTime(), self.col.usn(), ) def unsuspendCards(self, ids: List[int]) -> None: "Unsuspend cards." self.col.log(ids) self.col.db.execute( "update cards set queue=type,mod=?,usn=? " f"where queue = {QUEUE_TYPE_SUSPENDED} and id in " + ids2str(ids), intTime(), self.col.usn(), ) def buryCards(self, cids: List[int]) -> None: # type: ignore[override] self.col.log(cids) self.remFromDyn(cids) self.removeLrn(cids) self.col.db.execute( f""" update cards set queue={QUEUE_TYPE_SIBLING_BURIED},mod=?,usn=? where id in """ + ids2str(cids), intTime(), self.col.usn(), ) # Sibling spacing ########################################################################## def _burySiblings(self, card: Card) -> None: toBury = [] nconf = self._newConf(card) buryNew = nconf.get("bury", True) rconf = self._revConf(card) buryRev = rconf.get("bury", True) # loop through and remove from queues for cid, queue in self.col.db.execute( f""" select id, queue from cards where nid=? and id!=? and (queue={QUEUE_TYPE_NEW} or (queue={QUEUE_TYPE_REV} and due<=?))""", card.nid, card.id, self.today, ): if queue == QUEUE_TYPE_REV: if buryRev: toBury.append(cid) # if bury disabled, we still discard to give same-day spacing try: self._revQueue.remove(cid) except ValueError: pass else: # if bury disabled, we still discard to give same-day spacing if buryNew: toBury.append(cid) try: self._newQueue.remove(cid) except ValueError: pass # then bury if toBury: self.col.db.execute( f"update cards set queue={QUEUE_TYPE_SIBLING_BURIED},mod=?,usn=? where id in " + ids2str(toBury), intTime(), self.col.usn(), ) self.col.log(toBury)