mirror of
https://github.com/NickoScope/nickol-knx-mcp.git
synced 2026-09-29 19:31:12 +02:00
ci: ruff + mypy gates, Python 3.13/3.14 in the matrix, Dependabot
The CI ran tests only. Added a lint job with two narrow gates: ruff with select = E9,F — syntax errors and pyflakes (undefined names, unused imports, broken f-strings). Style rules are deliberately off: the full default set flagged 202 things, almost all cosmetic, and a gate nobody reads is not a gate. Fixed the 9 findings it did have (8 unused imports, 1 empty f-string). mypy, clean at 13 errors fixed. None of the 13 was a live bug, so every fix is an annotation or a local rename: the two accumulators in advanced, the block map in appprog_parser, a reused loop name in project and in repair, a TypedDict lost through dict(), a mixed-type summary dict in handover. Two are worth naming. _CH_TOKEN_RE now captures the letter prefix directly instead of re-matching its own capture, which removes an unguarded .group() on a possibly-None match; the regex accepts exactly the same strings. repair's status-DPT lookup now spells out the None case, which is what dict.get already did since None is not a key there. Verified the fixes moved nothing: the full suite passes, and the corpus guard reports no drift across all six real projects, the 3646-GA one included. Matrix extended to 3.13 and 3.14; both were smoke-tested locally first. Dependabot config for pip and github-actions, weekly and grouped so a quiet week is one PR. Dependabot security updates and CodeQL default setup enabled in the repository settings.
This commit is contained in:
@@ -16,7 +16,7 @@ from collections import Counter
|
||||
from typing import Any
|
||||
|
||||
from .project import LoadedProject
|
||||
from .pairing import find_status, base_tokens
|
||||
from .pairing import find_status
|
||||
from .intent import INTENT_FUNCTIONAL
|
||||
|
||||
|
||||
@@ -71,7 +71,8 @@ _MATTER = {
|
||||
def matter_readiness(project: LoadedProject) -> dict[str, Any]:
|
||||
"""Which controllable functions round-trip to a Matter cluster, and what's missing."""
|
||||
stats = _status_gas(project)
|
||||
ready, not_ready = [], []
|
||||
ready: list[dict[str, Any]] = []
|
||||
not_ready: list[dict[str, Any]] = []
|
||||
no_cluster: Counter = Counter()
|
||||
for ga in _functional_commands(project):
|
||||
cluster = _MATTER.get(ga.category)
|
||||
|
||||
@@ -31,7 +31,7 @@ _HW_SPLIT_RE = re.compile(r"<Hardware\b")
|
||||
_APPREF_RE = re.compile(r'<ApplicationProgramRef\s+RefId="([^"]+)"')
|
||||
_APPVER_RE = re.compile(r'ApplicationVersion="(\d+)"')
|
||||
_SIZE_RE = re.compile(r"(\d+)\s*(Bit|Byte)", re.I)
|
||||
_CH_TOKEN_RE = re.compile(r"\[([A-Za-z]{1,3}\d+)\]") # display token: [C1] [O12] [T3]
|
||||
_CH_TOKEN_RE = re.compile(r"\[([A-Za-z]{1,3})\d+\]") # display token: [C1] [O12] [T3]
|
||||
_LF_RE = re.compile(r"^\s*\[LF\]")
|
||||
|
||||
# KNX manufacturer id (hex in the M-code) -> vendor name (common ones; extend freely)
|
||||
@@ -87,7 +87,7 @@ def _channel_token(text: str, name: str) -> Optional[str]:
|
||||
"""Return the repeating-block key for an object, e.g. 'C', 'O', 'T', or None (general)."""
|
||||
m = _CH_TOKEN_RE.search(text or "")
|
||||
if m:
|
||||
return re.match(r"([A-Za-z]{1,3})", m.group(1)).group(1)
|
||||
return m.group(1)
|
||||
m = re.search(r"\bch\[(\d+)\]", name or "") # internal 'oX.ch[0].y' form
|
||||
return "ch" if m else None
|
||||
|
||||
@@ -169,7 +169,9 @@ def _parse_comobjects(xml_text: str) -> list[dict[str, Any]]:
|
||||
|
||||
def _detect_blocks(objs: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
"""Best-effort per-channel block/stride + general/[LF] split from object names."""
|
||||
general, lf, chan = [], [], {}
|
||||
general: list = []
|
||||
lf: list = []
|
||||
chan: dict[str, Any] = {}
|
||||
for o in objs:
|
||||
text = o["name"] or ""
|
||||
if _LF_RE.match(text):
|
||||
@@ -301,9 +303,9 @@ def parse_project(path: str, password: Optional[str] = None) -> dict[str, Any]:
|
||||
continue
|
||||
if app_path not in app_cache:
|
||||
xml_text = safe_read(z, app_path, pwd).decode("utf-8", "replace")
|
||||
ver = _APPVER_RE.search(xml_text[:8000])
|
||||
verm = _APPVER_RE.search(xml_text[:8000])
|
||||
app_cache[app_path] = (_parse_comobjects(xml_text),
|
||||
ver.group(1) if ver else None)
|
||||
verm.group(1) if verm else None)
|
||||
objs, ver = app_cache[app_path]
|
||||
bd = _detect_blocks(objs)
|
||||
no_dpt = sum(1 for o in objs if o["dpt"] is None)
|
||||
|
||||
@@ -134,8 +134,8 @@ def classify_dpt(main: Optional[int], sub: Optional[int]) -> DptInfo:
|
||||
return {"category": CATEGORY_UNKNOWN, "kind": KIND_UNKNOWN,
|
||||
"ha_platform": "unknown", "value_type": None, "label": "No DPT assigned"}
|
||||
if (main, sub) in _EXACT:
|
||||
return dict(_EXACT[(main, sub)]) # copy
|
||||
return _EXACT[(main, sub)].copy() # copy
|
||||
if main in _MAIN_FALLBACK:
|
||||
return dict(_MAIN_FALLBACK[main])
|
||||
return _MAIN_FALLBACK[main].copy()
|
||||
return {"category": CATEGORY_UNKNOWN, "kind": KIND_UNKNOWN,
|
||||
"ha_platform": "unknown", "value_type": None, "label": f"DPT {main}"}
|
||||
|
||||
@@ -17,7 +17,7 @@ misclassification hides.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Optional
|
||||
from typing import Any
|
||||
|
||||
from .project import (LoadedProject, _override_kind_by_name, _domain_from_text,
|
||||
_SOFT_DPT)
|
||||
|
||||
@@ -12,7 +12,7 @@ from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import io
|
||||
from typing import Any, Optional
|
||||
from typing import Any
|
||||
from xml.sax.saxutils import escape
|
||||
|
||||
from .project import LoadedProject
|
||||
|
||||
@@ -88,12 +88,13 @@ def _feedback_coverage(project: LoadedProject,
|
||||
total = len(commands)
|
||||
with_status = max(total - gaps, 0)
|
||||
pct = round(100 * with_status / total) if total else 0
|
||||
return {
|
||||
summary: dict[str, Any] = {
|
||||
"commands": total, "with_status": with_status, "missing": gaps,
|
||||
"pct": pct,
|
||||
"formula": f"{with_status} / {total} functional command GAs have a status = {pct}%"
|
||||
if total else "no functional command GAs — coverage undefined",
|
||||
}
|
||||
return summary
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
|
||||
@@ -34,7 +34,7 @@ def generate_knx_iot_turtle(project: LoadedProject) -> str:
|
||||
"# Pragmatic skeleton (datapoint per functional group address), NOT the full",
|
||||
"# certified KNX IoT ontology. Review before use.",
|
||||
"",
|
||||
f'proj:project a knx:Installation ;',
|
||||
'proj:project a knx:Installation ;',
|
||||
f' rdfs:label "{_esc(project.info.get("name","KNX project"))}" ;',
|
||||
f' knx:groupAddressStyle "{_esc(project.info.get("group_address_style","ThreeLevel"))}" .',
|
||||
"",
|
||||
|
||||
@@ -434,8 +434,8 @@ def build_loaded_from_raw(raw: KNXProject, path: str) -> LoadedProject:
|
||||
# 15/29 real position feedbacks (issue #12 re-test by Kris1166).
|
||||
if not (has_shutter_role or has_move):
|
||||
continue
|
||||
for rec in recs:
|
||||
_promote_to_shutter(rec, has_move)
|
||||
for member in recs:
|
||||
_promote_to_shutter(member, has_move)
|
||||
|
||||
return LoadedProject(
|
||||
path=path,
|
||||
|
||||
@@ -14,7 +14,6 @@ from typing import Any, Optional
|
||||
|
||||
from .project import LoadedProject
|
||||
from .analyze import detect_missing_status, detect_dpt_issues, _expected_subdpt
|
||||
from .intent import INTENT_FUNCTIONAL
|
||||
|
||||
|
||||
# command DPT main -> the status/feedback DPT to synthesise for it
|
||||
@@ -125,11 +124,11 @@ def suggest_repairs(project: LoadedProject) -> dict[str, Any]:
|
||||
if f["code"] != "missing_status_address":
|
||||
continue
|
||||
addr = f["address"]
|
||||
ga = project.gas.get(addr)
|
||||
if ga is None:
|
||||
sga = project.gas.get(addr)
|
||||
if sga is None:
|
||||
continue
|
||||
sdpt = _STATUS_DPT.get(ga.dpt_main, "1.011")
|
||||
new = _next_free(used, ga.main if ga.main is not None else 1, prefer_middle=4)
|
||||
sdpt = _STATUS_DPT.get(sga.dpt_main, "1.011") if sga.dpt_main is not None else "1.011"
|
||||
new = _next_free(used, sga.main if sga.main is not None else 1, prefer_middle=4)
|
||||
proposals.append({
|
||||
"code": "missing_status", "action": "add_ga", "address": new, "for": addr,
|
||||
"name": f"{ga.name}{_suffix(ga.name, ' (статус)', ' (status)')}", "dpt": sdpt,
|
||||
|
||||
@@ -426,7 +426,7 @@ def validate_template(tmpl: dict[str, Any]) -> dict[str, Any]:
|
||||
findings.append(_finding("error", "slot_id_duplicate",
|
||||
f"Slot id '{sid}' is used more than once.", slot=sid))
|
||||
else:
|
||||
seen_slots.add(sid)
|
||||
seen_slots.add(str(sid))
|
||||
slabels = slot.get("labels") or {}
|
||||
for lang in SUPPORTED_LANGUAGES:
|
||||
if not (slabels.get(lang) or "").strip():
|
||||
|
||||
@@ -240,7 +240,7 @@ def _classify_channel(project: dict[str, Any], links: list[_Link], chan_name: st
|
||||
return None
|
||||
evidence.append("object_text")
|
||||
cur = t_cur_src[0]
|
||||
knx: dict[str, Any] = {"ga_temperature_current": _ga_conf(cur, "state")}; used.add(cur.ga)
|
||||
knx = {"ga_temperature_current": _ga_conf(cur, "state")}; used.add(cur.ga)
|
||||
if cur.role == "sink":
|
||||
review.append("measured temperature taken from the thermostat's external-sensor INPUT object")
|
||||
if sp_cmd:
|
||||
|
||||
Reference in New Issue
Block a user