GitHub/Anki
1
0
Fork 0
mirror of https://github.com/ankitects/anki.git synced 2025-09-24 16:56:36 -04:00
Code Issues Projects Releases Packages 1 Wiki Activity Actions

remove duplicate methods from sched.py

Browse source
This commit is contained in:
Damien Elmes 2020-03-04 11:50:10 +10:00
parent 1c09d80558
commit 961026b04b
1 changed files with 1 additions and 555 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

556
pylib/anki/sched.py
View file

@ -9,14 +9,12 @@ import time
import weakref import weakref
from heapq import * from heapq import *
from operator import itemgetter from operator import itemgetter
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union from typing import Any, Callable, Dict, List, Optional, Tuple, Union
import anki import anki
from anki import hooks from anki import hooks
from anki.cards import Card from anki.cards import Card
from anki.consts import * from anki.consts import *
from anki.lang import _
from anki.rsbackend import FormatTimeSpanContext
from anki.schedv2 import Scheduler as V2 from anki.schedv2 import Scheduler as V2
from anki.utils import ids2str, intTime from anki.utils import ids2str, intTime
@ -45,28 +43,6 @@ class Scheduler(V2):
self._haveQueues = False self._haveQueues = False
self._updateCutoff() self._updateCutoff()
def getCard(self) -> Optional[Card]:
"Pop the next card from the queue. None if finished."
self._checkDay()
if not self._haveQueues:
self.reset()
card = self._getCard()
if card:
self.col.log(card)
if not self._burySiblingsOnAnswer:
self._burySiblings(card)
self.reps += 1
card.startTimer()
return card
return None
def reset(self) -> None:
self._updateCutoff()
self._resetLrn()
self._resetRev()
self._resetNew()
self._haveQueues = True
def answerCard(self, card: Card, ease: int) -> None: def answerCard(self, card: Card, ease: int) -> None:
self.col.log() self.col.log()
assert 1 <= ease <= 4 assert 1 <= ease <= 4
@ -118,29 +94,6 @@ class Scheduler(V2):
new, lrn, rev = counts new, lrn, rev = counts
return (new, lrn, rev) return (new, lrn, rev)
def dueForecast(self, days: int = 7) -> List[Any]:
"Return counts over next DAYS. Includes today."
daysd = dict(
self.col.db.all(
f"""
select due, count() from cards
where did in %s and queue = {QUEUE_TYPE_REV}
and due between ? and ?
group by due
order by due"""
% self._deckLimit(),
self.today,
self.today + days - 1,
)
)
for d in range(days):
d = self.today + d
if d not in daysd:
daysd[d] = 0
# return in sorted order
ret = [x[1] for x in sorted(daysd.items())]
return ret
def countIdx(self, card: Card) -> int: def countIdx(self, card: Card) -> int:
if card.queue == QUEUE_TYPE_DAY_LEARN_RELEARN: if card.queue == QUEUE_TYPE_DAY_LEARN_RELEARN:
return QUEUE_TYPE_LRN return QUEUE_TYPE_LRN
@ -187,64 +140,6 @@ order by due"""
self.col.usn(), self.col.usn(),
) )
# Rev/lrn/time daily stats
##########################################################################
def _updateStats(self, card: Card, type: str, cnt: int = 1) -> None:
key = type + "Today"
for g in [self.col.decks.get(card.did)] + self.col.decks.parents(card.did):
# add
g[key][1] += cnt
self.col.decks.save(g)
def extendLimits(self, new: int, rev: int) -> None:
cur = self.col.decks.current()
parents = self.col.decks.parents(cur["id"])
children = [
self.col.decks.get(did)
for (name, did) in self.col.decks.children(cur["id"])
]
for g in [cur] + parents + children:
# add
g["newToday"][1] -= new
g["revToday"][1] -= rev
self.col.decks.save(g)
def _walkingCount(
self,
limFn: Optional[Callable[[Any], Optional[int]]] = None,
cntFn: Optional[Callable[[int, int], int]] = None,
) -> int:
tot = 0
pcounts: Dict[int, int] = {}
# for each of the active decks
nameMap = self.col.decks.nameMap()
for did in self.col.decks.active():
# early alphas were setting the active ids as a str
did = int(did)
# get the individual deck's limit
lim = limFn(self.col.decks.get(did))
if not lim:
continue
# check the parents
parents = self.col.decks.parents(did, nameMap)
for p in parents:
# add if missing
if p["id"] not in pcounts:
pcounts[p["id"]] = limFn(p)
# take minimum of child and parent
lim = min(pcounts[p["id"]], lim)
# see how many cards we actually have
cnt = cntFn(did, lim)
# if non-zero, decrement from parent counts
for p in parents:
pcounts[p["id"]] -= cnt
# we may also be a parent
pcounts[did] = lim - cnt
# and add to running total
tot += cnt
return tot
# Deck list # Deck list
########################################################################## ##########################################################################
@ -284,18 +179,6 @@ order by due"""
lims[deck["name"]] = [nlim, rlim] lims[deck["name"]] = [nlim, rlim]
return data return data
def deckDueTree(self) -> Any:
return self._groupChildren(self.deckDueList())
def _groupChildren(self, grps: List[List[Any]]) -> Any:
# first, split the group names into components
for g in grps:
g[0] = g[0].split("::")
# and sort based on those components
grps.sort(key=itemgetter(0))
# then run main function
return self._groupChildrenMain(grps)
def _groupChildrenMain(self, grps: List[List[Any]]) -> Any: def _groupChildrenMain(self, grps: List[List[Any]]) -> Any:
tree = [] tree = []
# group and recurse # group and recurse
@ -367,108 +250,6 @@ order by due"""
# New cards # New cards
########################################################################## ##########################################################################
def _resetNewCount(self) -> None:
cntFn = lambda did, lim: self.col.db.scalar(
f"""
select count() from (select 1 from cards where
did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
did,
lim,
)
self.newCount = self._walkingCount(self._deckNewLimitSingle, cntFn)
def _resetNew(self) -> None:
self._resetNewCount()
self._newDids = self.col.decks.active()[:]
self._newQueue: List[Any] = []
self._updateNewCardRatio()
def _fillNew(self) -> Optional[bool]:
if self._newQueue:
return True
if not self.newCount:
return False
while self._newDids:
did = self._newDids[0]
lim = min(self.queueLimit, self._deckNewLimit(did))
if lim:
# fill the queue with the current did
self._newQueue = self.col.db.list(
f"""
select id from cards where did = ? and queue = {QUEUE_TYPE_NEW} order by due,ord limit ?""",
did,
lim,
)
if self._newQueue:
self._newQueue.reverse()
return True
# nothing left in the deck; move to next
self._newDids.pop(0)
if self.newCount:
# 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._resetNew()
return self._fillNew()
return None
def _getNewCard(self) -> Optional[Card]:
if self._fillNew():
self.newCount -= 1
return self.col.getCard(self._newQueue.pop())
return None
def _updateNewCardRatio(self) -> None:
if self.col.conf["newSpread"] == NEW_CARDS_DISTRIBUTE:
if self.newCount:
self.newCardModulus = (self.newCount + self.revCount) // self.newCount
# if there are cards to review, ensure modulo >= 2
if self.revCount:
self.newCardModulus = max(2, self.newCardModulus)
return
self.newCardModulus = 0
def _timeForNewCard(self) -> Optional[bool]:
"True if it's time to display a new card when distributing."
if not self.newCount:
return False
if self.col.conf["newSpread"] == NEW_CARDS_LAST:
return False
elif self.col.conf["newSpread"] == NEW_CARDS_FIRST:
return True
elif self.newCardModulus:
return self.reps != 0 and self.reps % self.newCardModulus == 0
return None
def _deckNewLimit(
self, did: int, fn: Optional[Callable[[Dict[str, Any]], int]] = None
) -> int:
if not fn:
fn = self._deckNewLimitSingle
sel = self.col.decks.get(did)
lim = -1
# for the deck and each of its parents
for g in [sel] + self.col.decks.parents(did):
rem = fn(g)
if lim == -1:
lim = rem
else:
lim = min(rem, lim)
return lim
def _newForDeck(self, did: int, lim: int) -> int:
"New count for a single deck."
if not lim:
return 0
lim = min(lim, self.reportLimit)
return self.col.db.scalar(
f"""
select count() from
(select 1 from cards where did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
did,
lim,
)
def _deckNewLimitSingle(self, g: Dict[str, Any]) -> int: def _deckNewLimitSingle(self, g: Dict[str, Any]) -> int:
"Limit for deck without parent limits." "Limit for deck without parent limits."
if g["dyn"]: if g["dyn"]:
@ -545,42 +326,6 @@ limit %d"""
return card return card
return None return None
# daily learning
def _fillLrnDay(self) -> Optional[bool]:
if not self.lrnCount:
return False
if self._lrnDayQueue:
return True
while self._lrnDids:
did = self._lrnDids[0]
# fill the queue with the current did
self._lrnDayQueue = self.col.db.list(
f"""
select id from cards where
did = ? and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit ?""",
did,
self.today,
self.queueLimit,
)
if self._lrnDayQueue:
# order
r = random.Random()
r.seed(self.today)
r.shuffle(self._lrnDayQueue)
# is the current did empty?
if len(self._lrnDayQueue) < self.queueLimit:
self._lrnDids.pop(0)
return True
# nothing left in the deck; move to next
self._lrnDids.pop(0)
return None
def _getLrnDayCard(self) -> Optional[Card]:
if self._fillLrnDay():
self.lrnCount -= 1
return self.col.getCard(self._lrnDayQueue.pop())
return None
def _answerLrnCard(self, card: Card, ease: int) -> None: def _answerLrnCard(self, card: Card, ease: int) -> None:
# ease 1=no, 2=yes, 3=remove # ease 1=no, 2=yes, 3=remove
conf = self._lrnConf(card) conf = self._lrnConf(card)
@ -643,18 +388,6 @@ did = ? and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit ?""",
card.queue = QUEUE_TYPE_DAY_LEARN_RELEARN card.queue = QUEUE_TYPE_DAY_LEARN_RELEARN
self._logLrn(card, ease, conf, leaving, type, lastLeft) self._logLrn(card, ease, conf, leaving, type, lastLeft)
def _delayForGrade(self, conf: Dict[str, Any], left: int) -> float:
left = left % 1000
try:
delay = conf["delays"][-left]
except IndexError:
if conf["delays"]:
delay = conf["delays"][0]
else:
# user deleted final step; use dummy value
delay = 1
return delay * 60
def _lrnConf(self, card: Card) -> Dict[str, Any]: def _lrnConf(self, card: Card) -> Dict[str, Any]:
if card.type == CARD_TYPE_REV: if card.type == CARD_TYPE_REV:
return self._lapseConf(card) return self._lapseConf(card)
@ -900,12 +633,6 @@ did = ? and queue = {QUEUE_TYPE_REV} and due <= ? limit ?""",
return None return None
def _getRevCard(self) -> Optional[Card]:
if self._fillRev():
self.revCount -= 1
return self.col.getCard(self._revQueue.pop())
return None
def totalRevForCurrentDeck(self) -> int: def totalRevForCurrentDeck(self) -> int:
return self.col.db.scalar( return self.col.db.scalar(
f""" f"""
@ -1025,35 +752,11 @@ select id from cards where did in %s and queue = {QUEUE_TYPE_REV} and due <= ? l
# interval capped? # interval capped?
return min(interval, conf["maxIvl"]) return min(interval, conf["maxIvl"])
def _fuzzedIvl(self, ivl: int) -> int:
min, max = self._fuzzIvlRange(ivl)
return random.randint(min, max)
def _fuzzIvlRange(self, ivl: int) -> List[int]:
if ivl < 2:
return [1, 1]
elif ivl == 2:
return [2, 3]
elif ivl < 7:
fuzz = int(ivl * 0.25)
elif ivl < 30:
fuzz = max(2, int(ivl * 0.15))
else:
fuzz = max(4, int(ivl * 0.05))
# fuzz at least a day
fuzz = max(fuzz, 1)
return [ivl - fuzz, ivl + fuzz]
def _constrainedIvl(self, ivl: float, conf: Dict[str, Any], prev: int) -> int: # type: ignore[override] 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." "Integer interval after interval factor and prev+1 constraints applied."
new = ivl * conf.get("ivlFct", 1) new = ivl * conf.get("ivlFct", 1)
return int(max(new, prev + 1)) return int(max(new, prev + 1))
def _daysLate(self, card: Card) -> int:
"Number of days later than scheduled."
due = card.odue if card.odid else card.due
return max(0, self.today - due)
def _updateRevIvl(self, card: Card, ease: int) -> None: def _updateRevIvl(self, card: Card, ease: int) -> None:
idealIvl = self._nextRevIvl(card, ease) idealIvl = self._nextRevIvl(card, ease)
card.ivl = min( card.ivl = min(
@ -1113,9 +816,6 @@ due = odue, odue = 0, odid = 0, usn = ? where %s"""
self.col.usn(), self.col.usn(),
) )
def remFromDyn(self, cids: List[int]) -> None:
self.emptyDyn(None, "id in %s and odid" % ids2str(cids))
def _dynOrder(self, o: int, l: int) -> str: def _dynOrder(self, o: int, l: int) -> str:
if o == DYN_OLDEST: if o == DYN_OLDEST:
t = "(select max(id) from revlog where cid=c.id)" t = "(select max(id) from revlog where cid=c.id)"
@ -1251,17 +951,6 @@ did = ?, queue = %s, due = ?, usn = ? where id = ?"""
resched=conf["resched"], resched=conf["resched"],
) )
def _revConf(self, card: Card) -> Dict[str, Any]:
conf = self._cardConf(card)
# normal deck
if not card.odid:
return conf["rev"]
# dynamic deck
return self.col.decks.confForDid(card.odid)["rev"]
def _deckLimit(self) -> str:
return ids2str(self.col.decks.active())
def _resched(self, card: Card) -> bool: def _resched(self, card: Card) -> bool:
conf = self._cardConf(card) conf = self._cardConf(card)
if not conf["dyn"]: if not conf["dyn"]:
@ -1294,87 +983,9 @@ did = ?, queue = %s, due = ?, usn = ? where id = ?"""
if unburied < self.today: if unburied < self.today:
self.unburyCards() self.unburyCards()
def _checkDay(self) -> None:
# check if the day has rolled over
if time.time() > self.dayCutoff:
self.reset()
# Deck finished state # Deck finished state
########################################################################## ##########################################################################
def finishedMsg(self) -> str:
return (
"<b>"
+ _("Congratulations! You have finished this deck for now.")
+ "</b><br><br>"
+ self._nextDueMsg()
)
def _nextDueMsg(self) -> str:
line = []
# the new line replacements are so we don't break translations
# in a point release
if self.revDue():
line.append(
_(
"""\
Today's review limit has been reached, but there are still cards
waiting to be reviewed. For optimum memory, consider increasing
the daily limit in the options."""
).replace("\n", " ")
)
if self.newDue():
line.append(
_(
"""\
There are more new cards available, but the daily limit has been
reached. You can increase the limit in the options, but please
bear in mind that the more new cards you introduce, the higher
your short-term review workload will become."""
).replace("\n", " ")
)
if self.haveBuried():
if self.haveCustomStudy:
now = " " + _("To see them now, click the Unbury button below.")
else:
now = ""
line.append(
_(
"""\
Some related or buried cards were delayed until a later session."""
)
+ now
)
if self.haveCustomStudy and not self.col.decks.current()["dyn"]:
line.append(
_(
"""\
To study outside of the normal schedule, click the Custom Study button below."""
)
)
return "<p>".join(line)
def revDue(self) -> Optional[int]:
"True if there are any rev cards due."
return self.col.db.scalar(
(
f"select 1 from cards where did in %s and queue = {QUEUE_TYPE_REV} "
"and due <= ? limit 1"
)
% self._deckLimit(),
self.today,
)
def newDue(self) -> Optional[int]:
"True if there are any new cards due."
return self.col.db.scalar(
(
f"select 1 from cards where did in %s and queue = {QUEUE_TYPE_NEW} "
"limit 1"
)
% self._deckLimit()
)
def haveBuried(self) -> bool: def haveBuried(self) -> bool:
sdids = ids2str(self.col.decks.active()) sdids = ids2str(self.col.decks.active())
cnt = self.col.db.scalar( cnt = self.col.db.scalar(
@ -1386,18 +997,6 @@ To study outside of the normal schedule, click the Custom Study button below."""
# Next time reports # Next time reports
########################################################################## ##########################################################################
def nextIvlStr(self, card: Card, ease: int, short: bool = False) -> str:
"Return the next interval for CARD as a string."
ivl_secs = self.nextIvl(card, ease)
if not ivl_secs:
return _("(end)")
s = self.col.backend.format_time_span(
ivl_secs, FormatTimeSpanContext.ANSWER_BUTTONS
)
if ivl_secs < self.col.conf["collapseTime"]:
s = "<" + s
return s
def nextIvl(self, card: Card, ease: int) -> float: def nextIvl(self, card: Card, ease: int) -> float:
"Return the next interval for CARD, in seconds." "Return the next interval for CARD, in seconds."
if card.queue in (QUEUE_TYPE_NEW, QUEUE_TYPE_LRN, QUEUE_TYPE_DAY_LEARN_RELEARN): if card.queue in (QUEUE_TYPE_NEW, QUEUE_TYPE_LRN, QUEUE_TYPE_DAY_LEARN_RELEARN):
@ -1472,13 +1071,6 @@ update cards set queue={QUEUE_TYPE_SIBLING_BURIED},mod=?,usn=? where id in """
self.col.usn(), self.col.usn(),
) )
def buryNote(self, nid: int) -> None:
"Bury all cards for note until next session."
cids = self.col.db.list(
"select id from cards where nid = ? and queue >= 0", nid
)
self.buryCards(cids)
# Sibling spacing # Sibling spacing
########################################################################## ##########################################################################
@ -1522,149 +1114,3 @@ and (queue={QUEUE_TYPE_NEW} or (queue={QUEUE_TYPE_REV} and due<=?))""",
self.col.usn(), self.col.usn(),
) )
self.col.log(toBury) self.col.log(toBury)
# Resetting
##########################################################################
def forgetCards(self, ids: List[int]) -> None:
"Put cards at the end of the new queue."
self.remFromDyn(ids)
self.col.db.execute(
f"update cards set type={CARD_TYPE_NEW},queue={QUEUE_TYPE_NEW},ivl=0,due=0,odue=0,factor=?"
" where id in " + ids2str(ids),
STARTING_FACTOR,
)
pmax = (
self.col.db.scalar(f"select max(due) from cards where type={CARD_TYPE_NEW}")
or 0
)
# takes care of mod + usn
self.sortCards(ids, start=pmax + 1)
self.col.log(ids)
def reschedCards(self, ids: List[int], imin: int, imax: int) -> None:
"Put cards in review queue with a new interval in days (min, max)."
d = []
t = self.today
mod = intTime()
for id in ids:
r = random.randint(imin, imax)
d.append(
dict(
id=id,
due=r + t,
ivl=max(1, r),
mod=mod,
usn=self.col.usn(),
fact=STARTING_FACTOR,
)
)
self.remFromDyn(ids)
self.col.db.executemany(
f"""
update cards set type={CARD_TYPE_REV},queue={QUEUE_TYPE_REV},ivl=:ivl,due=:due,odue=0,
usn=:usn,mod=:mod,factor=:fact where id=:id""",
d,
)
self.col.log(ids)
def resetCards(self, ids: List[int]) -> None:
"Completely reset cards for export."
sids = ids2str(ids)
# we want to avoid resetting due number of existing new cards on export
nonNew = self.col.db.list(
f"select id from cards where id in %s and (queue != {QUEUE_TYPE_NEW} or type != {CARD_TYPE_NEW})"
% sids
)
# reset all cards
self.col.db.execute(
f"update cards set reps=0,lapses=0,odid=0,odue=0,queue={QUEUE_TYPE_NEW}"
" where id in %s" % sids
)
# and forget any non-new cards, changing their due numbers
self.forgetCards(nonNew)
self.col.log(ids)
# Repositioning new cards
##########################################################################
def sortCards(
self,
cids: List[int],
start: int = 1,
step: int = 1,
shuffle: bool = False,
shift: bool = False,
) -> None:
scids = ids2str(cids)
now = intTime()
nids = []
nidsSet: Set[int] = set()
for id in cids:
nid = self.col.db.scalar("select nid from cards where id = ?", id)
if nid not in nidsSet:
nids.append(nid)
nidsSet.add(nid)
if not nids:
# no new cards
return
# determine nid ordering
due = {}
if shuffle:
random.shuffle(nids)
for c, nid in enumerate(nids):
due[nid] = start + c * step
# pylint: disable=undefined-loop-variable
high = start + c * step
# shift?
if shift:
low = self.col.db.scalar(
f"select min(due) from cards where due >= ? and type = {CARD_TYPE_NEW} "
"and id not in %s" % scids,
start,
)
if low is not None:
shiftby = high - low + 1
self.col.db.execute(
f"""
update cards set mod=?, usn=?, due=due+? where id not in %s
and due >= ? and queue = {QUEUE_TYPE_NEW}"""
% scids,
now,
self.col.usn(),
shiftby,
low,
)
# reorder cards
d = []
for id, nid in self.col.db.execute(
f"select id, nid from cards where type = {CARD_TYPE_NEW} and id in " + scids
):
d.append(dict(now=now, due=due[nid], usn=self.col.usn(), cid=id))
self.col.db.executemany(
"update cards set due=:due,mod=:now,usn=:usn where id = :cid", d
)
def randomizeCards(self, did: int) -> None:
cids = self.col.db.list("select id from cards where did = ?", did)
self.sortCards(cids, shuffle=True)
def orderCards(self, did: int) -> None:
cids = self.col.db.list("select id from cards where did = ? order by nid", did)
self.sortCards(cids)
def resortConf(self, conf) -> None:
for did in self.col.decks.didsForConf(conf):
if conf["new"]["order"] == 0:
self.randomizeCards(did)
else:
self.orderCards(did)
# for post-import
def maybeRandomizeDeck(self, did: Optional[int] = None) -> None:
if not did:
did = self.col.decks.selected()
conf = self.col.decks.confForDid(did)
# in order due?
if conf["new"]["order"] == NEW_CARDS_RANDOM:
self.randomizeCards(did)