Fix DG-Home1 Elphapex detection and harden hashboard parsing

pyasic/miners/backends/elphapex.py

@@ -217,48 +217,133 @@ async def _get_errors(  # type: ignore[override]
         return errors
 
     async def _get_hashboards(self, web_stats: dict | None = None) -> list[HashBoard]:
-        if self.expected_hashboards is None:
+        """Build the list of ``HashBoard`` records from ``stats.cgi``."""
+        # Tolerates partially-supported devices (model resolved but
+        # ``expected_hashboards`` is ``None``), missing/malformed payloads,
+        # and sparsely-populated chains (e.g. DG-Home1 ships with 1 of 4
+        # chains populated). See ``_resolve_expected_hashboards`` for the
+        # slot-count fallback rationale.
+        if web_stats is None:
+            try:
+                web_stats = await self.web.stats()
+            except APIError:
+                web_stats = None
+
+        expected = self._resolve_expected_hashboards(web_stats)
+        if expected is None:
             return []
 
         hashboards = [
             HashBoard(slot=idx, expected_chips=self.expected_chips)
-            for idx in range(self.expected_hashboards)
+            for idx in range(expected)
         ]
+        for board in self._iter_stats_chains(web_stats):
+            self._apply_chain_to_board(hashboards, board)
+        return hashboards
 
-        if web_stats is None:
-            try:
-                web_stats = await self.web.stats()
-            except APIError:
-                return hashboards
+    def _resolve_expected_hashboards(self, web_stats: dict | None) -> int | None:
+        """Return the number of hashboard slots to model."""
+        # Falls back to ``STATS[0].chain_num`` (or ``len(STATS[0].chain)``)
+        # when the device is only partially supported and
+        # ``self.expected_hashboards`` is ``None``. Without this,
+        # ``range(self.expected_hashboards)`` raises ``TypeError`` and bubbles
+        # up as ``APIError`` for DG-Home1 on pyasic 0.79.0
+        # (see UpstreamData/pyasic#311, #428).
+        if self.expected_hashboards is not None:
+            return self.expected_hashboards
+        if not isinstance(web_stats, dict):
+            return None
+        try:
+            stats0 = web_stats["STATS"][0]
+        except (LookupError, TypeError):
+            return None
+        chain_num = stats0.get("chain_num")
+        if chain_num is None and isinstance(stats0.get("chain"), list):
+            chain_num = len(stats0["chain"])
+        if isinstance(chain_num, int) and chain_num > 0:
+            return chain_num
+        return None
 
-        if web_stats is not None:
+    @staticmethod
+    def _iter_stats_chains(web_stats: dict | None) -> list[dict]:
+        """Return the per-chain stats list, or an empty list if missing."""
+        if not isinstance(web_stats, dict):
+            return []
+        try:
+            chains = web_stats["STATS"][0]["chain"]
+        except (LookupError, TypeError):
+            return []
+        return chains if isinstance(chains, list) else []
+
+    def _apply_chain_to_board(self, hashboards: list[HashBoard], board: dict) -> None:
+        """Populate a single ``HashBoard`` from a ``stats.cgi`` chain entry."""
+        board_index = board.get("index") if isinstance(board, dict) else None
+        if not isinstance(board_index, int) or board_index >= len(hashboards):
+            # Unknown / out-of-range chain index; skip rather than crash.
+            return
+
+        hb = hashboards[board_index]
+        self._set_board_hashrate(hb, board.get("rate_real", 0))
+
+        asic_num = board.get("asic_num")
+        if isinstance(asic_num, int):
+            hb.chips = asic_num
+
+        temp = self._average_pcb_temp(board.get("temp_pcb"))
+        if temp is not None:
+            hb.temp = temp
+
+        chip_temp = self._average_chip_temp(board.get("temp_chip"))
+        if chip_temp is not None:
+            hb.chip_temp = chip_temp
+
+        hb.serial_number = board.get("sn") or None
+        # Treat a chain with no live ASICs as missing so downstream consumers
+        # can distinguish a populated-but-broken board from an empty slot
+        # (DG-Home1 ships with 1 of 4 chains populated).
+        hb.missing = not (isinstance(asic_num, int) and asic_num > 0)
+
+    def _set_board_hashrate(self, hb: HashBoard, rate_real: float | int) -> None:
+        """Convert ``rate_real`` (MH/s on Elphapex) into the algo's default unit."""
+        try:
+            hb.hashrate = self.algo.hashrate(
+                rate=rate_real,
+                unit=self.algo.unit.MH,  # type: ignore[attr-defined]
+            ).into(
+                self.algo.unit.default  # type: ignore[attr-defined]
+            )
+        except (TypeError, ValueError):
+            pass
+
+    @staticmethod
+    def _average_pcb_temp(temps: object) -> float | None:
+        """Average non-zero PCB temperature readings, or ``None`` if empty."""
+        if not isinstance(temps, list):
+            return None
+        readings = [t for t in temps if isinstance(t, (int, float)) and t != 0]
+        if not readings:
+            return None
+        return sum(readings) / len(readings)
+
+    @staticmethod
+    def _average_chip_temp(temps: object) -> float | None:
+        """Average ``temp_chip`` entries (millidegree strings on Elphapex)."""
+        # Inactive chains report empty strings here; a single ``None`` chain
+        # on DG-Home1 used to trigger ``ZeroDivisionError`` in the legacy
+        # parser.
+        if not isinstance(temps, list):
+            return None
+        readings: list[float] = []
+        for raw in temps:
+            if raw in (None, ""):
+                continue
             try:
-                for board in web_stats["STATS"][0]["chain"]:
-                    hashboards[board["index"]].hashrate = self.algo.hashrate(
-                        rate=board["rate_real"],
-                        unit=self.algo.unit.MH,  # type: ignore[attr-defined]
-                    ).into(
-                        self.algo.unit.default  # type: ignore[attr-defined]
-                    )
-                    hashboards[board["index"]].chips = board["asic_num"]
-                    board_temp_data = list(
-                        filter(lambda x: not x == 0, board["temp_pcb"])
-                    )
-                    if not len(board_temp_data) == 0:
-                        hashboards[board["index"]].temp = sum(board_temp_data) / len(
-                            board_temp_data
-                        )
-                    chip_temp_data = list(
-                        filter(lambda x: not x == "", board["temp_chip"])
-                    )
-                    hashboards[board["index"]].chip_temp = sum(
-                        [int(i) / 1000 for i in chip_temp_data]
-                    ) / len(chip_temp_data)
-                    hashboards[board["index"]].serial_number = board["sn"]
-                    hashboards[board["index"]].missing = False
-            except LookupError:
-                pass
-        return hashboards
+                readings.append(int(raw) / 1000)
+            except (TypeError, ValueError):
+                continue
+        if not readings:
+            return None
+        return sum(readings) / len(readings)
 
     async def _get_fault_light(
         self, web_get_blink_status: dict | None = None

pyasic/miners/factory.py

@@ -720,7 +720,12 @@ class MinerTypes(enum.Enum):
         None: type("ElphapexUnknown", (ElphapexMiner, ElphapexMake), {}),
         "DG1+": ElphapexDG1Plus,
         "DG1": ElphapexDG1,
-        "DG1-Home": ElphapexDG1Home,
+        # NOTE: lookups apply ``str(miner_model).upper()`` so keys must be uppercase.
+        # The DG-Home1 firmware reports ``INFO.type = "DG-Home1"`` from
+        # ``/cgi-bin/stats.cgi``; ``get_miner_model_elphapex`` normalises that to
+        # ``"DG1-HOME"`` for consistency with ``DG1`` / ``DG1+`` naming.
+        "DG1-HOME": ElphapexDG1Home,
+        "DG-HOME1": ElphapexDG1Home,
     },
     MinerTypes.FLUMINER: {
         None: type("FluminerUnknown", (Fluminer, FluminerMake), {}),
@@ -1645,14 +1650,45 @@ async def get_miner_model_elphapex(self, ip: str) -> str | None:
             ip, "/cgi-bin/get_system_info.cgi", auth=auth
         )
 
+        # Try ``get_system_info.cgi`` first — older Elphapex firmware exposes
+        # ``minertype`` here. Newer DG-Home1 firmware (V1.0.5) does not, so we
+        # also accept ``type`` / ``model`` / ``hostname`` from the same response.
         if web_json_data is not None:
-            try:
-                miner_model = web_json_data["minertype"]
-                return miner_model
-            except (TypeError, LookupError):
-                pass
+            for key in ("minertype", "type", "model", "hostname"):
+                try:
+                    miner_model = web_json_data[key]
+                except (TypeError, LookupError):
+                    continue
+                if miner_model:
+                    return self._normalize_elphapex_model(miner_model)
+
+        # Fallback: ``stats.cgi`` includes ``INFO.type`` (e.g. ``"DG-Home1"``)
+        # on firmware that omits ``minertype`` from ``get_system_info.cgi``.
+        stats_data = await self.send_web_command(ip, "/cgi-bin/stats.cgi", auth=auth)
+        if stats_data is not None:
+            info = stats_data.get("INFO") if isinstance(stats_data, dict) else None
+            if isinstance(info, dict):
+                for key in ("type", "model", "miner_version"):
+                    miner_model = info.get(key)
+                    if miner_model:
+                        return self._normalize_elphapex_model(miner_model)
         return None
 
+    @staticmethod
+    def _normalize_elphapex_model(miner_model: str) -> str:
+        """Normalise Elphapex model strings to the lookup-table form."""
+        # DG-Home1 firmware V1.0.5 reports ``"DG-Home1"`` from ``stats.cgi``
+        # (and the firmware version string is ``"DG-Home1_V1.0.5"``), but the
+        # lookup table is keyed on the upstream-canonical ``"DG1-Home"``.
+        # Strip any firmware-version suffix and remap known aliases.
+        model = str(miner_model).strip()
+        # Drop firmware suffix like ``_V1.0.5`` if we got fed ``miner_version``.
+        model = model.split("_")[0]
+        # Known aliases for the DG-Home1 (also covers ``DG-HOME1``, ``DG_Home1``).
+        if model.upper().replace("_", "-") in ("DG-HOME1", "DG-HOME"):
+            return "DG1-Home"
+        return model
+
     async def get_miner_model_fluminer(self, ip: str) -> str | None:
         web_json_data = await self.send_web_command(ip, "/api/overview")