chore: clear typehinting errors

This commit is contained in:
aclist 2026-05-15 23:26:47 +09:00
parent 2d265c4b9b
commit 156c1f8ec4
16 changed files with 83 additions and 85 deletions

View File

@ -70,8 +70,11 @@ def parse_meta(file: Path) -> ModMeta | None:
if ntok == "-":
ntok += str(lex.get_token())
elif tok == "name":
ntok = lex.get_token().split('"')[1]
v.append(ntok)
ntok = lex.get_token()
if ntok is not None:
ntok = ntok.split('"')[1]
if ntok is not None:
v.append(ntok)
meta = ModMeta(*v)
return meta
@ -99,7 +102,7 @@ def get_delimited_mods(steam_path: Path) -> list[Any]:
size = get_mod_size(mod)
# NOTE: final col is cell renderer highlight toggle
clean.append([meta.name, symlink, mod_dir, size, False])
clean.sort(key=lambda row: row[0].casefold())
clean.sort(key=lambda row: str(row[0]).casefold())
return clean

View File

@ -431,20 +431,15 @@ def get_app_path(folders_path: Path, appid: int) -> Path:
for obj in j["libraryfolders"]:
if str(appid) in j["libraryfolders"][obj]["apps"]:
app_path = j["libraryfolders"][obj]["path"]
break
if Path(app_path).exists():
break
if app_path is None:
raise AppNotInstalledError(
f"Failed to find a libraryfolder for the appid {appid}"
)
app_path = Path(app_path)
if app_path.exists() is False:
raise AppMovedError(
f"Path '{app_path}' specified in libraryfolders does not exist"
)
return app_path
return Path(app_path)
def get_pretty_version(steam_path: Path, appid: int) -> str | None:

View File

