Finished State Machine in Jython

Tutorials & Examples
,
1

Based on this excellent post Washing Machine State Machine that I used in DSL, I finished my transition to Python and searched a generic approach that led me to use “transition” library.
Here’s where I’m now landed :

from transitions import Machine,State
from core.log import logging, LOG_PREFIX, log_traceback
from core.rules import rule
from core.triggers import when
from threading import Timer
from personal.notification import notification

class MyMachine(Machine):
	def __init__(self, name, powersource, nicename, thresholds):
		self.thresholds = thresholds
		self.eqptname = name
		self.powersource = powersource
		self.nicename = nicename

		Machine.__init__(self, name=name, states=['OFF','STANDBY','ACTIVE','FINISHED'], initial='OFF', after_state_change='enter_state')
		
		self.add_transition('consume', 'OFF', 'STANDBY', conditions='is_hidle')
		self.add_transition('consume', 'ACTIVE', 'FINISHED', conditions='is_hidle')
		self.add_transition('consume', '*', 'ACTIVE', conditions='is_running')
		self.add_transition('consume', '*', 'OFF', conditions='is_stopped')

	def is_stopped(self, power): 
		return self.state !='OFF' and power < self.thresholds[0]

	def is_hidle(self, power):
		return power >= self.thresholds[0] and power < self.thresholds[1]

	def is_running(self, power):
		return self.state!='ACTIVE' and power >= self.thresholds[2]

	def enter_state(self, power):
		notification(u"{} {}".format(self.nicename, self.state))
		events.postUpdate(self.eqptname + "_opstate",self.state)
		if self.state == "FINISHED":
			Timer(300,applianceTimerScript,[self]).start()

def applianceTimerScript(*args):
	machine = args[0]
	if machine.state == "FINISHED":
		notification(u"Cycle {} fini depuis 5mn, je coupe le courant".format(machine.nicename))
		events.sendCommand(machine.powersource,"OFF")

machine_array = [ 
	MyMachine("lavelinge","prise5_energy", u"Lave-Linge",[1.2,2,5]), 
	MyMachine("sechelinge","prise6_energy", u"Sèche-Linge",[1.4,2,10]) 
]

def machine_trigger_generator(machine_array):
	def generated_triggers(function):
		for machine in machine_array:
			when("Item {} changed".format(machine.powersource))(function)
		return function
	return generated_triggers

@rule("FSM : equipement consumes")
@machine_trigger_generator(machine_array)
def state_machine_lave_linge(event):
	declencheur = str(event.itemName)
	for fsm in machine_array:
		if fsm.powersource == declencheur: 
			puissance = event.itemState.floatValue()
			fsm.consume(puissance)

This seems to work quite well. I’m not sure to be using the Transitions at its full power, but it’s a start.

11 Likes
2

You beat me to it. That’s been on my todo list for awhile. To my eye it’s even simpler and easier to follow than the Rules DSL verion. Thanks for posting!

I wonder if there is some way to define the transitions externally, such as in configuration.py, and make this generic. Though once you move the transitions and functions outside the class there isn’t much left so there may not be much benefit. But it would be cool if we could provide a state machine library module or script where users just need to fill out a table in a config file or something like that and have one instead of needing to copy/paste/edit.

I’ll have to think on it. But first, adding Ephemeris to my generic TimeOfDay implementation. :smiley:

3

Congrats! We should get a special star next to our avatars! It would be nice if you would submit this to the helper library repo as an example script, to keep things in one place… things tend to get lost in the forum.

After seeing your use of unicode, I wonder if a review of the Python helper libraries might be needed to incorporate unicode into the logging. :thinking:

Thank you for sharing transitions… it’s definitely an interesting library! I’m still scratching my head trying to understand all the functionality it has that overlaps with OH. It has states, events, triggers, conditions, actions, etc. It’s like a mini combination event bus and rule engine!

This is not used… probably a remnant from my ToD example which pulls an OrderedDict from configuration.py for use in a trigger generator? :+1:

Side topic… I’m curious what you have in here. I have setup a similar module and plan to post it in the community section of the HL repo. Mine is handling audio, kodi and HABdroid notifications, with consideration for mode (no need to tell the house guests that the laundry is done!) and presence.

4

This would be great, beside the fact that for French it’s nearly mandatory (éèàêë…), I use a lot of unicode symbols in notifications targetted to Telegram.

Yes, but not so obvious to master. It took me a bit of time to understand how to handle conditional transitions, and I still have an issue to inhibit reflexive transitions, reason for this :

return self.state !='OFF' and power < self.thresholds[0]

that I do not find very clean.

Exactly :wink:

My notification is not great, more a bootstrap to initiate things :

# coding: utf-8

from core.jsr223 import scope
from core.actions import NotificationAction, Telegram, Voice
from configuration import admin_email, alert_email
from core.log import logging, LOG_PREFIX
from org.eclipse.smarthome.core.thing import ThingUID
from core.actions import Things

log = logging.getLogger("{}.personal.notification".format(LOG_PREFIX))

def notification(message, priority=0, android=True, mail=False, audio=False, logging=True): # priority 0 is info, 1 warning, 2 error
	
	if priority == 0:
		enhanced_message = u'\u2139' + ' ' + message 
	elif priority == 1:
		enhanced_message = u'\u26A0' + ' ' + message
	elif priority ==2:
		enhanced_message = u'\u2757' + ' ' + message
	else:
		enhanced_message = message
		
	if logging==True:
		log.info(enhanced_message)
	
	current_mode = scope.items["Mode"].toString()
	if android:
		if priority > 0 or current_mode in ["Night", "Late"]:
			NotificationAction.sendBroadcastNotification(enhanced_message)
		else:
			NotificationAction.sendNotification(admin_email, enhanced_message)
		Telegram.sendTelegram("bot1", enhanced_message)
	if audio:
		thingStatusInfo = Things.getThingStatusInfo('amazonechocontrol:echo:glh:echo')
		status = thingStatusInfo.getStatus().toString()
		if status == "ONLINE":
			message = message.replace("Openhab","Opeunne hab")
			message = message.replace("ACTIVE","en fonction")
			message = message.replace("STANDBY",u"est prêt")
			message = message.replace("FINISHED",u"est terminé")
		
			#scope.events.sendCommand("Echo_Dot_PlayAlarmSound",'ECHO:system_alerts_melodic_01')
			scope.events.sendCommand("Echo_Dot_TTS", message)
		else:
			log.warn("Echo dot is {}, I can not send him the message".format(status))
		Voice.say(message)
5

How do you install that library for it to work with jython?

6

I’m not sure it is the best way to proceed but this works :

pip install --target=/etc/openhab2/automation/lib/python transitions
3 Likes