site/WESP_REL/tests/test_scales_reader.py

import os
import tempfile
import time
import unittest
from unittest.mock import Mock, patch

from app import create_app, db
from app.services.hardware import ScalesReader
from app.services.hardware.hx711_wrapper import HX711UnavailableError
from config import TestingConfig


class ScalesReaderConfig(TestingConfig):
    _TMP_DIR = tempfile.mkdtemp(prefix="wesp-scales-reader-tests-")
    SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'recipes_test.db')}"
    SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports_test.db')}"}
    SIMULATION_MODE = True
    READ_INTERVAL = 0.01


class ScalesReaderTests(unittest.TestCase):
    def setUp(self) -> None:
        self.app = create_app(ScalesReaderConfig)
        self.ctx = self.app.app_context()
        self.ctx.push()
        db.create_all()

    def tearDown(self) -> None:
        db.session.remove()
        db.drop_all()
        self.ctx.pop()

    def test_reader_start_stop_and_weight_pipeline(self) -> None:
        reader = ScalesReader(
            simulation_mode=True,
            read_interval=0.01,
            samples_per_read=2,
            sample_history_size=50,
            median_window=4,
            avg_window=3,
            weights=(0.2, 0.3, 0.5),
            smoothing_factor=0.3,
            round_to_step=1.0,
            tare_raw_window=2,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        reader.start()
        time.sleep(0.08)
        w1 = reader.get_current_weight()
        self.assertIsInstance(w1, int)
        self.assertGreaterEqual(w1, 0)

        tare_val = reader.tare()
        self.assertIsInstance(tare_val, float)
        time.sleep(0.05)
        w2 = reader.get_current_weight()
        self.assertGreaterEqual(w2, 0)

        applied = reader.calibrate(1.25)
        self.assertEqual(applied, 1.25)
        self.assertEqual(reader.get_counts_per_kg(), 1.25)

        with self.assertRaises(ValueError):
            reader.calibrate(0)

        reader.stop()

    def test_negative_weight_not_clamped_to_zero(self) -> None:
        reader = ScalesReader(
            simulation_mode=False,
            read_interval=0.01,
            samples_per_read=1,
            sample_history_size=20,
            median_window=1,
            avg_window=1,
            weights=(1.0,),
            smoothing_factor=1.0,
            round_to_step=1.0,
            tare_raw_window=1,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        reader._counts_per_kg = 1.0
        reader._tare_raw = 0.0
        reader._scale_hardware_platform = True
        mock_hx = Mock()
        mock_hx._device = object()
        mock_hx.read_weight_sample.return_value = -100.0
        reader._hx711 = mock_hx
        reader.start()
        time.sleep(0.12)
        self.assertLess(reader.get_current_weight(), 0)
        reader.stop()

    def test_real_hardware_mode_inits_hx711_driver(self) -> None:
        reader = ScalesReader(
            simulation_mode=False,
            read_interval=0.01,
            samples_per_read=1,
            sample_history_size=20,
            median_window=1,
            avg_window=1,
            weights=(1.0,),
            smoothing_factor=1.0,
            round_to_step=1.0,
            tare_raw_window=1,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        reader._scale_hardware_platform = True
        device = Mock()
        device.get_raw_data_mean.return_value = 50000.0
        hx_mod = Mock()
        hx_mod.HX711.return_value = device
        with patch.dict("sys.modules", {"hx711": hx_mod}):
            reader._sync_hx711_wrapper_mode()
        self.assertFalse(reader._hx711.simulation_mode)
        self.assertIs(reader._hx711._device, device)
        reader._counts_per_kg = 1000.0
        reader._tare_raw = 0.0
        reader.start()
        time.sleep(0.12)
        self.assertGreater(reader.get_current_weight(), 0)
        reader.stop()

    def test_non_simulation_does_not_invent_weight_when_hx711_missing(self) -> None:
        reader = ScalesReader(
            simulation_mode=False,
            read_interval=0.01,
            samples_per_read=1,
            sample_history_size=20,
            median_window=3,
            avg_window=2,
            weights=(0.2, 0.3, 0.5),
            smoothing_factor=0.3,
            round_to_step=1.0,
            tare_raw_window=2,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        reader._scale_hardware_platform = True
        mock_hx = Mock()
        mock_hx._device = None
        mock_hx.simulation_mode = False
        mock_hx.read_weight_sample.side_effect = HX711UnavailableError("нет датчика")
        reader._hx711 = mock_hx
        reader.start()
        time.sleep(0.08)
        self.assertEqual(reader.get_current_weight(), 0)
        health = reader.get_scale_health()
        self.assertFalse(health["hx711_ok"])
        self.assertIsNotNone(health["hx711_error"])
        reader.stop()

    def test_non_arm_skips_hx711_poll_and_logs(self) -> None:
        from unittest.mock import patch

        reader = ScalesReader(
            simulation_mode=False,
            read_interval=0.01,
            samples_per_read=1,
            sample_history_size=20,
            median_window=1,
            avg_window=1,
            weights=(1.0,),
            smoothing_factor=1.0,
            round_to_step=1.0,
            tare_raw_window=1,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        reader._scale_hardware_platform = False
        mock_hx = Mock()
        mock_hx.read_weight_sample.side_effect = HX711UnavailableError("не должен вызываться")
        reader._hx711 = mock_hx
        reader.start()
        time.sleep(0.08)
        mock_hx.read_weight_sample.assert_not_called()
        health = reader.get_scale_health()
        self.assertTrue(health.get("hx711_skipped"))
        self.assertIsNone(health.get("hx711_error"))
        reader.stop()

    def test_simulation_toggle_and_weight_adjust(self) -> None:
        reader = ScalesReader(
            simulation_mode=False,
            read_interval=0.01,
            samples_per_read=1,
            sample_history_size=20,
            median_window=1,
            avg_window=1,
            weights=(1.0,),
            smoothing_factor=1.0,
            round_to_step=1.0,
            tare_raw_window=1,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        reader.start()
        state = reader.set_simulation_mode(True)
        self.assertTrue(state["simulation_mode"])
        w = reader.adjust_simulation_weight(5)
        self.assertEqual(w, 5)
        self.assertEqual(reader.get_current_weight(), 5)
        w2 = reader.adjust_simulation_weight(-2)
        self.assertEqual(w2, 3)
        reader.set_simulation_mode(False)
        off = reader.get_simulation_state()
        self.assertFalse(off["simulation_mode"])
        with self.assertRaises(ValueError):
            reader.adjust_simulation_weight(1)
        with self.assertRaises(ValueError):
            reader.set_simulation_weight(10)
        reader.stop()

    def test_set_simulation_weight_absolute(self) -> None:
        reader = ScalesReader(
            simulation_mode=True,
            read_interval=0.01,
            samples_per_read=1,
            sample_history_size=20,
            median_window=1,
            avg_window=1,
            weights=(1.0,),
            smoothing_factor=1.0,
            round_to_step=1.0,
            tare_raw_window=1,
            hx711_dout_pin=2,
            hx711_pd_sck_pin=3,
        )
        w = reader.set_simulation_weight(42)
        self.assertEqual(w, 42)
        reader.start()
        time.sleep(0.05)
        self.assertEqual(reader.get_current_weight(), 42)
        reader.stop()


if __name__ == "__main__":
    unittest.main()