brewpi/components/sud.py

from enum import Enum
from utils.value import AttributeChange

# Specific heat capacities [J/(kg*K)] used to derive a single lumped
# (mass, specific heat) pair for the pot's contents from pot_mass/
# grain_mass/water_mass. Approximate, typical values.
SPECIFIC_HEAT_WATER = 4190
SPECIFIC_HEAT_GRAIN = 1800
SPECIFIC_HEAT_BY_MATERIAL = {
	"Edelstahl 18/10": 500,
}


class SudState(Enum):
	IDLE = 0
	RAMPING = 1
	HOLDING = 2
	WAIT_USER = 3
	DONE = 4
	PAUSED = 5


def _merge_defaults(default, override):
	"""Deep-merges override onto default, recursing into nested dicts so
	values missing from override fall back to default's at every level."""
	merged = dict(default)
	for key, value in override.items():
		if key in merged and isinstance(merged[key], dict) and isinstance(value, dict):
			merged[key] = _merge_defaults(merged[key], value)
		else:
			merged[key] = value
	return merged


def _build_step(number, defaults, raw_step):
	"""Fills in a schedule step with default.step's values. 'ramp' is
	always synthesized from defaults, even if raw_step doesn't specify
	one of its own - every step ramps toward 'temperature' first, for as
	long as the controller's own gap-tracking FSM says it needs to (see
	Sud._advance()/temp_reached()), so its settings (e.g. 'rate') always
	have to be available. 'hold' stays opt-in: only a step that
	specifies it gets a hold-duration phase once the ramp is done.

	'temperature' is deliberately *not* defaulted from default.step (every
	sude/*.json's default.step.temperature is just inert template filler,
	never a real shared target) - a step omitting it has no new target of
	its own, so it's left None, telling SudTask not to push a new
	theta_soll and just keep whatever the previous step left running."""
	step = {'number': number}
	for key in ('descr', 'user_message', 'user_wait_for_continue', 'grain_mass', 'water_mass'):
		step[key] = raw_step.get(key, defaults.get(key))
	step['temperature'] = raw_step.get('temperature')
	step['ramp'] = _merge_defaults(defaults.get('ramp', {}), raw_step.get('ramp', {}))
	if 'hold' in raw_step:
		step['hold'] = _merge_defaults(defaults.get('hold', {}), raw_step['hold'])
	return step


EMPTY_SUD = {
	'Name': '',
	'Description': '',
	# pot_mass/pot_material are the physical kettle's own properties, not
	# really per-recipe (every sude/*.json so far uses this exact pot) -
	# defaulting to 0/None here used to make derive_plant_params() come out
	# with a zero-mass plant (a ZeroDivisionError in Pot.process() waiting
	# to happen) for as long as no Sud had ever been loaded yet. Matching
	# the real kettle here instead lets server/brewpi.py seed the Pot with
	# a physically sane default at startup, so its simulation actually
	# runs - and a real Sud's own doc still overrides these the moment one
	# is loaded (see SudTask.apply_plant_params()).
	'pot_mass': 5.96,
	'pot_material': "Edelstahl 18/10",
	# Pot energy loss coefficient [W/(kg*K)] and transport propagation
	# delay [s] - see derive_plant_params(). Defaults match the generic
	# values brewpi.py used to hardcode for every brew alike.
	'L': 0.2,
	'Td': 30,
	# Initial grain_mass/water_mass - what's actually in the pot before
	# the brew starts (e.g. water added but no malt yet), as opposed to
	# default.step's grain_mass/water_mass, which is just inert template
	# filler for steps that don't override it. Lets a caller derive
	# starting plant params directly (see SudTask.recv()'s Load handler)
	# without having to parse the first step out of the schedule.
	'grain_mass': 0,
	'water_mass': 0,
	'steps': [],
}


