The following shows how to integrate an electricity traffic light (German: Stromampel) into OpenHab. The purpose is to optimize use of renewable energy, by starting power-hungry devices (washing machines, etc.) only when the share of renewables in the grid is high.
The website energy-charts.info provides information on energy production and spot market prices and also makes the data available through an API. The Mastodon account stromampelbot@climatejustice.social uses the data to make nice charts like this:
The following shows how to access and use the same data in OpenHab. Note that data on renewable shares is currently only available for Germany, Austria and France.
.items
String iTrafficLight_raw "Raw JSON: [%s]"
Number iCurrentPowerTrafficLight "Stromampel: [MAP(stromampel.map):%s]" <energy>
Number iNextPowerTrafficLight "NĂ€chste Ampelphase: [MAP(stromampel.map):%s]" <energy>
DateTime iNextPowerPeriodStart "Neue Phase ab: [%1$tH:%1$tM]" <time>
.rules (donât forget to adapt country in sendGetHttpGetRequest if youâre not in Germany)
// The following rules makes use of https://api.energy-charts.info to tell users when best to use power
rule "Get new power traffic light data"
when
Time cron "0 0 2/6 ? * *" // update every six hours, offset at 2AM (forecast data available around 19:00 give or take)
then
var response = sendHttpGetRequest("https://api.energy-charts.info/traffic_signal?country=de")
// some day introduce some checks on response here...
iTrafficLight_raw.postUpdate(response)
end
rule "Update Power traffic light"
when
Time cron "0 */15 * ? * *" // update at every 15th minute
then
var String input_string = (iTrafficLight_raw.state as StringType).toString
val json_arr_length = Integer::parseInt(transform("JSONPATH", "$.[1].data.length()", input_string))
// find index corresponding to current time
val long now_epoch = now.toInstant.toEpochMilli
var long data_epoch = 0
var now_index = -1
var i = 0
var boolean flag = true
while(flag) {
data_epoch = Long::parseLong(transform("JSONPATH", "$.[0].xAxisValues[" + i + "]", input_string))
if(Math::abs(data_epoch - now_epoch) < 450000) {
now_index = i
flag = false
}
if (i == (json_arr_length - 1)) {
flag = false
}
i = i + 1
}
// get current status
val current_signal = Integer::parseInt(transform("JSONPATH", "$.[1].data[" + now_index + "]", input_string))
iCurrentPowerTrafficLight.postUpdate(current_signal)
// stop here and return if we hit the end of time
if (i >= json_arr_length) {
iNextPowerTrafficLight.postUpdate(3)
iNextPowerPeriodStart.postUpdate(now)
return
}
// now check for next status and time of next status - starting from index i above
var data_signal = 0
var next_period_index = 0
flag = true
while(flag) {
data_signal = Integer::parseInt(transform("JSONPATH", "$.[1].data[" + i + "]", input_string))
if(data_signal != current_signal) {
next_period_index = i
flag = false
}
if (i == (json_arr_length - 1)) {
flag = false
if(data_signal != current_signal) {
// nothing changes until end of time, so we don't know next phase
data_signal = 3 // set to unknown
next_period_index = i // slightly misleading as this is not the next phase, but just the end of time
}
}
i = i + 1
}
val next_signal = data_signal
val next_period_epoch = Long::parseLong(transform("JSONPATH", "$.[0].xAxisValues[" + next_period_index + "]", input_string))
val next_period_time = Instant.ofEpochMilli(next_period_epoch).atZone(ZoneId.systemDefault())
iNextPowerTrafficLight.postUpdate(next_signal)
iNextPowerPeriodStart.postUpdate(new DateTimeType(next_period_time))
end
stromampel.map (all states are displayed as full circle; color is added in sitemap)
0=\u2B24
1=\u2B24
2=\u2B24
3=\u2B24
NULL=unknown
.sitemap
Text item=iCurrentPowerTrafficLight valuecolor=[>2="white",==2="green",==1="yellow",<=1="red"]
Text item=iNextPowerPeriodStart
Text item=iNextPowerTrafficLight valuecolor=[>2="white",==2="green",==1="yellow",<=1="red"]
The code has been running for a week now on my system and works well so far. Let me know if you bump into any issues.