diff --git a/client/brewpi_gui.py b/client/brewpi_gui.py index 70c17a2..acfe3e8 100755 --- a/client/brewpi_gui.py +++ b/client/brewpi_gui.py @@ -129,10 +129,10 @@ class SudForecastPlot(FigureCanvasQTAgg): requiring user confirmation - a human's response time genuinely can't be forecast, so it's modeled as a zero-delay auto-confirm there instead of stopping. That assumption gets corrected once the - real confirmation actually happens (anchored at the real elapsed - time and temperature) - see tasks/sud.py's SudTask. - _continue_forecast_after_confirm() - at which point show_forecast() - is called again with the corrected curve, replacing the optimistic + schedule actually reaches the next real step boundary (anchored at + the real elapsed time and temperature) - see tasks/sud.py's + SudTask._reanchor_forecast() - at which point show_forecast() is + called again with the corrected curve, replacing the optimistic guess outright rather than the GUI patching it up itself.""" def __init__(self): @@ -894,9 +894,9 @@ class Window(QtWidgets.QMainWindow, Ui_MainWindow): # Computed asynchronously server-side and can arrive slightly # after the schedule itself - applies regardless of whether a # run is in progress, since this can also be a later piecewise - # segment appended after a real user confirmation (see - # tasks/sud.py's SudTask._continue_forecast_after_confirm()), not - # just the initial one computed at Load. + # segment appended after a real step transition (see tasks/ + # sud.py's SudTask._reanchor_forecast()), not just the initial + # one computed at Load. if self.sud_empty: return t_min = [seconds / 60.0 for seconds in forecast['T']] diff --git a/components/sud_forecast.py b/components/sud_forecast.py index dc7846c..303c3ae 100644 --- a/components/sud_forecast.py +++ b/components/sud_forecast.py @@ -66,23 +66,22 @@ class SudForecastEstimator: self.theta_amb = theta_amb def estimate(self, doc, start_theta=None): - """Returns (t, theta, final_state, confirm_points): t/theta are - parallel lists of elapsed simulated seconds and temperature, - covering doc['steps'] from the start all the way to the end - (final_state is SudState.DONE), or, in the pathological case of - a step whose target can never actually be reached, wherever - MAX_TICKS cut the simulation off. + """Returns (t, theta, final_state): t/theta are parallel lists of + elapsed simulated seconds and temperature, covering doc['steps'] + from the start all the way to the end (final_state is + SudState.DONE), or, in the pathological case of a step whose + target can never actually be reached, wherever MAX_TICKS cut the + simulation off. A step requiring user confirmation doesn't stop the simulation either - a human's response time genuinely can't be forecast, so it's modeled as zero delay (auto-confirmed the instant that step's hold completes) rather than leaving the estimate stuck - there forever. confirm_points records every place that - assumption was made, as (step_index, t) pairs, so the caller - (tasks/sud.py's SudTask) can correct it once a real - confirmation actually happens: truncate the forecast at that - point and splice in a freshly anchored simulation of the - remaining steps in place of the optimistic guess. + there forever. That optimistic guess gets corrected for real once + the caller (tasks/sud.py's SudTask) sees the schedule actually + reach the next step boundary - see SudTask._reanchor_forecast(), + called from on_step_changed() on every real transition, not just + confirmations. start_theta defaults to the configured ambient temperature - i.e. a cold start, same as the GUI's static estimate.""" @@ -91,7 +90,7 @@ class SudForecastEstimator: sud = Sud() if not sud.load(doc) or not sud.schedule: - return [0.0], [start_theta], SudState.DONE, [] + return [0.0], [start_theta], SudState.DONE # Plant params (M/C/L/Td) are deliberately *not* seeded here from # any default - sud.start() below synchronously fires the first @@ -157,13 +156,11 @@ class SudForecastEstimator: t = [0.0] theta = [pot.get_temperature()] - confirm_points = [] sud.start() ticks = 0 while sud.state != SudState.DONE and ticks < MAX_TICKS: if sud.state == SudState.WAIT_USER: - confirm_points.append((sud.index, t[-1])) sud.confirm() continue @@ -184,4 +181,4 @@ class SudForecastEstimator: theta.append(pot.get_temperature()) ticks += 1 - return t, theta, sud.state, confirm_points + return t, theta, sud.state diff --git a/tasks/sud.py b/tasks/sud.py index 27b5f38..17bbc45 100644 --- a/tasks/sud.py +++ b/tasks/sud.py @@ -59,13 +59,13 @@ class SudTask(ATask): # one. self.forecast_estimator = forecast_estimator # The forecast as last computed/corrected (see send_forecast()/ - # _continue_forecast_after_confirm()) - T in simulated seconds, - # Theta in degrees, parallel lists, both monotonically growing as - # corrections get spliced in. Covers the whole schedule from the - # very first Load, including through steps requiring user - # confirmation - those are simulated as a zero-delay auto-confirm - # (see components/sud_forecast.py's SudForecastEstimator. - # estimate()) rather than left unforecast. + # _reanchor_forecast()) - T in simulated seconds, Theta in degrees, + # parallel lists, both monotonically growing as corrections get + # spliced in. Covers the whole schedule from the very first Load, + # including through steps requiring user confirmation - those are + # simulated as a zero-delay auto-confirm (see components/ + # sud_forecast.py's SudForecastEstimator.estimate()) rather than + # left unforecast, until _reanchor_forecast() corrects it for real. self.forecast_t = [] self.forecast_theta = [] # Whether the forecast above actually reaches the schedule's real @@ -73,14 +73,13 @@ class SudTask(ATask): # of a step whose target can never be reached (see # components/sud_forecast.py's MAX_TICKS). self.forecast_finished = True - # Where each user-confirmation step's zero-delay assumption sits - # in the forecast timeline above, keyed by the schedule's - # (absolute) step index - see SudForecastEstimator.estimate()'s - # confirm_points. Consulted by _continue_forecast_after_confirm() - # to know where to cut the optimistic guess loose and splice in a - # freshly anchored simulation once that confirmation actually - # happens for real. - self.forecast_confirm_marks = {} + # Bumped by every call to send_forecast()/_reanchor_forecast() - + # see either's own comment for why: it lets a call that's still + # awaiting its worker-thread simulation tell, once it resumes, + # whether a newer call has since started and already committed a + # fresher result - if so, it discards its own rather than + # corrupting forecast_t/forecast_theta with stale data. + self._forecast_generation = 0 msg_handler.set_recv_handler(self.recv) def apply_plant_params(self, grain_mass, water_mass): @@ -131,6 +130,14 @@ class SudTask(ATask): self.tc.set_theta_soll(step['temperature']) self.tc.set_heatrate_soll(ramp['rate']) self.apply_stirrer(phase) + # Every real step boundary (full step change or ramp->hold + # within one) is a trustworthy checkpoint to correct the + # forecast against - see _reanchor_forecast(). Catches drift + # from anything the original simulation couldn't have known + # (a malt fill-in's actual cooldown, a longer/shorter ramp than + # modeled, ...) at the next opportunity, not just at the next + # user confirmation. + asyncio.create_task(self._reanchor_forecast()) asyncio.create_task(self.send({'Step': { 'Index': self.sud.index, @@ -184,31 +191,46 @@ class SudTask(ATask): actually will, never lining up with the actual trace even at t=0. Deliberately the raw sensor reading (theta_ist_set), not the controller's own get_theta_ist() (theta_ist) - - _continue_forecast_after_confirm() can trust that one because a - real run has been actively ticking for a while by the time it - runs, but this is called right after Load, before this - controller may have processed even a single tick yet (e.g. its - plant params/model only just got configured - see SudTask. - recv()), so theta_ist itself could still be sitting at its - never-updated __init__ default. + _reanchor_forecast() can trust that one because a real run has + been actively ticking for a while by the time it runs, but this + is called right after Load, before this controller may have + processed even a single tick yet (e.g. its plant params/model + only just got configured - see SudTask.recv()), so theta_ist + itself could still be sitting at its never-updated __init__ + default. - Those zero-delay assumptions get corrected piecewise as real - confirmations actually happen - see - _continue_forecast_after_confirm().""" + Those zero-delay assumptions get corrected piecewise as the + schedule actually reaches each step boundary - see + _reanchor_forecast().""" if self.forecast_estimator is None: return + # Both this and _reanchor_forecast() can end up running + # concurrently - e.g. a fresh Start triggers this explicitly *and* + # (via Sud.start() synchronously firing on_step_changed() for the + # first step) a _reanchor_forecast() of its own; a step whose hold + # duration is already 0 can likewise advance twice within a single + # tick, firing on_step_changed() twice back to back. Each such call + # awaits a worker-thread simulation, so without this guard whichever + # one resumes second would blindly splice its own tail onto + # whatever the other already finished writing, producing a + # spurious connecting line across the plot. Bumping/checking this + # generation counter across the await ensures only the very latest + # call's result is ever committed - any older one discards itself. + self._forecast_generation += 1 + generation = self._forecast_generation # Runs the simulation in a worker thread - it's CPU-bound and can # take a couple hundred ms for a long schedule, which would # otherwise stall every other task (heater, sensor, ...) for that # whole window. loop = asyncio.get_event_loop() start_theta = self.tc.get_theta_ist_set() - t, theta, final_state, confirm_points = await loop.run_in_executor( + t, theta, final_state = await loop.run_in_executor( None, self.forecast_estimator.estimate, doc, start_theta) + if generation != self._forecast_generation: + return self.forecast_t = t self.forecast_theta = theta self.forecast_finished = (final_state == SudState.DONE) - self.forecast_confirm_marks = dict(confirm_points) await self._send_forecast() async def _send_forecast(self): @@ -219,43 +241,49 @@ class SudTask(ATask): 'Finished': self.forecast_finished, }}) - async def _continue_forecast_after_confirm(self, confirmed_index, confirmed_target): - """Corrects the optimistic, zero-delay guess send_forecast() (or a - previous call to this method) made at confirmed_index's - user-confirmation step, now that the confirmation has actually - happened for real - see SudForecastEstimator.estimate()'s - docstring for why that assumption can't just be trusted as-is. - Truncates the forecast right back to that point and splices in a - freshly anchored simulation of the rest of the schedule, anchored - at the real elapsed time (self.sud.elapsed) - so an actual delay - shows up honestly as a gap rather than as the assumed zero - and - the real current temperature (more trustworthy than whatever the - now-superseded guess predicted it would be by now). + async def _reanchor_forecast(self): + """Corrects the optimistic, zero-delay guesses baked into the + forecast (see SudForecastEstimator.estimate()'s docstring) now + that the schedule has actually reached a real step boundary - + called from on_step_changed() on every transition, whether it's + a user confirmation or fully automatic (e.g. a ramp reaching its + target, or a hold's duration running out). Truncates the forecast + back to right now and splices in a freshly anchored simulation of + the rest of the schedule, anchored at the real elapsed time + (self.sud.elapsed) and the real current temperature + (self.tc.get_theta_ist()) - so any divergence the original + simulation couldn't have predicted (a malt fill-in's actual + cooldown, a longer/shorter ramp than modeled, ...) gets corrected + at the next opportunity instead of leaving the forecast stuck + showing what was once guessed. - confirmed_target is the real controller's theta_soll_set at the - moment of confirmation (captured by recv() *before* calling - Sud.confirm(), since that synchronously pushes the next step's - own target onto it) - used to fill the real-world wait itself - (see below). + No-op if the estimator isn't configured. - No-op if confirmed_index was never part of a computed forecast in - the first place (e.g. the estimator isn't configured).""" + Guards against the same concurrent-call race send_forecast() does + (see its own comment) by working off local copies of the forecast + lists throughout, only ever committing them to self.forecast_t/ + forecast_theta right at the end, and only if this is still the + latest call by then - an older call resuming after a newer one + has already committed must discard its own (now-stale) result + rather than splice it onto what the newer call already wrote.""" if self.forecast_estimator is None: return - mark_t = self.forecast_confirm_marks.pop(confirmed_index, None) - if mark_t is None: - return - # Drop the now-stale tail (everything beyond the confirmation - # point) - both the curve itself and any confirm marks that fell - # within it, since they're about to be replaced by fresh ones. - cut = bisect.bisect_right(self.forecast_t, mark_t) - self.forecast_t = self.forecast_t[:cut] - self.forecast_theta = self.forecast_theta[:cut] - self.forecast_confirm_marks = {idx: t for idx, t in self.forecast_confirm_marks.items() if t <= mark_t} + self._forecast_generation += 1 + generation = self._forecast_generation + real_elapsed = self.sud.elapsed + # Drop the now-stale tail (everything beyond right now) - it's + # about to be replaced by a freshly anchored simulation. + cut = bisect.bisect_right(self.forecast_t, real_elapsed) + forecast_t = self.forecast_t[:cut] + forecast_theta = self.forecast_theta[:cut] schedule = self.sud.schedule index = self.sud.index if not (0 <= index < len(schedule)): + if generation != self._forecast_generation: + return + self.forecast_t = forecast_t + self.forecast_theta = forecast_theta self.forecast_finished = True await self._send_forecast() return @@ -271,26 +299,22 @@ class SudTask(ATask): 'steps': schedule[index:], } start_theta = self.tc.get_theta_ist() - real_elapsed = self.sud.elapsed loop = asyncio.get_event_loop() - t, theta, final_state, confirm_points = await loop.run_in_executor(None, self.forecast_estimator.estimate, doc, start_theta) - # Bridge the gap between the confirmation point and the real - # elapsed time it actually happened at - the real controller - # stays enabled and actively holding through WAIT_USER, so - # confirmed_target (its setpoint at the time) is what it was - # actually converging toward during the wait, not whatever - # (possibly still mid-ramp) value the forecast happened to log - # the instant WAIT_USER tripped. - if self.forecast_t and self.forecast_t[-1] < real_elapsed: - self.forecast_t.append(real_elapsed) - self.forecast_theta.append(confirmed_target) - self.forecast_t.extend(real_elapsed + seconds for seconds in t) - self.forecast_theta.extend(theta) - # confirm_points' indices/times are relative to this sub-schedule - # (starting fresh at doc['steps'][0]) - rebase both onto the real - # schedule's absolute indices and the master forecast timeline. - self.forecast_confirm_marks.update( - (index + local_index, real_elapsed + local_t) for local_index, local_t in confirm_points) + t, theta, final_state = await loop.run_in_executor(None, self.forecast_estimator.estimate, doc, start_theta) + if generation != self._forecast_generation: + return + # Bridge any gap between the last surviving old point and right + # now (e.g. the old forecast's timeline had already drifted + # behind real_elapsed) with the same real measurement the fresh + # simulation below starts from, so the spliced curve doesn't + # visibly jump. + if forecast_t and forecast_t[-1] < real_elapsed: + forecast_t.append(real_elapsed) + forecast_theta.append(start_theta) + forecast_t.extend(real_elapsed + seconds for seconds in t) + forecast_theta.extend(theta) + self.forecast_t = forecast_t + self.forecast_theta = forecast_theta self.forecast_finished = (final_state == SudState.DONE) await self._send_forecast() @@ -309,10 +333,10 @@ class SudTask(ATask): if fresh_start: asyncio.create_task(self.send_forecast(self.sud.save())) elif 'Confirm' in pair[0]: - confirmed_index = self.sud.index - confirmed_target = self.tc.get_theta_soll_set() + # Sud.confirm() synchronously fires on_step_changed() for + # the now-current step, which schedules the forecast + # reanchor itself - see _reanchor_forecast(). self.sud.confirm() - asyncio.create_task(self._continue_forecast_after_confirm(confirmed_index, confirmed_target)) elif 'Pause' in pair[0]: self.sud.pause() elif 'Stop' in pair[0]: diff --git a/tasks/sud_log.py b/tasks/sud_log.py index cf24c89..cc76ae2 100644 --- a/tasks/sud_log.py +++ b/tasks/sud_log.py @@ -33,8 +33,8 @@ class SudLogTask(ATask): is of no use before the run is over anyway. The forecast file is written from SudTask.forecast_t/forecast_theta as they stand at that point - the *final*, most-corrected version (see tasks/sud.py's - SudTask._continue_forecast_after_confirm()), not the optimistic - guess computed back at Load. + SudTask._reanchor_forecast()), not the optimistic guess computed + back at Load. Samples are taken once per interval regardless of Sud.state (RAMPING/HOLDING/WAIT_USER/PAUSED all keep recording, mirroring