Leave out unusable rooms instead of pausing; add keep-last-value option

A single stale, unavailable, missing or invalid room sensor paused
learning and forecasts for every room. That is not acceptable once the
MPC may control heating.

Now such a room is left out: the weighted temperature, learning and the
optimizer use the remaining rooms with their weights re-normalized, and
a room learns only across poll pairs where it was usable at both ends.
Readiness considers only the rooms used. Everything pauses only when no
room is usable or a shared input (relay, thermostat, outdoor
temperature) is missing.

A new setup/Configure checkbox, room_stale_use_last_value (default off),
keeps using a stale room sensor's last value instead, for sensors that
report only on change. It never applies to unavailable or invalid
sensors, or to a sensor silent for more than 24 hours.

Status adds excluded_rooms, weight_coverage and a per-room "used" flag.
English and Romanian labels and help text added.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/ha/test_config_flow.py

@@ -15,6 +15,7 @@ async def test_config_flow(hass, inputs):
     validated = result["data_schema"](main)
     assert validated["room_max_age_minutes"] == 30
     assert validated["weather_max_age_minutes"] == 120
+    assert validated["room_stale_use_last_value"] is False
     result = await hass.config_entries.flow.async_configure(result["flow_id"], validated)
     assert result["step_id"] == "weather"
     result = await hass.config_entries.flow.async_configure(
@@ -35,7 +36,7 @@ async def test_options_flow_keeps_settings(hass, entry, inputs):
     result = await hass.config_entries.options.async_init(entry.entry_id)
     main = result["data_schema"]({"climate_entity": "climate.reference",
                                  "heater_entity": "switch.boiler", "room_entities": ["sensor.room"],
-                                 "room_max_age_minutes": 45})
+                                 "room_max_age_minutes": 45, "room_stale_use_last_value": True})
     result = await hass.config_entries.options.async_configure(result["flow_id"], main)
     result = await hass.config_entries.options.async_configure(
         result["flow_id"], result["data_schema"]({"outside_temperature_sources": ["sensor.outside"]}))
@@ -44,7 +45,14 @@ async def test_options_flow_keeps_settings(hass, entry, inputs):
     assert result["type"] is FlowResultType.CREATE_ENTRY
     await hass.async_block_till_done()
     assert entry.options["room_max_age_minutes"] == 45
-    assert hass.data[DOMAIN][entry.entry_id].config["room_max_age_minutes"] == 45
+    coordinator = hass.data[DOMAIN][entry.entry_id]
+    assert coordinator.config["room_max_age_minutes"] == 45
+    assert coordinator.config["room_stale_use_last_value"] is True
+    # The next options form shows the saved choice.
+    result = await hass.config_entries.options.async_init(entry.entry_id)
+    assert result["data_schema"]({"climate_entity": "climate.reference", "heater_entity": "switch.boiler",
+                                  "room_entities": ["sensor.room"]})["room_stale_use_last_value"] is True
+    hass.config_entries.options.async_abort(result["flow_id"])
 
 
 async def test_v1_entry_migrates_on_setup(hass, inputs):

tests/test_coordinator.py

@@ -539,3 +539,79 @@ class BoundaryTests(unittest.IsolatedAsyncioTestCase):
                          ('valid', 0, 2400))
         weather = data['input_quality']['weather']['temperature']['sensor.outside']
         self.assertEqual(weather['max_age_seconds_24h'], 0)  # kept reporting; tracked separately
