fix: memory leak, use ruff

VERSION

@@ -1 +1 @@
-1.7.0
+1.7.1

pyproject.toml

@@ -41,57 +41,13 @@ version = {file = "VERSION"}
 
 [project.optional-dependencies]
 dev = [
-    "autoflake",
-    "bandit",
-    "black",
-    "flake8",
-    "flake8-bugbear",
-    "flake8-quotes",
-    "isort",
+    "ruff",
+    "memory_profiler",
     "mypy",
     "pip-tools",
     "pyupgrade",
 ]
 
-[tool.black]
-line-length = 80
-target-version = ['py310']
-exclude = '''
-
-(
-  /(
-      \.eggs         # exclude a few common directories in the
-    | \.git          # root of the project
-    | \.hg
-    | \.mypy_cache
-    | \.tox
-    | \.venv
-    | _build
-    | buck-out
-    | build
-    | dist
-  )/
-  | es_utils
-  | alembic
-  | documentation
-)
-'''
-
-[tool.bandit]
-exclude_dirs = ["./.git", "./documentation", "./tests", "./.venv", "./build"]
-skips = []
-
-[tool.isort]
-src_paths = "src,tests"
-profile = "black"
-line_length = 80
-sections="FUTURE,STDLIB,THIRDPARTY,FIRSTPARTY,LOCALFOLDER"
-no_lines_before = "LOCALFOLDER"
-multi_line_output = 3
-include_trailing_comma = true
-use_parentheses = true
-force_grid_wrap = 0
-
 [tool.mypy]
 python_version = "3.10"
 ignore_missing_imports = true
@@ -105,5 +61,57 @@ no_implicit_optional = true
 show_error_codes = true
 files = "src"
 
-[tool.pytest.ini_config]
-log_level = "INFO"
+[tool.ruff]
+target-version = "py310"
+line-length = 120
+exclude = [
+    ".bzr",
+    ".git",
+    ".svn",
+    ".hg",
+    "typings"
+]
+
+[tool.ruff.lint]
+# https://docs.astral.sh/ruff/rules/
+select = [
+    "E", # pycodestyle
+    "F", # pyflakes
+    "I", # isort
+    "Q", # flake8-quotes-q
+    "S",  # flake8-bandit-s
+    "UP", # pyupgrade
+    "W",  # pycodestyle
+]
+
+[tool.ruff.lint.flake8-quotes]
+docstring-quotes = "double"
+
+[tool.ruff.lint.isort]
+combine-as-imports = true
+force-wrap-aliases = true
+forced-separate = ["tests"]
+known-local-folder = [
+    "torchlight",
+    "src",
+    "tests"
+]
+known-first-party = [
+]
+# required-imports = ["from __future__ import annotations"]
+
+[tool.ruff.lint.per-file-ignores]
+"*/__init__.py" = [
+    "F401"
+]
+"alembic/*" = [
+    "S608", # Possible SQL injection vector through string-based query construction
+]
+"tests/**/*.py" = [
+    "S101", # asserts allowed in tests
+    "S311", # Standard pseudo-random generators are not suitable for cryptographic purposes
+    # "S603", # `subprocess` call: check for execution of untrusted input
+]
+
+[tool.ruff.lint.pycodestyle]
+max-line-length = 120

src/torchlight/AccessManager.py

@@ -8,15 +8,11 @@
 
 
 class AccessManager:
-    def __init__(
-        self, config_folder: str, config_filename: str = "admins.json"
-    ) -> None:
+    def __init__(self, config_folder: str, config_filename: str = "admins.json") -> None:
         self.logger = logging.getLogger(self.__class__.__name__)
         self.config_folder = os.path.abspath(config_folder)
         self.config_filename = config_filename
-        self.config_filepath = os.path.abspath(
-            os.path.join(config_folder, config_filename)
-        )
+        self.config_filepath = os.path.abspath(os.path.join(config_folder, config_filename))
         self.access_dict: OrderedDict = OrderedDict()
         self.admins: list[SourcemodAdmin] = []
 
