intercal/omnibot22
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

markov: Add utility method for last access time update

Browse Source 
* Chain.__touch() is a new function that updates the last time a markov
  chain was accessed
* Fix a bug that would not reliably update the last access time of the
  chain during Chain.add()

Signed-off-by: Alek Ratzloff <alekratz@gmail.com>
This commit is contained in:
Alek Ratzloff
2022-06-03 18:18:18 -07:00
parent a823871039
commit da7d1501e2
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
plugins/markov.py
View File

@@ -44,11 +44,15 @@ class Chain:
self.__last_access = 0.0 self.__last_access = 0.0
self.__dirty = False self.__dirty = False
def __touch(self):
self.__last_access = asyncio.get_running_loop().time()
@property @property
def last_access(self) -> float: def last_access(self) -> float:
return self.__last_access return self.__last_access
def add(self, text: str): def add(self, text: str):
self.__touch()
parts: List[Any] = text.strip().split() parts: List[Any] = text.strip().split()
if not parts: if not parts:
return return
@@ -60,7 +64,7 @@ class Chain:
self.__dirty = True self.__dirty = True
def get(self, key: str) -> dict[str | None, int]: def get(self, key: str) -> dict[str | None, int]:
self.__last_access = asyncio.get_running_loop().time() self.__touch()
if self.__cache: if self.__cache:
if key in self.__cache: if key in self.__cache:
return self.__cache[key] return self.__cache[key]
@@ -72,7 +76,7 @@ class Chain:
return self.__cache[key] return self.__cache[key]
def set(self, key: str, value: Mapping[str | None, int]): def set(self, key: str, value: Mapping[str | None, int]):
self.__last_access = asyncio.get_running_loop().time() self.__touch()
if not self.__cache: if not self.__cache:
# Attempt the cache before writing to it # Attempt the cache before writing to it
self.__load() self.__load()
@@ -80,7 +84,7 @@ class Chain:
self.__dirty = True self.__dirty = True
def __load(self): def __load(self):
self.__last_access = asyncio.get_running_loop().time() self.__touch()
if self.__cache: if self.__cache:
return return
if not self.path.exists(): if not self.path.exists():