Type pylib/anki/schedv2.py

This commit is contained in:
Alan Du 2020-02-27 19:24:38 -05:00
parent b451f4e3f2
commit 2879dc1158

View file

@ -128,14 +128,15 @@ class Scheduler:
self._restorePreviewCard(card) self._restorePreviewCard(card)
self._removeFromFiltered(card) self._removeFromFiltered(card)
def counts(self, card: None = None) -> tuple: def counts(self, card: Optional[Card] = None) -> Tuple[int, int, int]:
counts = [self.newCount, self.lrnCount, self.revCount] counts = [self.newCount, self.lrnCount, self.revCount]
if card: if card:
idx = self.countIdx(card) idx = self.countIdx(card)
counts[idx] += 1 counts[idx] += 1
return tuple(counts) new, lrn, rev = counts
return (new, lrn, rev)
def dueForecast(self, days=7) -> List: def dueForecast(self, days: int = 7) -> List[Any]:
"Return counts over next DAYS. Includes today." "Return counts over next DAYS. Includes today."
daysd = dict( daysd = dict(
self.col.db.all( self.col.db.all(
@ -158,7 +159,7 @@ order by due"""
ret = [x[1] for x in sorted(daysd.items())] ret = [x[1] for x in sorted(daysd.items())]
return ret return ret
def countIdx(self, card: Card) -> Any: def countIdx(self, card: Card) -> int:
if card.queue in (QUEUE_TYPE_DAY_LEARN_RELEARN, QUEUE_TYPE_PREVIEW): if card.queue in (QUEUE_TYPE_DAY_LEARN_RELEARN, QUEUE_TYPE_PREVIEW):
return 1 return 1
return card.queue return card.queue
@ -179,7 +180,7 @@ order by due"""
g[key][1] += cnt g[key][1] += cnt
self.col.decks.save(g) self.col.decks.save(g)
def extendLimits(self, new, rev) -> None: def extendLimits(self, new: int, rev: int) -> None:
cur = self.col.decks.current() cur = self.col.decks.current()
parents = self.col.decks.parents(cur["id"]) parents = self.col.decks.parents(cur["id"])
children = [ children = [
@ -193,8 +194,10 @@ order by due"""
self.col.decks.save(g) self.col.decks.save(g)
def _walkingCount( def _walkingCount(
self, limFn: Optional[Callable] = None, cntFn: Optional[Callable] = None self,
) -> Any: limFn: Optional[Callable[[Any], Optional[int]]] = None,
cntFn: Optional[Callable[[int, int], int]] = None,
) -> int:
tot = 0 tot = 0
pcounts: Dict[int, int] = {} pcounts: Dict[int, int] = {}
# for each of the active decks # for each of the active decks
@ -228,7 +231,7 @@ order by due"""
# Deck list # Deck list
########################################################################## ##########################################################################
def deckDueList(self) -> List[list]: def deckDueList(self) -> List[List[Any]]:
"Returns [deckname, did, rev, lrn, new]" "Returns [deckname, did, rev, lrn, new]"
self._checkDay() self._checkDay()
self.col.decks.checkIntegrity() self.col.decks.checkIntegrity()
@ -270,9 +273,7 @@ order by due"""
def deckDueTree(self) -> Any: def deckDueTree(self) -> Any:
return self._groupChildren(self.deckDueList()) return self._groupChildren(self.deckDueList())
def _groupChildren( def _groupChildren(self, grps: List[List[Any]]) -> Any:
self, grps: List[List]
) -> Tuple[Tuple[Any, Any, Any, Any, Any, Any], ...]:
# first, split the group names into components # first, split the group names into components
for g in grps: for g in grps:
g[0] = g[0].split("::") g[0] = g[0].split("::")
@ -281,7 +282,7 @@ order by due"""
# then run main function # then run main function
return self._groupChildrenMain(grps) return self._groupChildrenMain(grps)
def _groupChildrenMain(self, grps: Any) -> Any: def _groupChildrenMain(self, grps: List[List[Any]]) -> Any:
tree = [] tree = []
# group and recurse # group and recurse
def key(grp): def key(grp):
@ -379,7 +380,7 @@ did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
self._newQueue: List[int] = [] self._newQueue: List[int] = []
self._updateNewCardRatio() self._updateNewCardRatio()
def _fillNew(self) -> Any: def _fillNew(self) -> Optional[bool]:
if self._newQueue: if self._newQueue:
return True return True
if not self.newCount: if not self.newCount:
@ -406,6 +407,7 @@ did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
# removed from the queue but not buried # removed from the queue but not buried
self._resetNew() self._resetNew()
return self._fillNew() return self._fillNew()
return None
def _getNewCard(self) -> Optional[Card]: def _getNewCard(self) -> Optional[Card]:
if self._fillNew(): if self._fillNew():
@ -423,7 +425,7 @@ did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
return return
self.newCardModulus = 0 self.newCardModulus = 0
def _timeForNewCard(self) -> Optional[int]: def _timeForNewCard(self) -> Optional[bool]:
"True if it's time to display a new card when distributing." "True if it's time to display a new card when distributing."
if not self.newCount: if not self.newCount:
return False return False
@ -432,10 +434,10 @@ did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
elif self.col.conf["newSpread"] == NEW_CARDS_FIRST: elif self.col.conf["newSpread"] == NEW_CARDS_FIRST:
return True return True
elif self.newCardModulus: elif self.newCardModulus:
return self.reps and self.reps % self.newCardModulus == 0 return self.reps != 0 and self.reps % self.newCardModulus == 0
else: else:
# shouldn't reach # shouldn't reach
return False return None
def _deckNewLimit( def _deckNewLimit(
self, did: int, fn: Callable[[Dict[str, Any]], int] = None self, did: int, fn: Callable[[Dict[str, Any]], int] = None
@ -453,7 +455,7 @@ did = ? and queue = {QUEUE_TYPE_NEW} limit ?)""",
lim = min(rem, lim) lim = min(rem, lim)
return lim return lim
def _newForDeck(self, did: int, lim: int) -> Any: def _newForDeck(self, did: int, lim: int) -> int:
"New count for a single deck." "New count for a single deck."
if not lim: if not lim:
return 0 return 0
@ -466,14 +468,14 @@ select count() from
lim, lim,
) )
def _deckNewLimitSingle(self, g: Dict[str, Any]) -> Any: 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"]:
return self.dynReportLimit return self.dynReportLimit
c = self.col.decks.confForDid(g["id"]) c = self.col.decks.confForDid(g["id"])
return max(0, c["new"]["perDay"] - g["newToday"][1]) return max(0, c["new"]["perDay"] - g["newToday"][1])
def totalNewForCurrentDeck(self) -> Any: def totalNewForCurrentDeck(self) -> int:
return self.col.db.scalar( return self.col.db.scalar(
f""" f"""
select count() from cards where id in ( select count() from cards where id in (
@ -533,7 +535,7 @@ select count() from cards where did in %s and queue = {QUEUE_TYPE_PREVIEW}
self._lrnDids = self.col.decks.active()[:] self._lrnDids = self.col.decks.active()[:]
# sub-day learning # sub-day learning
def _fillLrn(self) -> Any: def _fillLrn(self) -> Union[bool, List[Any]]:
if not self.lrnCount: if not self.lrnCount:
return False return False
if self._lrnQueue: if self._lrnQueue:
@ -745,10 +747,7 @@ did = ? and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit ?""",
return tot + tod * 1000 return tot + tod * 1000
def _leftToday( def _leftToday(
self, self, delays: List[int], left: int, now: Optional[int] = None,
delays: Union[List[int], List[Union[float, int]]],
left: int,
now: Optional[int] = None,
) -> int: ) -> int:
"The number of steps that can be completed by the day cutoff." "The number of steps that can be completed by the day cutoff."
if not now: if not now:
@ -803,7 +802,7 @@ did = ? and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit ?""",
else: else:
ivl = -self._delayForGrade(conf, card.left) ivl = -self._delayForGrade(conf, card.left)
def log(): def log() -> None:
self.col.db.execute( self.col.db.execute(
"insert into revlog values (?,?,?,?,?,?,?,?,?)", "insert into revlog values (?,?,?,?,?,?,?,?,?)",
int(time.time() * 1000), int(time.time() * 1000),
@ -824,7 +823,7 @@ did = ? and queue = {QUEUE_TYPE_DAY_LEARN_RELEARN} and due <= ? limit ?""",
time.sleep(0.01) time.sleep(0.01)
log() log()
def _lrnForDeck(self, did: int) -> Any: def _lrnForDeck(self, did: int) -> int:
cnt = ( cnt = (
self.col.db.scalar( self.col.db.scalar(
f""" f"""
@ -849,13 +848,13 @@ and due <= ? limit ?)""",
# Reviews # Reviews
########################################################################## ##########################################################################
def _currentRevLimit(self) -> Any: def _currentRevLimit(self) -> int:
d = self.col.decks.get(self.col.decks.selected(), default=False) d = self.col.decks.get(self.col.decks.selected(), default=False)
return self._deckRevLimitSingle(d) return self._deckRevLimitSingle(d)
def _deckRevLimitSingle( def _deckRevLimitSingle(
self, d: Dict[str, Any], parentLimit: Optional[int] = None self, d: Dict[str, Any], parentLimit: Optional[int] = None
) -> Any: ) -> int:
# invalid deck selected? # invalid deck selected?
if not d: if not d:
return 0 return 0