@@ -25,6 +21,7 @@ def Load(self) -> None:
 
         with open(self.config_filepath) as fp:
             self.access_dict = json.load(fp, object_pairs_hook=OrderedDict)
+            self.admins.clear()
             for admin_dict in self.access_dict["admins"]:
                 self.admins.append(
                     SourcemodAdmin(
@@ -39,9 +36,7 @@ def Load(self) -> None:
         self.logger.info(f"Loaded {self.admins}")
 
     def Save(self) -> None:
-        self.logger.info(
-            f"Saving {len(self.admins)} admin access to {self.config_filepath}"
-        )
+        self.logger.info(f"Saving {len(self.admins)} admin access to {self.config_filepath}")
 
         for admin in self.admins:
             admin_cfg = {
@@ -62,9 +57,7 @@ def Save(self) -> None:
             else:
                 self.access_dict["admins"][index] = admin_cfg
 
-        self.access_dict["admins"] = sorted(
-            self.access_dict["admins"], key=lambda x: x["level"], reverse=True
-        )
+        self.access_dict["admins"] = sorted(self.access_dict["admins"], key=lambda x: x["level"], reverse=True)
 
         with open(self.config_filepath, "w") as fp:
             json.dump(self.access_dict, fp, indent="\t")

src/torchlight/Advertiser.py

@@ -24,10 +24,7 @@ def Think(self, delta: int) -> None:
             if not clip["timestamp"]:
                 continue
 
-            if (
-                clip["timestamp"] + clip["duration"] + self.config["MaxSpan"]
-                < now
-            ):
+            if clip["timestamp"] + clip["duration"] + self.config["MaxSpan"] < now:
                 if not clip["active"]:
                     del self.last_clips[key]
                 continue
@@ -36,14 +33,8 @@ def Think(self, delta: int) -> None:
 
         self.next_ad_stop -= delta
         ceil_duration = math.ceil(duration)
-        if (
-            ceil_duration > self.ad_stop
-            and self.next_ad_stop <= 0
-            and ceil_duration % self.config["AdStop"] == 0
-        ):
-            self.torchlight.SayChat(
-                "Hint: Type {darkred}!stop{default} to stop all currently playing sounds."
-            )
+        if ceil_duration > self.ad_stop and self.next_ad_stop <= 0 and ceil_duration % self.config["AdStop"] == 0:
+            self.torchlight.SayChat("Hint: Type {darkred}!stop{default} to stop all currently playing sounds.")
             self.ad_stop = ceil_duration
             self.next_ad_stop = 0
         elif ceil_duration < self.ad_stop:
@@ -83,9 +74,7 @@ def OnStop(self, clip: AudioClip) -> None:
 
         self.last_clips[hash(clip)]["dominant"] = False
 
-    def OnUpdate(
-        self, clip: AudioClip, old_position: int, new_position: int
-    ) -> None:
+    def OnUpdate(self, clip: AudioClip, old_position: int, new_position: int) -> None:
         delta = new_position - old_position
         last_clip = self.last_clips[hash(clip)]
 

src/torchlight/AntiSpam.py

@@ -23,11 +23,10 @@ def CheckAntiSpam(self, player: Player) -> bool:
             and self.disabled_time > self.torchlight.loop.time()
             and player.admin.level < self.config["ImmunityLevel"]
         ):
+            cooldown = math.ceil(self.disabled_time - self.torchlight.loop.time())
             self.torchlight.SayPrivate(
                 player,
-                "Torchlight is currently on cooldown! ({} seconds left)".format(
-                    math.ceil(self.disabled_time - self.torchlight.loop.time())
-                ),
+                f"Torchlight is currently on cooldown! ({cooldown} seconds left)",
             )
             return False
 
@@ -41,22 +40,15 @@ def SpamCheck(self, audio_clips: list[AudioClip], delta: int) -> None:
             if not last_clip["timestamp"]:
                 continue
 
-            if (
-                last_clip["timestamp"]
-                + last_clip["duration"]
-                + self.config["MaxUsageSpan"]
-                < now
-            ):
+            if last_clip["timestamp"] + last_clip["duration"] + self.config["MaxUsageSpan"] < now:
                 if not last_clip["active"]:
                     del self.last_clips[key]
                 continue
 
             duration += last_clip["duration"]
 
         if duration > self.config["MaxUsageTime"]:
-            self.disabled_time = (
-                self.torchlight.loop.time() + self.config["PunishDelay"]
-            )
+            self.disabled_time = self.torchlight.loop.time() + self.config["PunishDelay"]
             self.torchlight.SayChat(
                 "Blocked voice commands for the next {} seconds. Used {} seconds within {} seconds.".format(
                     self.config["PunishDelay"],

src/torchlight/AsyncClient.py

@@ -29,6 +29,7 @@ def __init__(
         self.recv_future: Future | None = None
         self.callbacks: list[tuple[str, Callable]] = []
 
+    # @profile
     async def Connect(self) -> None:
         while True:
             self.logger.warn("Reconnecting...")
@@ -63,9 +64,7 @@ def OnReceive(self, data: str | bytes) -> None:
         try:
             json_obj = json.loads(data)
         except Exception:
-            self.logger.warn(
-                "OnReceive: Unable to decode data as json, skipping"
-            )
+            self.logger.warn("OnReceive: Unable to decode data as json, skipping")
             return
 
         if "method" in json_obj and json_obj["method"] == "publish":
@@ -84,9 +83,7 @@ async def Send(self, json_obj: Any) -> Any | None:
         if not self.protocol:
             return None
 
-        data = json.dumps(
-            json_obj, ensure_ascii=False, separators=(",", ":")
-        ).encode("UTF-8")
+        data = json.dumps(json_obj, ensure_ascii=False, separators=(",", ":")).encode("UTF-8")
 
         async with self.send_lock:
             if not self.protocol:

src/torchlight/AudioClip.py

@@ -56,20 +56,14 @@ def OnStop(self) -> None:
 
         if str(self.level) in self.config:
             if self.player.storage:
-                if (
-                    self.player.storage["Audio"]["TimeUsed"]
-                    >= self.config[str(self.level)]["TotalTime"]
-                ):
+                if self.player.storage["Audio"]["TimeUsed"] >= self.config[str(self.level)]["TotalTime"]:
                     self.torchlight.SayPrivate(
                         self.player,
                         "You have used up all of your free time! ({} seconds)".format(
                             self.config[str(self.level)]["TotalTime"]
                         ),
                     )
-                elif (
-                    self.player.storage["Audio"]["LastUseLength"]
-                    >= self.config[str(self.level)]["MaxLength"]
-                ):
+                elif self.player.storage["Audio"]["LastUseLength"] >= self.config[str(self.level)]["MaxLength"]:
                     self.torchlight.SayPrivate(
                         self.player,
                         "Your audio clip exceeded the maximum length! ({} seconds)".format(
@@ -86,13 +80,11 @@ def OnUpdate(self, old_position: int, new_position: int) -> None:
         self.player.storage["Audio"]["TimeUsed"] += delta
         self.player.storage["Audio"]["LastUseLength"] += delta
 
-        if not str(self.level) in self.config:
+        if str(self.level) not in self.config:
             return
 
         if (
-            self.player.storage["Audio"]["TimeUsed"]
-            >= self.config[str(self.level)]["TotalTime"]
-            or self.player.storage["Audio"]["LastUseLength"]
-            >= self.config[str(self.level)]["MaxLength"]
+            self.player.storage["Audio"]["TimeUsed"] >= self.config[str(self.level)]["TotalTime"]
+            or self.player.storage["Audio"]["LastUseLength"] >= self.config[str(self.level)]["MaxLength"]
         ):
             self.Stop()