brewpi/components/sud.py

import json
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. A step is a
	ramp or a hold depending on which of those keys it specifies; the other
	is left out rather than synthesized from defaults."""
	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))
	for key in ('ramp', 'hold'):
		if key in raw_step:
			step[key] = _merge_defaults(defaults.get(key, {}), raw_step[key])
	return step


class Sud(AttributeChange):
	def __init__(self, path):
		AttributeChange.__init__(self)
		self.path = path
		with open(path) as f:
			data = json.load(f)

		(self.name, self.description, self.schedule,
			self.pot_mass, self.pot_material) = self._parse_data(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) 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')

		return name, description, schedule, pot_mass, pot_material

	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

	def save(self):
		"""Returns the sud.json document currently on disk, for a client to
		edit and hand back to load()."""
		with open(self.path) as f:
			return json.load(f)

	def load(self, data):
		"""Replaces the schedule with a new sud.json document and persists
		it to disk, resetting to IDLE as if freshly loaded from that file.
		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 = parsed
		with open(self.path, 'w') as f:
			json.dump(data, f, indent='\t')

		self._reset_run_state()
		return True

	def derive_plant_params(self, grain_mass, water_mass):
		"""Lumped (mass, specific heat) lifted from pot_mass/pot_material and
		the current step's grain_mass/water_mass, for use as Pot's "M"/"C"
		params. grain_mass/water_mass vary per step (e.g. malt going in,
		water boiling off), so this is recomputed on every step change
		rather than once at startup."""
		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,
		}

	def start(self):
		"""Starts a fresh run from IDLE, 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 != SudState.IDLE:
			return
		self.index = -1
		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:
			self._finish_step()

	def tick(self, 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]
		if 'ramp' in next_step:
			self.state = SudState.RAMPING
		else:
			self.hold_remaining = next_step['hold'].get('duration', 0)
			self.state = SudState.HOLDING

		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()