+
+    def two_rooms(self, keep_last=False):
+        self.entry.data['room_entities'] = ['sensor.room', 'sensor.bedroom']
+        self.entry.data['weights'] = {'sensor.room': 3, 'sensor.bedroom': 1}
+        self.entry.data['room_stale_use_last_value'] = keep_last
+        self.set_state('sensor.bedroom', '18', unit_of_measurement='°C')
+        self.c = coordinator.HeatingMPCCoordinator(self.hass, self.entry)
+
+    def fresh(self, *entities):
+        for entity in entities or ('sensor.room', 'sensor.bedroom', 'sensor.outside'):
+            self.states[entity].last_reported = self.now
+
+    async def test_stale_room_is_left_out_and_other_rooms_continue(self):
+        self.two_rooms()
+        await self.sample()
+        self.fresh('sensor.room', 'sensor.outside')
+        self.states['sensor.bedroom'].last_reported = self.now - timedelta(minutes=31)
+        data = await self.sample()
+        self.assertEqual(data['status'], 'learning')
+        self.assertEqual(data['excluded_rooms'], ['sensor.bedroom'])
+        rooms = data['input_quality']['rooms']
+        self.assertEqual((rooms['sensor.bedroom']['status'], rooms['sensor.bedroom']['used']), ('stale', False))
+        self.assertTrue(rooms['sensor.room']['used'])
+        self.assertEqual((data['weighted_temperature'], data['weight_coverage']), (20, 0.75))
+        self.assertEqual(data['sample_result'], 'learned')
+        counts = lambda: (self.c.learners['sensor.room'].count, self.c.learners['sensor.bedroom'].count)
+        self.assertEqual(counts(), (1, 0))
+        self.fresh()  # bedroom reports again: no learning across its gap
+        data = await self.sample()
+        self.assertEqual((data['excluded_rooms'], data['weight_coverage']), ([], 1.0))
+        self.assertEqual(data['weighted_temperature'], 19.5)
+        self.assertEqual(counts(), (2, 0))
+        self.fresh()
+        await self.sample()
+        self.assertEqual(counts(), (3, 1))
+
+    async def test_left_out_room_is_not_used_for_prediction(self):
+        self.two_rooms()
+        model = self.c.learners['sensor.room']  # the bedroom stays untrained
+        model.count, model.on_count, model.off_count = 200, 100, 100
+        self.states['sensor.bedroom'].state = 'unavailable'
+        with patch.object(coordinator, 'optimize', wraps=coordinator.optimize) as spy:
+            data = await self.sample()
+        self.assertEqual(data['status'], 'ready')
+        temps, weights, learners = (spy.call_args.args[i] for i in (0, 4, 5))
+        self.assertEqual((set(temps), weights, set(learners)),
+                         ({'sensor.room'}, {'sensor.room': 3.0}, {'sensor.room'}))
+        self.assertEqual(set(data['prediction_60m']), {'sensor.room'})
+
+    async def test_all_rooms_unusable_still_pauses(self):
+        self.two_rooms()
+        for room in ('sensor.room', 'sensor.bedroom'):
+            self.states[room].last_reported = self.now - timedelta(minutes=31)
+        data = await self.sample()
+        self.assertEqual(data['status'], 'waiting_for_sensors')
+        self.assertEqual(data['excluded_rooms'], ['sensor.room', 'sensor.bedroom'])
+        self.assertIsNone(data['proposal'])
+
+    async def test_keep_last_value_option_uses_stale_room_for_up_to_a_day(self):
+        self.two_rooms(keep_last=True)
+        self.states['sensor.bedroom'].last_reported = self.now - timedelta(hours=2)
+        await self.sample()
+        self.fresh('sensor.room', 'sensor.outside')
+        data = await self.sample()
+        bedroom = data['input_quality']['rooms']['sensor.bedroom']
+        self.assertEqual((bedroom['status'], bedroom['used']), ('stale', True))
+        self.assertEqual((data['excluded_rooms'], data['temperatures']['sensor.bedroom']), ([], 18))
+        self.assertEqual(self.c.learners['sensor.bedroom'].count, 1)
+        for state, age in (('18', timedelta(hours=24, seconds=1)), ('unavailable', timedelta(0))):
+            with self.subTest(state=state):
+                self.fresh('sensor.room', 'sensor.outside')
+                self.states['sensor.bedroom'].state = state
+                self.states['sensor.bedroom'].last_reported = self.now - age
+                data = await self.sample()
+                self.assertEqual(data['excluded_rooms'], ['sensor.bedroom'])
+                self.assertEqual(data['status'], 'learning')

tests/test_data_quality.py

@@ -48,5 +48,7 @@ class FreshnessTests(unittest.TestCase):
             data = json.loads((ROOT / name).read_text())
             for section, step in (('config', 'user'), ('options', 'init')):
                 labels = data[section]['step'][step]['data']
-                for key in ('room_max_age_minutes', 'weather_max_age_minutes'):
+                for key in ('room_max_age_minutes', 'weather_max_age_minutes', 'room_stale_use_last_value'):
                     self.assertTrue(labels[key])
+                help_text = data[section]['step'][step]['data_description']
+                self.assertIn('24', help_text['room_stale_use_last_value'])