@ -86,7 +86,11 @@ class A2SInfo:
raise AttributeError("No record to convert")
ip = self.record.ip
qport = self.record.qport
return source_info_to_dict(ip, qport, self.info)
try:
return source_info_to_dict(ip, qport, self.info)
except Exception as e:
logger.warning(e)
raise e
def is_modded(self) -> bool:
if self.info is None:
@ -239,7 +243,7 @@ def parse_json(json: list) -> list:
return rows
def source_info_to_dict(ip: str, qport: int, info: "SourceInfo") -> dict[str, Any] | None:
def source_info_to_dict(ip: str, qport: int, info: "SourceInfo") -> dict[str, Any]:
try:
name = info.server_name
mapname = info.map_name
@ -268,7 +272,7 @@ def source_info_to_dict(ip: str, qport: int, info: "SourceInfo") -> dict[str, An
except Exception as e:
# TODO: generalized function
logger.critical(f"{type(e).__name__}: {e} ({ip}:{qport})")
return None
raise e
def query_direct(ip: str, qport: int, timeout: float = 3.0) -> dict[str, Any] | None:
@ -448,6 +452,8 @@ def get_rules(record: Record) -> list["DayzMod"]:
def query_playercount(record: Record) -> tuple[int, int] | None:
try:
res = query_direct(record.ip, record.qport)
if res is None:
return None
players = int(res["players"])
r = res["gametype"].split("lqs")
try:
@ -462,6 +468,8 @@ def query_playercount(record: Record) -> tuple[int, int] | None:
def query_by_ip(addr: str) -> A2SInfo:
record = short_ip_to_record(addr)
if record is None:
return A2SInfo(Record("0", 0, 0), None)
return query_by_record(record, update_gameport=True)
def query_by_id(server_id: int, key: str) -> A2SInfo:
@ -469,6 +477,8 @@ def query_by_id(server_id: int, key: str) -> A2SInfo:
Used with numeric Battlemetrics IDs
"""
record = map_id_to_record(key, server_id)
if record is None:
return A2SInfo(Record("0", 0, 0), None)
return query_by_record(record)
@ -485,6 +495,7 @@ def query_by_record(record: Record, update_gameport: bool = False) -> A2SInfo:
def short_ip_to_record(addr: str) -> Optional[Record]:
r = addr.split(":")
# TODO: raise exceptions instead of Nonetype
if len(r) != 2:
return None
return Record(r[0], 0, int(r[1]))

View File

@ -141,7 +141,8 @@ def vdf2json(path: Path) -> str:
indent -= 1
jbuf += _istr(indent, "}")
ntok = lex.get_token()
lex.push_token(ntok)
if ntok is not None:
lex.push_token(ntok)
if ntok and ntok != "}":
jbuf += ","
jbuf += "\n"
@ -151,12 +152,14 @@ def vdf2json(path: Path) -> str:
jbuf += _istr(indent, tok + ": {\n")
indent += 1
else:
jbuf += _istr(indent, tok + ": " + ntok)
ntok = lex.get_token()
lex.push_token(ntok)
if ntok != "}":
jbuf += ","
jbuf += "\n"
if ntok is not None:
jbuf += _istr(indent, tok + ": " + ntok)
ntok = lex.get_token()
if ntok is not None:
lex.push_token(ntok)
if ntok != "}":
jbuf += ","
jbuf += "\n"
def gen_shortcut() -> None:

View File

@ -50,35 +50,41 @@ def rc2json(file: Path) -> str:
while True:
tok = lex.get_token()
ntok: str | bool | None = lex.get_token()
ntok = lex.get_token()
value: str | bool
if ntok is not None:
ntok = ntok.strip('""')
value = ntok
if tok in deprecated:
continue
elif tok in toggles:
if ntok is not None:
ntok = str2bool(ntok)
value = str2bool(ntok)
elif tok == "preferred_client":
tok = "client"
elif tok == "api_key":
tok = "bm_api"
elif tok == "ip_list":
while True:
ntok = lex.get_token().strip('""')
ntok = lex.get_token()
if ntok is not None:
ntok = ntok.strip('""')
value = ntok
if ntok == ")":
break
# TODO: make test for this
# NOTE: strip malformed records from ancient config file versions
if len(ntok.split(":")) == 3 and ntok.split(":")[2] != "":
ips.append(ntok)
if ntok is not None:
if len(ntok.split(":")) == 3 and ntok.split(":")[2] != "":
ips.append(ntok)
continue
if not tok:
break
keys[tok] = ntok
keys[tok] = value
keys["ip_list"] = ips
keys["use_miles"] = False

View File

@ -25,7 +25,6 @@ from dzgui.init.migrate import (
)
from dzgui.init.prefix import get_version
from dzgui.init.prereqs import has_steam_client
from dzgui.init.proc import has_cmd
from dzgui.init.update import allow_updates, check_updates
from dzgui.strings import boot
@ -37,7 +36,7 @@ from dzgui.util.symlink import rebuild_symlinks
from dzgui.util.strings import init, flags
from dzgui.views.base import App
from dzgui.views.dialogs.early_alert import EarlyAlertDialog, EarlyIgnoreDialog
from dzgui.views.dialogs.early_alert import EarlyAlertDialog
if TYPE_CHECKING:
from pathlib import Path

View File

@ -111,7 +111,12 @@ class ConnectionManager:
self.thread_man.set_cleanup_func(failure_func, destroy_first=True)
return
self.history = res.as_row()
try:
self.history = res.as_row()
except Exception:
self.thread_man.set_cleanup_func(failure_func, destroy_first=True)
return
record = res.get_record()
# NOTE: store metadata for later connection

View File

@ -108,7 +108,6 @@ class FastInsertListStore(ListStore):
self.append(row)
def append(self, row: list[Any] | tuple[Any, ...] | None = ...) -> TreeIter: # type: ignore
# FIXME: argument cannot be none
"""
Optimized for speed, but makes no assurances about row homogeneity
and may segfault if types and length are not identical to ListStore.

View File

@ -32,7 +32,7 @@ class ProxyModelManager:
{}
)
self.proxy_model: "FastInsertListStore" = None
self.proxy_model: "FastInsertListStore"
self.filter_man = filter_man
# TODO: list typehints
@ -49,6 +49,8 @@ class ProxyModelManager:
self.proxy_model.append(row)
def append_row_to_control(self, row: list) -> None:
if self.control_model is None:
raise AttributeError("Trying to add rows to a non-existent model")
self.control_model.append(row)
self.filter(FilterMode.INITIAL, skip_cache=True)
@ -97,8 +99,10 @@ class ProxyModelManager:
def clear_proxy_model(self) -> None:
self.proxy_model.clear()
def get_proxy_model(self) -> "FastInsertListStore":
return self.proxy_model
def get_proxy_model(self) -> Union["FastInsertListStore", None]:
if hasattr(self, "proxy_model"):
return self.proxy_model
return None
def filter(
self, mode: FilterMode, skip_cache: bool = False
@ -243,7 +247,7 @@ class ProxyModelManager:
return self.filtered
def set_cache(
self, filters: tuple, model: Optional["FastInsertListStore"], rows: list
self, filters: tuple, model: "FastInsertListStore", rows: list
) -> None:
self.filter_cache[filters] = (model, rows)
@ -260,7 +264,7 @@ class ProxyModelManager:
def get_filtered(self) -> list:
return self.filtered
def set_proxy_model(self, model: Optional["FastInsertListStore"]) -> None:
def set_proxy_model(self, model: "FastInsertListStore") -> None:
"""
FastInsertListStore representation of the raw model after filtration
"""
@ -272,7 +276,11 @@ class ProxyModelManager:
"""
self.control_model = rows
def get_control(self) -> list:
def get_control(self) -> list[Any]:
if self.control_model is None:
raise AttributeError(
"Expected a populated control model, but it is Nonetype"
)
return self.control_model
def wipe_cache(self, full: bool = False) -> None:

View File

@ -303,7 +303,6 @@ class ServerModelManager:
self.thread_man.set_cleanup_func(func)
return
config_man.add_saved_server(fqip)
# 2026-05-04
# FIXME: this is valid if saved servers tab is already open,
# but not if app was just booted
if proxy_man.has_control_model() is False:

View File

@ -54,9 +54,9 @@ def get_coords(ips: "Path", ip: str) -> Coords:
prefix = f"^{split[0]}.{split[1]}."
if shutil.which("rg") is not None:
args = ["/usr/bin/rg", prefix, ips]
args = ["/usr/bin/rg", prefix, str(ips)]
else:
args = ["/usr/bin/grep", "-E", prefix, ips]
args = ["/usr/bin/grep", "-E", prefix, str(ips)]
proc = subprocess.run(args, capture_output=True, text=True)
if proc.returncode != 0:

View File

@ -1,4 +1,5 @@
from typing import Self, Union, TYPE_CHECKING
from warnings import deprecated
from dzgui.const.enum import NotebookPage, ServerTab
from dzgui.util.strings import esc_to_return, question_to_return
@ -131,10 +132,11 @@ class Statusbar(Gtk.Grid):
cid = self.statusbar.get_context_id(str(context))
self.statusbar.pop(cid)
def get_text(self) -> str:
area = self.statusbar.get_message_area()
label = area.get_children()[0]
return str(label.get_text())
@deprecated("currently unused")
# def get_text(self) -> str:
# area = self.statusbar.get_message_area()
# label = area.get_children()[0]
# return str(label.get_text())
def set_by_context(
self, context: Union[NotebookPage, "ServerTab"], string: str

View File

@ -30,11 +30,11 @@ class ContextMixin(TreeView):
case Gdk.EventType.BUTTON_PRESS:
if event.button != 3:
return False
self._process_button_event(event)
self._process_button_event(event) # type: ignore
case Gdk.EventType.KEY_PRESS:
if not is_ctrl_mask(event):
if not is_ctrl_mask(event): # type: ignore
return False
if event.keyval is not Gdk.KEY_l:
if event.keyval is not Gdk.KEY_l: # type: ignore
return False
case _:
return False

View File

@ -94,7 +94,7 @@ class Options(Gtk.Box):
)
self.player_box.set_halign(Gtk.Align.START)
# TODO: make submit field a standalone class
self.player_box.get_children()[0].set_width_chars(30)
self.player_box.get_children()[0].set_width_chars(30) # type: ignore
self.fullscreen_toggle = self.make_binary_radio(
strings.options.last_used,
@ -159,18 +159,9 @@ class Options(Gtk.Box):
[LeftLabel(strings.options.name), self.player_box],
]
self.mod_install_toggle = self.make_binary_radio(
strings.options.manual_dl, strings.options.auto_dl, Preferences.INSTALL
)
self.force_button = Gtk.Button(label=strings.options.update)
self.force_button.connect("clicked", self._on_force_update_clicked)
# NOTE: sensitivity state is updated after config file is loaded
self.force_button.set_sensitive(False)
eb = InfoEventBox(options.workshop_eventbox, controller)
workshop_button = SteamWorkshopButton() # label=strings.self_workshop)
workshop_button = SteamWorkshopButton()
workshop_button.connect(
"clicked", lambda _: self.controller.open_user_workshop(self.uid)
)
@ -346,10 +337,6 @@ class Options(Gtk.Box):
self.bm_entry.set_text(self.old_bm)
pass
def _on_force_update_clicked(self, button: Gtk.Button) -> None:
# TODO: unimplemented
print("UNIMPLEMENTED")
def _on_start_tab_changed(self, combo: Gtk.ComboBoxText) -> None:
_iter = combo.get_active_iter()
if _iter is None:
@ -373,18 +360,6 @@ class Options(Gtk.Box):
self.populate_settings()
button.handler_unblock_by_func(self._on_radio_toggled)
if context == Preferences.INSTALL:
if self.controller.is_auto_install():
self.force_button.set_sensitive(True)
WorkshopLinkDialog(
self.controller,
strings.options.manual_sub_msg,
strings.self_workshop,
self.uid,
)
else:
self.force_button.set_sensitive(False)
def _is_valid_text(self, text: str, context: Preferences) -> bool:
if text.isspace():
return False
@ -458,11 +433,10 @@ class Options(Gtk.Box):
config = query.get_config(prefs.paths.config)
# TODO: use newer config enums
name = config["name"]
default_steam_path = config["default_steam_path"]
steam = config["steam_api"]
bm = config["bm_api"]
install = config["auto_install"]
name = self.controller.query_config(Preferences.NAME)
default_steam_path = self.controller.query_config(Preferences.DEFAULT)
steam = self.controller.query_config(Preferences.STEAM)
bm = self.controller.query_config(Preferences.BM)
steam_path = Path(default_steam_path)
# NOTE: this is a best effort guess at the most recent user
@ -481,9 +455,7 @@ class Options(Gtk.Box):
# NOTE: suppress toggle signal until radios are built
self._suppress_toggles(True)
self.force_button.set_sensitive(install)
for el, conf_state in [
(self.mod_install_toggle, install),
(self.fullscreen_toggle, config["fullscreen"]),
(self.distance_toggle, config["use_miles"]),
]:
@ -519,7 +491,6 @@ class Options(Gtk.Box):
def _suppress_toggles(self, state: bool) -> None:
for toggle in [
self.mod_install_toggle,
self.fullscreen_toggle,
self.distance_toggle,
]:

View File

@ -291,8 +291,6 @@ class PreConnectionAssistant(Gtk.ScrolledWindow):
name = prereqs.name
self.title.set_text(name)
# FIXME: append to tree status bar
if total_mods < 1:
self._hide_mod_area()
else:

View File

@ -64,7 +64,6 @@ class ServerModTreeView(ContextMixin, TreeView): # type: ignore
if model is None:
raise AttributeError("Trying to call a method on a non-existent model")
tree_iter = model.get_iter(path)
# FIXME: https://docs.gtk.org/gtk3/method.TreeModel.get.html
mod = model.get_value(tree_iter, 1)
return str(mod)