diff --git a/pyasic/miners/backends/luxminer.py b/pyasic/miners/backends/luxminer.py index 48dbc735..7671b01a 100644 --- a/pyasic/miners/backends/luxminer.py +++ b/pyasic/miners/backends/luxminer.py @@ -177,6 +177,90 @@ async def atm_enabled(self) -> bool | None: pass return None + async def _switch_profile(self, profile_name: str) -> dict: + """Switch LuxOS profile with ATM temporarily disabled if needed. + + Handles the ATM disable/re-enable dance required by LuxOS when + switching profiles. ATM is always re-enabled if it was on, even + if the profile switch fails. + + Parameters: + profile_name: The name of the profile to switch to. + + Returns: + The raw response dict from profileset. + + Raises: + APIError: If the RPC call returns an API-level error. + Exception: If the profile switch fails for other reasons. + """ + re_enable_atm = False + try: + if await self.atm_enabled(): + re_enable_atm = True + await self.rpc.atmset(enabled=False) + return await self.rpc.profileset(profile_name) + finally: + if re_enable_atm: + try: + await self.rpc.atmset(enabled=True) + except Exception as e: + logging.warning( + f"{self} - Failed to re-enable ATM after profile switch: {e}" + ) + + @staticmethod + def _match_profile_name(name: str, profile_names: list[str]) -> str | None: + """Match user input to an available profile name. + + Tries exact match first, then case-insensitive, then checks if + appending "MHz" yields a match (so "190" matches "190MHz"). + """ + if name in profile_names: + return name + + lower = name.lower() + for p in profile_names: + if p.lower() == lower: + return p + + for p in profile_names: + if p.lower() == lower + "mhz": + return p + + return None + + async def set_profile(self, name: str) -> bool: + config = await self.get_config() + + # Validate profile name exists + if not hasattr(config.mining_mode, "available_presets"): + logging.warning(f"{self} - Mining mode does not support profiles") + return False + + available_presets = getattr(config.mining_mode, "available_presets", []) + profile_names = [p.name for p in available_presets if p.name is not None] + + matched_name = self._match_profile_name(name, profile_names) + if matched_name is None: + logging.warning( + f"{self} - Profile '{name}' not found in available profiles: {profile_names}" + ) + return False + + try: + result = await self._switch_profile(matched_name) + except APIError: + raise + except Exception as e: + logging.warning(f"{self} - Failed to set profile: {e}") + return False + + if result["PROFILE"][0]["Profile"] == matched_name: + return True + else: + return False + async def set_power_limit(self, wattage: int) -> bool: config = await self.get_config() @@ -201,17 +285,10 @@ async def set_power_limit(self, wattage: int) -> bool: return False # Set power to highest preset <= wattage - # If ATM enabled, must disable it before setting power limit new_preset = max(valid_presets, key=lambda x: valid_presets[x]) - re_enable_atm = False try: - if await self.atm_enabled(): - re_enable_atm = True - await self.rpc.atmset(enabled=False) - result = await self.rpc.profileset(new_preset) - if re_enable_atm: - await self.rpc.atmset(enabled=True) + result = await self._switch_profile(new_preset) except APIError: raise except Exception as e: diff --git a/pyasic/miners/base.py b/pyasic/miners/base.py index 1b374f73..c67a362c 100644 --- a/pyasic/miners/base.py +++ b/pyasic/miners/base.py @@ -174,6 +174,17 @@ async def resume_mining(self) -> bool: """ return False + async def set_profile(self, name: str) -> bool: + """Set the mining profile by name. + + Parameters: + name: The name of the profile to switch to. + + Returns: + A boolean value of the success of setting the profile. + """ + return False + async def set_power_limit(self, wattage: int) -> bool: """Set the power limit to be used by the miner. diff --git a/tests/miners_tests/backends_tests/luxminer_tests/__init__.py b/tests/miners_tests/backends_tests/luxminer_tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/miners_tests/backends_tests/luxminer_tests/test_set_preset.py b/tests/miners_tests/backends_tests/luxminer_tests/test_set_preset.py new file mode 100644 index 00000000..c1a841b0 --- /dev/null +++ b/tests/miners_tests/backends_tests/luxminer_tests/test_set_preset.py @@ -0,0 +1,136 @@ +"""Tests for LuxOS set_profile() method.""" + +import logging +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +from pyasic.config.mining import MiningModeConfig +from pyasic.config.mining.presets import MiningPreset + + +def _make_preset(name, power=1000, frequency=400, voltage=8.9): + """Helper to create a MiningPreset.""" + return MiningPreset( + name=name, + power=power, + hashrate=80.0, + tuned=True, + modded_psu=False, + frequency=frequency, + voltage=voltage, + ) + + +def _make_config( + active_preset_name="415MHz", available_preset_names=("190MHz", "415MHz", "565MHz") +): + """Helper to create a mock config with presets.""" + presets = [_make_preset(n) for n in available_preset_names] + active = next((p for p in presets if p.name == active_preset_name), presets[0]) + config = MagicMock() + config.mining_mode.available_presets = presets + config.mining_mode.active_preset = active + return config + + +class TestSetProfile(unittest.IsolatedAsyncioTestCase): + """Test LuxOS set_profile() behavior.""" + + def _make_miner(self, atm_enabled=True, config=None, profileset_result=None): + """Create a mock LuxOS miner with controllable behavior.""" + from pyasic.miners.backends.luxminer import LUXMiner + + miner = LUXMiner.__new__(LUXMiner) + miner.rpc = MagicMock() + miner.rpc.atmset = AsyncMock() + miner.rpc.profileset = AsyncMock( + return_value=profileset_result or {"PROFILE": [{"Profile": "415MHz"}]} + ) + miner.atm_enabled = AsyncMock(return_value=atm_enabled) + miner.get_config = AsyncMock(return_value=config or _make_config()) + miner.ip = "192.168.1.237" + return miner + + async def test_switches_preset_with_atm_on(self): + """When ATM is on, should disable ATM, switch, then re-enable ATM.""" + miner = self._make_miner( + atm_enabled=True, + profileset_result={"PROFILE": [{"Profile": "190MHz"}]}, + ) + + result = await miner.set_profile("190MHz") + + self.assertTrue(result) + miner.rpc.atmset.assert_any_call(enabled=False) + miner.rpc.profileset.assert_called_once_with("190MHz") + miner.rpc.atmset.assert_any_call(enabled=True) + + async def test_switches_preset_with_atm_off(self): + """When ATM is off, should switch without toggling ATM.""" + miner = self._make_miner( + atm_enabled=False, + profileset_result={"PROFILE": [{"Profile": "190MHz"}]}, + ) + + result = await miner.set_profile("190MHz") + + self.assertTrue(result) + miner.rpc.atmset.assert_not_called() + miner.rpc.profileset.assert_called_once_with("190MHz") + + async def test_invalid_preset_name_returns_false(self): + """Should return False when preset name doesn't exist.""" + miner = self._make_miner() + + result = await miner.set_profile("nonexistent_profile") + + self.assertFalse(result) + miner.rpc.profileset.assert_not_called() + + async def test_atm_re_enable_failure_returns_true_and_warns(self): + """If preset switches but ATM re-enable fails, return True and log warning.""" + miner = self._make_miner( + atm_enabled=True, + profileset_result={"PROFILE": [{"Profile": "190MHz"}]}, + ) + # First call (disable) succeeds, second call (re-enable) fails + miner.rpc.atmset.side_effect = [None, Exception("ATM re-enable failed")] + + with self.assertLogs(level=logging.WARNING): + result = await miner.set_profile("190MHz") + + self.assertTrue(result) + miner.rpc.profileset.assert_called_once_with("190MHz") + + async def test_fuzzy_match_case_insensitive(self): + """Should match preset name case-insensitively.""" + miner = self._make_miner( + atm_enabled=False, + profileset_result={"PROFILE": [{"Profile": "190MHz"}]}, + ) + + result = await miner.set_profile("190mhz") + + self.assertTrue(result) + miner.rpc.profileset.assert_called_once_with("190MHz") + + async def test_fuzzy_match_number_only(self): + """Should match '190' to '190MHz'.""" + miner = self._make_miner( + atm_enabled=False, + profileset_result={"PROFILE": [{"Profile": "190MHz"}]}, + ) + + result = await miner.set_profile("190") + + self.assertTrue(result) + miner.rpc.profileset.assert_called_once_with("190MHz") + + async def test_profileset_failure_returns_false(self): + """If RPC profileset fails, should return False.""" + miner = self._make_miner(atm_enabled=False) + miner.rpc.profileset.side_effect = Exception("RPC error") + + result = await miner.set_profile("190MHz") + + self.assertFalse(result)