class Sud(AttributeChange):
	def __init__(self):
		"""Starts out with no schedule loaded - one of potentially several
		sude/*.json files on disk is brought in later via load(), kept
		purely in memory (see load()) until replaced or the server
		restarts."""
		AttributeChange.__init__(self)
		self._data = EMPTY_SUD

		(self.name, self.description, self.schedule, self.pot_mass,
			self.pot_material, self.L, self.Td, self.grain_mass,
			self.water_mass) = self._parse_data(self._data)

		self._paused_from = None
		self._reset_run_state()

	@staticmethod
	def _parse_data(data):
		"""Parses a sud.json document into the (name, description, schedule,
		pot_mass, pot_material, L, Td, grain_mass, water_mass) tuple Sud
		needs. Computed up front rather than assigned straight onto self,
		so a malformed load() can't leave a half-applied schedule in
		place."""
		name = data.get('Name', '')
		description = data.get('Description', '')

		step_defaults = data.get('default', {}).get('step', {})
		schedule = [_build_step(i + 1, step_defaults, raw) for i, raw in enumerate(data['steps'])]

		pot_mass = data.get('pot_mass', 0)
		pot_material = data.get('pot_material')
		L = data.get('L', EMPTY_SUD['L'])
		Td = data.get('Td', EMPTY_SUD['Td'])
		grain_mass = data.get('grain_mass', EMPTY_SUD['grain_mass'])
		water_mass = data.get('water_mass', EMPTY_SUD['water_mass'])

		return name, description, schedule, pot_mass, pot_material, L, Td, grain_mass, water_mass

	def _reset_run_state(self):
		"""Resets run-time progress back to a freshly-loaded, not-yet-started
		IDLE state. Shared by __init__, load(), and stop()."""
		self.index = -1
		self.hold_remaining = 0.0
		self.state = SudState.IDLE
		self.step = None
		self.user_message = None
		self._paused_from = None
		self.elapsed = 0.0

	def save(self):
		"""Returns the currently running sud.json document (EMPTY_SUD if
		none has been loaded yet), for a client to edit and hand back to
		load(). Purely in-memory - never read from or written to disk."""
		return self._data

	def load(self, data):
		"""Replaces the running schedule with a new sud.json document, kept
		in memory only (never written to disk - a client wanting to persist
		a schedule does so explicitly itself, e.g. by writing save()'s
		result to one of the sude/*.json files), resetting to IDLE as if
		freshly loaded. Refused while a brew is in progress, or if data is
		malformed - in either case the current schedule is left untouched."""
		if self.state not in (SudState.IDLE, SudState.DONE):
			return False
		try:
			parsed = self._parse_data(data)
		except (KeyError, TypeError):
			return False

		(self.name, self.description, self.schedule, self.pot_mass,
			self.pot_material, self.L, self.Td, self.grain_mass,
			self.water_mass) = parsed
		self._data = data

		self._reset_run_state()
		return True

	def derive_plant_params(self, grain_mass, water_mass):
		"""Full Pot plant params - "M"/"C" lumped (mass, specific heat)
		from pot_mass/pot_material and the current step's grain_mass/
		water_mass (these vary per step, e.g. malt going in, water
		boiling off, so this is recomputed on every step change rather
		than once at startup), plus "L"/"Td" straight from this Sud's
		own doc (constant for the whole brew - see Sud.load())."""
		c_pot = SPECIFIC_HEAT_BY_MATERIAL.get(self.pot_material, SPECIFIC_HEAT_WATER)
		mass = water_mass + grain_mass + self.pot_mass
		capacitance = (water_mass * SPECIFIC_HEAT_WATER
			+ grain_mass * SPECIFIC_HEAT_GRAIN
			+ self.pot_mass * c_pot)

		return {
			'M': mass,
			'C': capacitance / mass if mass > 0 else SPECIFIC_HEAT_WATER,
			'L': self.L,
			'Td': self.Td,
		}

	def start(self):
		"""Starts a fresh run from IDLE or DONE (i.e. restarts from the
		beginning once a previous run has finished), or resumes a paused
		one - whichever applies. No-op otherwise."""
		if self.state == SudState.PAUSED:
			self.state = self._paused_from
			self._paused_from = None
			return
		if self.state not in (SudState.IDLE, SudState.DONE):
			return
		self.index = -1
		self.elapsed = 0.0
		self._advance()

	def pause(self):
		"""Freezes progress (the hold countdown and ramp-reached checks both
		no-op while PAUSED) without losing where we are; start() resumes."""
		if self.state in (SudState.RAMPING, SudState.HOLDING):
			self._paused_from = self.state
			self.state = SudState.PAUSED

	def stop(self):
		"""Aborts the run, discarding progress back to IDLE."""
		if self.state not in (SudState.IDLE, SudState.DONE):
			self._reset_run_state()

	def confirm(self):
		if self.state == SudState.WAIT_USER:
			self._advance()

	def temp_reached(self):
		if self.state != SudState.RAMPING:
			return
		if 'hold' in self.step:
			# Same step, ramp phase done - move into its hold phase rather
			# than finishing the step. Re-assign self.step (AttributeChange
			# fires on every assignment, not just on change) to re-trigger
			# the step-changed callback so e.g. the hold's stirrer settings
			# get (re-)applied.
			self.hold_remaining = self.step['hold'].get('duration', 0) * 60.0
			self.state = SudState.HOLDING
			self.step = self.step
		else:
			self._finish_step()

	def tick(self, dt):
		"""Advances the run by dt simulated seconds. 'elapsed' is the tick-
		counted ground truth for how far the run has actually progressed -
		clients used to reconstruct this themselves from wall-clock time
		times the configured warp factor, but the warp factor is only
		nominal (real asyncio/Python scheduling overhead means the actual
		achieved speedup runs measurably below it, especially under heavy
		message/print load), making that reconstruction drift from the
		truth. Frozen while PAUSED/IDLE/DONE, same as hold_remaining."""
		if self.state not in (SudState.IDLE, SudState.DONE, SudState.PAUSED):
			self.elapsed += dt
		if self.state == SudState.HOLDING:
			self.hold_remaining -= dt
			if self.hold_remaining <= 0:
				self._finish_step()

	def _advance(self):
		self.index += 1
		if self.index >= len(self.schedule):
			self.state = SudState.DONE
			self.user_message = None
			self.step = None
			return

		next_step = self.schedule[self.index]
		# Every step ramps toward 'temperature' first - whether that takes
		# any real time depends on the actual gap, which only the
		# temperature controller's own FSM can tell (see temp_reached()'s
		# caller, SudTask.on_process()). 'hold' only starts counting down
		# once the controller reports the gap closed.
		self.state = SudState.RAMPING
		self.user_message = next_step.get('user_message')
		self.step = next_step

	def _finish_step(self):
		if self.step.get('user_wait_for_continue', False):
			self.state = SudState.WAIT_USER
		else:
			self._advance()