Home Automation – Picnic Projects

December 4, 2018December 4, 2018

Rotary Phone Home Control

In our house old rotary phones are used as a remote for Sonos, Philips Hue and to ring the door bell. Turning the dial on one of the phones starts playing an MP3 on the phone. Turning the dial of the phone in another room dims the light, or turns it on or off.

We use an old Ericofoon, a red and green Dutch T65:

I connected the 2 pins of the phones to an ESP8226 (nodemcu). You can buy them for cheap on Chinese webshops.

The C code below is counting pulses. I used platformio to compile the ESP8226 code and upload it. The ESP8226 requests a Raspberry Pi to evaluate an URL. The Python code for the Raspberry Pi will be subject of another post.

/* DHT */
# define DHTTYPE DHT11
# define DHTPIN D2

DHT dht(DHTPIN, DHTTYPE);

const char* ssid = "mywifi";
const char* password = "blackmirr0r";

/* LED */
Chrono chrono_led; 
int    pin_led = D4; // LED_BUILTIN
int    state_led = LOW;

/* Rotary All */
enum   enum_state {DOWN_FIRST, UP, DOWN};

/* server */
ESP8266WebServer server(80);

void log(int dial, int elapsed, int state)
{
   if (state == 0)
   {
      Serial.printf("dial=%5d, elapsed=%5d, state=DOWN_FIRST\n", dial, elapsed);
   }
   else if (state == 1)
   {
      Serial.printf("dial=%5d, elapsed=%5d, state=UP\n", dial, elapsed);
   }
   else
   {
      Serial.printf("dial=%5d, elapsed=%5d, state=DOWN\n", dial, elapsed);
   }
}

void call_server(int device, int dial)
{
   HTTPClient http;
   int httpCode;
   char url[128];

   sprintf(url, "http://192.168.1.11:8002/casa/index.php?action=dial&amp;dial=%d&amp;device=%d", dial, device);
   Serial.println(url);
 
   http.begin(url);

   httpCode = http.GET(); // httpCode will be negative on error
   if (httpCode &gt; 0)
   {
      if (httpCode == HTTP_CODE_OK)
      {
         String payload = http.getString();
         Serial.println(payload);
      }
   } 
   else
   {
      Serial.printf("[HTTP] GET... failed, error: %s\n", http.errorToString(httpCode).c_str());
   }

   http.end();
}

void loop_blink_led()
{
   if (state_led == LOW)
   {
      if (chrono_led.elapsed() &gt; 1000)
      {
         state_led = HIGH;
         digitalWrite(pin_led, state_led);
         chrono_led.restart();
      }
   }
   else
   {
      if (chrono_led.elapsed() &gt; 10000)
      {
         state_led = LOW;
         digitalWrite(pin_led, state_led);
         chrono_led.restart();
      }
   }
}

class Rotary
{
   public:
      Rotary(int, int, double, double);
      void setup();
      void loop();
      void print();
   private:
      Chrono chrono_rotary; 
      int    pin_in  = D6;
      int    device;
      int    previous_value; 
      double period_min = 25; // in ms; minimum duration of a high 
      double period_max = 80; // in ms; maximum duration of a full period (1 high + 1 low)
      int    dial = 0;
      enum_state state = DOWN_FIRST;
};

Rotary::Rotary(int d, int pin, double pmin, double pmax)
{
   pin_in = pin;
   period_min = pmin;
   period_max = pmax;
   device = d;
};

void Rotary::setup()
{
   pinMode(pin_in, INPUT_PULLUP);
   chrono_rotary.restart();
};

void Rotary::loop()
{
   int val;
   val = digitalRead(pin_in);
   
   switch(state)
   {
      case DOWN_FIRST:
         if (val == HIGH)
         {
            state = UP;
            log(dial,chrono_rotary.elapsed(),state);
            chrono_rotary.restart();
         }
         break;
      
      case UP:
         if (val == LOW)
         {
            if (chrono_rotary.elapsed() &gt; period_min)
            {
               state = DOWN;
               dial++;
               log(dial,chrono_rotary.elapsed(),state);
               chrono_rotary.restart();
            }
            else
            {
               state = DOWN_FIRST;
               log(dial,chrono_rotary.elapsed(),state);
            }
         }
         break;

      case DOWN:
         if (chrono_rotary.elapsed() &gt; period_max)
         {
            state = DOWN_FIRST;
            log(dial,chrono_rotary.elapsed(),state);
            call_server(device, dial);
            dial = 0;
         }
         else if ((val == HIGH) &amp;&amp; (chrono_rotary.elapsed() &gt; period_min))
         {
            state = UP;
            log(dial,chrono_rotary.elapsed(),state);
            chrono_rotary.restart();
         }
         break;
   }      
};
void Rotary::print()
{
   int val;
   val = digitalRead(pin_in);
   if (previous_value != val)
   {
      Serial.printf("val=%5d, elapsed=%5d\n", val, chrono_rotary.elapsed());
      chrono_rotary.restart();
      previous_value = val;
   }
}   

Rotary T65_red  (11, D6, 25, 80);
Rotary ericofoon(13, D7, 1, 130);

void setup() {
   Serial.begin(9600);
   Serial.println("T65 Rotary");
   
   T65_red.setup();
   ericofoon.setup();
   
   pinMode(pin_led, OUTPUT);
   chrono_led.restart();
      
   WiFi.mode(WIFI_STA);
   WiFi.begin(ssid, password);
   while (WiFi.waitForConnectResult() != WL_CONNECTED) {
      Serial.println("Connection Failed! Rebooting...");
      delay(5000);
      ESP.restart();
   }

   Serial.println("Ready");
   Serial.print("IP address: ");
   Serial.println(WiFi.localIP());
   
   dht.begin();
   
   server.on("/", [](){
      String buf = "";
      float rh = dht.readHumidity();
      float temp = dht.readTemperature();
      // buf = "T=" + String(temp) + ",RH=" + String(rh);
      buf = String(temp) + "," + String(rh);
      Serial.println(buf);
      server.send(200, "text/plain", buf);
      state_led = LOW;
      digitalWrite(pin_led, state_led);
      delay(1000);
      state_led = HIGH;
      digitalWrite(pin_led, state_led);
   });
   server.begin();
}

void loop()
{
   server.handleClient();
   T65_red.loop();
   delay(0.1);
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

/* DHT */

# define DHTTYPE DHT11

# define DHTPIN D2

DHT dht(DHTPIN, DHTTYPE);

const char* ssid = "mywifi";

const char* password = "blackmirr0r";

/* LED */

Chrono chrono_led;

int pin_led = D4; // LED_BUILTIN

int state_led = LOW;

/* Rotary All */

enum enum_state {DOWN_FIRST, UP, DOWN};

/* server */

ESP8266WebServer server(80);

void log(int dial, int elapsed, int state)

{

if (state == 0)

{

Serial.printf("dial=%5d, elapsed=%5d, state=DOWN_FIRST\n", dial, elapsed);

}

else if (state == 1)

{

Serial.printf("dial=%5d, elapsed=%5d, state=UP\n", dial, elapsed);

}

else

{

Serial.printf("dial=%5d, elapsed=%5d, state=DOWN\n", dial, elapsed);

}

void call_server(int device, int dial)

{

HTTPClient http;

int httpCode;

char url[128];

sprintf(url, "http://192.168.1.11:8002/casa/index.php?action=dial&dial=%d&device=%d", dial, device);

Serial.println(url);

http.begin(url);

httpCode = http.GET(); // httpCode will be negative on error

if (httpCode > 0)

{

if (httpCode == HTTP_CODE_OK)

{

String payload = http.getString();

Serial.println(payload);

}

else

{

Serial.printf("[HTTP] GET... failed, error: %s\n", http.errorToString(httpCode).c_str());

}

http.end();

}

void loop_blink_led()

{

if (state_led == LOW)

{

if (chrono_led.elapsed() > 1000)

{

state_led = HIGH;

digitalWrite(pin_led, state_led);

chrono_led.restart();

}

else

{

if (chrono_led.elapsed() > 10000)

{

state_led = LOW;

digitalWrite(pin_led, state_led);

chrono_led.restart();

}

class Rotary

{

public:

Rotary(int, int, double, double);

void setup();

void loop();

void print();

private:

Chrono chrono_rotary;

int pin_in = D6;

int device;

int previous_value;

double period_min = 25; // in ms; minimum duration of a high

double period_max = 80; // in ms; maximum duration of a full period (1 high + 1 low)

int dial = 0;

enum_state state = DOWN_FIRST;

};

Rotary::Rotary(int d, int pin, double pmin, double pmax)

{

pin_in = pin;

period_min = pmin;

period_max = pmax;

device = d;

};

void Rotary::setup()

{

pinMode(pin_in, INPUT_PULLUP);

chrono_rotary.restart();

};

void Rotary::loop()

{

int val;

val = digitalRead(pin_in);

switch(state)

{

case DOWN_FIRST:

if (val == HIGH)

{

state = UP;

log(dial,chrono_rotary.elapsed(),state);

chrono_rotary.restart();

}

break;

case UP:

if (val == LOW)

{

if (chrono_rotary.elapsed() > period_min)

{

state = DOWN;

dial++;

log(dial,chrono_rotary.elapsed(),state);

chrono_rotary.restart();

}

else

{

state = DOWN_FIRST;

log(dial,chrono_rotary.elapsed(),state);

}

break;

case DOWN:

if (chrono_rotary.elapsed() > period_max)

{

state = DOWN_FIRST;

log(dial,chrono_rotary.elapsed(),state);

call_server(device, dial);

dial = 0;

}

else if ((val == HIGH) && (chrono_rotary.elapsed() > period_min))

{

state = UP;

log(dial,chrono_rotary.elapsed(),state);

chrono_rotary.restart();

}

break;

}

};

void Rotary::print()

{

int val;

val = digitalRead(pin_in);

if (previous_value != val)

{

Serial.printf("val=%5d, elapsed=%5d\n", val, chrono_rotary.elapsed());

chrono_rotary.restart();

previous_value = val;

}

Rotary T65_red (11, D6, 25, 80);

Rotary ericofoon(13, D7, 1, 130);

void setup() {

Serial.begin(9600);

Serial.println("T65 Rotary");

T65_red.setup();

ericofoon.setup();

pinMode(pin_led, OUTPUT);

chrono_led.restart();

WiFi.mode(WIFI_STA);

WiFi.begin(ssid, password);

while (WiFi.waitForConnectResult() != WL_CONNECTED) {

Serial.println("Connection Failed! Rebooting...");

delay(5000);

ESP.restart();

}

Serial.println("Ready");

Serial.print("IP address: ");

Serial.println(WiFi.localIP());

dht.begin();

server.on("/", [](){

String buf = "";

float rh = dht.readHumidity();

float temp = dht.readTemperature();

// buf = "T=" + String(temp) + ",RH=" + String(rh);

buf = String(temp) + "," + String(rh);

Serial.println(buf);

server.send(200, "text/plain", buf);

state_led = LOW;

digitalWrite(pin_led, state_led);

delay(1000);

state_led = HIGH;

digitalWrite(pin_led, state_led);

});

server.begin();

}

void loop()

{

server.handleClient();

T65_red.loop();

delay(0.1);

}

December 4, 2018December 4, 2018

Vintage Phone Home Control Center

An old phone, a raspberry pi, a handful of resistors and a simple python script make a very nice home control center.
The phone is a T65 TDK. It was very popular in the Netherlands in the 70’s and 80’s. It comes in a few colors, but preferably orange.

The phone now controls our Sonos audio system and our Philips Hue lights. The little black button rings the front door bell, great gimmick for the kids.

Button functions:

1: a radio station
2: volume up
3: another radio station
4: previous song
5: play / pause
6: next song
7: a playlist
8: volume down
9: another playlist
*: Philips Hue off
0: Philips Hue dimmed
#: Philips Hue bright

And the Python script

import RPi.GPIO as GPIO
import urllib2, socket
import iot

# Use pin numbers (not GPIO numbers)
# these are the inner rows with numbers in the circles
# [3,5,7,8,10,11,12,13,15,16,18,19,21,22,23,24,26]
      
def main_loop():
   col1 = GPIO.input(10)
   col2 = GPIO.input(16)
   col3 = GPIO.input(26)
   
   row1 = GPIO.input(22)
   row2 = GPIO.input(8)
   row3 = GPIO.input(12)
   row4 = GPIO.input(24)
   
   but  = GPIO.input(18)

   key = -1
   
   if col1 == True and row1 == True:
      key = 1
   elif col2 == True and row1 == True:
      key = 2
   elif col3 == True and row1 == True:
      key = 3
   elif col1 == True and row2 == True:
      key = 4
   elif col2 == True and row2 == True:
      key = 5
   elif col3 == True and row2 == True:
      key = 6
   elif col1 == True and row3 == True:
      key = 7
   elif col2 == True and row3 == True:
      key = 8
   elif col3 == True and row3 == True:
      key = 9
   elif col1 == True and row4 == True:
      key = 10
   elif col2 == True and row4 == True:
      key = 11
   elif col3 == True and row4 == True:
      key = 12
   elif but == True:
      key = 13
      
   if key <> -1:
      url = 'http://192.168.1.11:8002/casa/index.php?action=dial&device=10&dial=%d' % key
      try: 
         response = urllib2.urlopen(url, timeout=5)
      except IOError:
         iot.iot_send_string(999,'T65 Orange time out: ' + url)

GPIO.setmode(GPIO.BOARD)

pins = [8,10,12,16,18,22,24,26]
for pin in pins:
   GPIO.setup(pin, GPIO.IN)
   
iot.iot_send_string(999,'T65 Orange booted')

while 1:
   try:
      main_loop()
   except:
      iot.iot_send_trace()

import RPi.GPIO as GPIO

import urllib2, socket

import iot

# Use pin numbers (not GPIO numbers)

# these are the inner rows with numbers in the circles

# [3,5,7,8,10,11,12,13,15,16,18,19,21,22,23,24,26]

def main_loop():

col1 = GPIO.input(10)

col2 = GPIO.input(16)

col3 = GPIO.input(26)

row1 = GPIO.input(22)

row2 = GPIO.input(8)

row3 = GPIO.input(12)

row4 = GPIO.input(24)

but = GPIO.input(18)

key = -1

if col1 == True and row1 == True:

key = 1

elif col2 == True and row1 == True:

key = 2

elif col3 == True and row1 == True:

key = 3

elif col1 == True and row2 == True:

key = 4

elif col2 == True and row2 == True:

key = 5

elif col3 == True and row2 == True:

key = 6

elif col1 == True and row3 == True:

key = 7

elif col2 == True and row3 == True:

key = 8

elif col3 == True and row3 == True:

key = 9

elif col1 == True and row4 == True:

key = 10

elif col2 == True and row4 == True:

key = 11

elif col3 == True and row4 == True:

key = 12

elif but == True:

key = 13

if key <> -1:

url = 'http://192.168.1.11:8002/casa/index.php?action=dial&device=10&dial=%d' % key

try:

response = urllib2.urlopen(url, timeout=5)

except IOError:

iot.iot_send_string(999,'T65 Orange time out: ' + url)

GPIO.setmode(GPIO.BOARD)

pins = [8,10,12,16,18,22,24,26]

for pin in pins:

GPIO.setup(pin, GPIO.IN)

iot.iot_send_string(999,'T65 Orange booted')

while 1:

try:

main_loop()

except:

iot.iot_send_trace()

The script calls an URL on the home server: another raspberry pi which controls all connected devices. A description of this home server will be the subject for a future picnic.

October 4, 2018October 4, 2018

Control Sonos with Python using SoCo 0.9

I made a python wrapper function around SoCo 0.9 because the documentation of SoCo is very limited. I hope it helps 🙂

from soco import SoCo # used SoCo version 0.9
import sys
import random
import datetime
import subprocess
import Queue
import threading
import iot
import unicodedata
from time import sleep

html_escape_table = {
    "&": "&amp;",
    '"': "&quot;",
    "'": "&apos;",
    ">": "&gt;",
    "<": "&lt;",
    "+": "&#43;",
    ",": "&#44;",
    }

def escape(text):
    return "".join(html_escape_table.get(c,c) for c in text)

def remove_accents(input_str):
   nkfd_form = unicodedata.normalize('NFKD', unicode(input_str))
   return u"".join([c for c in nkfd_form if not unicodedata.combining(c)])

class Sonos:
   def __init__(self, device):
      self.device = device
      self.path = 'x-file-cifs://192.168.1.3/music/'
      ip = '192.168.1.66'
      self.debug = False
      self.mysoco = SoCo(ip)
      self.q_song = Queue.Queue()
      self.t_song = threading.Thread(target=self.__worker_insert_song, args=(self.q_song, ))
      self.t_song.daemon = True
      self.t_song.start()

   def __iot_send_string(self, event):
      if self.device <> -1:
         iot.iot_send_string(self.device, event)
      
   def __play_song_and_get_info(self, path_and_song, track_info):
      duration = 0
      transport_info = self.mysoco.get_current_transport_info()
      if track_info == {}:
         if transport_info['current_transport_state'] == 'PLAYING':
            track_info = self.mysoco.get_current_track_info()
            if self.debug:
               for x in track_info:
                  print x + track_info[x]
         else:
            track_info['empty'] = 1;
      self.mysoco.play_uri(path_and_song)
      if 'Bowie' in path_and_song:
         self.mysoco.seek('00:00:30')
      if transport_info['current_transport_state'] == 'PLAYING':
         y = self.mysoco.get_current_track_info()
         if 'duration' in y:
            d = y['duration'].split(':')
            duration = int(d[0])*3600 + int(d[1])*60 + int(d[2])
      return(track_info, duration)

   def __resume_previous_song(self, track_info):
      if 'mp3radio' in track_info['uri'] or 'aac://' in track_info['uri']:
         self.mysoco.play_uri(track_info['uri'])
      elif 'playlist_position' not in track_info:
         self.mysoco.play_uri(track_info['uri'])
      else:
         self.mysoco.play_from_queue(int(track_info['playlist_position'])-1)
         if 'position' in track_info:
            self.mysoco.seek(track_info['position'])

   def __worker_insert_song(self, q_song):
      track_info = {}
      duration = 0
      while 1:
         if self.q_song.qsize() > 0:
            path_and_song = q_song.get()
            [track_info, duration] = self.__play_song_and_get_info(path_and_song, track_info)
            t0 = datetime.datetime.now()
         elif duration <> 0:
            if (datetime.datetime.now() - t0).total_seconds() > duration:
               if track_info <> [] and 'empty' not in track_info:
                  self.__resume_previous_song(track_info)
               track_info = {}
               duration = 0
         sleep(1)

   def play_song(self, song):
      self.mysoco.play_uri(self.path + song)
      self.__iot_send_string(self.path + song)

   def play_spotify_song(self, song):
      self.mysoco.play_uri(song)
      self.__iot_send_string(song)

   def insert_song(self, song):
      self.q_song.put(self.path + song)

   def play_genre(self, genre):
      l = self.mysoco.get_music_library_information('genres')
      for i in l:
         if i.title == genre:
            self.mysoco.clear_queue()
            self.mysoco.add_uri_to_queue(i.uri)
            j = len(self.mysoco.get_queue())
            self.mysoco.play_from_queue(int(j*random.random()))
            self.mysoco.play_mode = 'SHUFFLE'
            self.__iot_send_string('Genre "'+genre+'" started')
      sleep(0.1)

   def play_playlist(self, playlist):
      l = self.mysoco.get_sonos_playlists()
      for i in l:
         if i.title == playlist:
            self.mysoco.clear_queue()
            self.mysoco.add_uri_to_queue(i.uri)
            j = len(self.mysoco.get_queue())
            self.mysoco.play_from_queue(int(j*random.random()))
            self.mysoco.play_mode = 'SHUFFLE'
            self.__iot_send_string('Playlist "'+playlist+'" started')
      sleep(0.1)

   def play_radio(self, radio_station):
      l = self.mysoco.get_favorite_radio_stations()
      self.mysoco.clear_queue()
      j = 0
      for i in l['favorites']:
         self.mysoco.add_uri_to_queue(i['uri'])
         if i['title'] == radio_station:
            self.mysoco.play_from_queue(j)
            self.__iot_send_string('Radio station "'+radio_station+'" started')
         j += 1
      sleep(0.1)

   def say(self, text):
      filename = 'sonos_say.wav'
      if sys.platform == 'win32':
         subprocess.call(['./espeak','-v','nl', '-w', "w:\\"+filename, text])
      else:
         subprocess.call(["espeak -v nl -w music/"+filename, text])
      self.play_song(filename)
      sleep(0.1)

   def toggle_play(self):
      x = self.mysoco.get_current_transport_info()
      if x['current_transport_state'] == 'PLAYING':
         self.mysoco.pause()
      else:
         self.mysoco.play()
      sleep(0.1)

   def change_volume(self, value):
      self.mysoco.volume += value
      sleep(0.25)

   def previous(self):
      self.mysoco.previous()
      sleep(0.1)


   def next(self):
      self.mysoco.next()
      sleep(0.1)

   def get_current_track_info(self):
      return self.mysoco.get_current_track_info()

   def get_current_transport_info(self):
      return self.mysoco.get_current_transport_info()

   def get_current_song_string(self):
      song = ''
      transport_info = self.get_current_transport_info()
      if transport_info['current_transport_state'] == 'PLAYING':
         x = self.get_current_track_info()
         song_uri = x['uri']
         if 'mp3radio://' in x['uri'] or 'aac://' in x['uri']:
            radio = ' "radio station" '
         else:
            radio = ''
         song = 'Now playing' + radio
         flag = 0
         if len(x['title']) > 0:
            song = ' "' + x['title'] + '"'
            flag = 1
         if len(x['artist']) > 0: 
            song = song + ' by "' + x['artist'] + '"'
         if len(x['album']) > 0:
            song = song + ' from album "' + x['album'] + '"'
         if flag == 0 and len(x['uri']) > 0:
            s = song_uri.split('://')
            song = song + ' "' + s[1] + '"'
            flag = 1
         song = remove_accents(song)
      return(song)
      
   def get_current_song_artist_album(self):
      song = ''
      artist = ''
      album = ''
      uri = ''
      transport_info = self.get_current_transport_info()
      if transport_info['current_transport_state'] == 'PLAYING':
         x = self.get_current_track_info()
         if len(x['title']) > 0:
            song = x['title']
            song = remove_accents(song)
            song = escape(song)
         if len(x['artist']) > 0: 
            artist = x['artist']
            artist = remove_accents(artist)
            artist = escape(artist)
         if len(x['album']) > 0:
            album = x['album']
            album = remove_accents(album)
            album = escape(album)
         if len(x['uri']) > 0:
            uri = x['uri']
            uri = remove_accents(uri)
            uri = escape(uri)
      return(song,artist,album,uri)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

from soco import SoCo # used SoCo version 0.9

import sys

import random

import datetime

import subprocess

import Queue

import threading

import iot

import unicodedata

from time import sleep

html_escape_table = {

"&": "&",

'"': """,

"'": "'",

">": ">",

"<": "<",

"+": "+",

",": ",",

}

def escape(text):

return "".join(html_escape_table.get(c,c) for c in text)

def remove_accents(input_str):

nkfd_form = unicodedata.normalize('NFKD', unicode(input_str))

return u"".join([c for c in nkfd_form if not unicodedata.combining(c)])

class Sonos:

def __init__(self, device):

self.device = device

self.path = 'x-file-cifs://192.168.1.3/music/'

ip = '192.168.1.66'

self.debug = False

self.mysoco = SoCo(ip)

self.q_song = Queue.Queue()

self.t_song = threading.Thread(target=self.__worker_insert_song, args=(self.q_song, ))

self.t_song.daemon = True

self.t_song.start()

def __iot_send_string(self, event):

if self.device <> -1:

iot.iot_send_string(self.device, event)

def __play_song_and_get_info(self, path_and_song, track_info):

duration = 0

transport_info = self.mysoco.get_current_transport_info()

if track_info == {}:

if transport_info['current_transport_state'] == 'PLAYING':

track_info = self.mysoco.get_current_track_info()

if self.debug:

for x in track_info:

print x + track_info[x]

else:

track_info['empty'] = 1;

self.mysoco.play_uri(path_and_song)

if 'Bowie' in path_and_song:

self.mysoco.seek('00:00:30')

if transport_info['current_transport_state'] == 'PLAYING':

y = self.mysoco.get_current_track_info()

if 'duration' in y:

d = y['duration'].split(':')

duration = int(d[0])*3600 + int(d[1])*60 + int(d[2])

return(track_info, duration)

def __resume_previous_song(self, track_info):

if 'mp3radio' in track_info['uri'] or 'aac://' in track_info['uri']:

self.mysoco.play_uri(track_info['uri'])

elif 'playlist_position' not in track_info:

self.mysoco.play_uri(track_info['uri'])

else:

self.mysoco.play_from_queue(int(track_info['playlist_position'])-1)

if 'position' in track_info:

self.mysoco.seek(track_info['position'])

def __worker_insert_song(self, q_song):

track_info = {}

duration = 0

while 1:

if self.q_song.qsize() > 0:

path_and_song = q_song.get()

[track_info, duration] = self.__play_song_and_get_info(path_and_song, track_info)

t0 = datetime.datetime.now()

elif duration <> 0:

if (datetime.datetime.now() - t0).total_seconds() > duration:

if track_info <> [] and 'empty' not in track_info:

self.__resume_previous_song(track_info)

track_info = {}

duration = 0

sleep(1)

def play_song(self, song):

self.mysoco.play_uri(self.path + song)

self.__iot_send_string(self.path + song)

def play_spotify_song(self, song):

self.mysoco.play_uri(song)

self.__iot_send_string(song)

def insert_song(self, song):

self.q_song.put(self.path + song)

def play_genre(self, genre):

l = self.mysoco.get_music_library_information('genres')

for i in l:

if i.title == genre:

self.mysoco.clear_queue()

self.mysoco.add_uri_to_queue(i.uri)

j = len(self.mysoco.get_queue())

self.mysoco.play_from_queue(int(j*random.random()))

self.mysoco.play_mode = 'SHUFFLE'

self.__iot_send_string('Genre "'+genre+'" started')

sleep(0.1)

def play_playlist(self, playlist):

l = self.mysoco.get_sonos_playlists()

for i in l:

if i.title == playlist:

self.mysoco.clear_queue()

self.mysoco.add_uri_to_queue(i.uri)

j = len(self.mysoco.get_queue())

self.mysoco.play_from_queue(int(j*random.random()))

self.mysoco.play_mode = 'SHUFFLE'

self.__iot_send_string('Playlist "'+playlist+'" started')

sleep(0.1)

def play_radio(self, radio_station):

l = self.mysoco.get_favorite_radio_stations()

self.mysoco.clear_queue()

j = 0

for i in l['favorites']:

self.mysoco.add_uri_to_queue(i['uri'])

if i['title'] == radio_station:

self.mysoco.play_from_queue(j)

self.__iot_send_string('Radio station "'+radio_station+'" started')

j += 1

sleep(0.1)

def say(self, text):

filename = 'sonos_say.wav'

if sys.platform == 'win32':

subprocess.call(['./espeak','-v','nl', '-w', "w:\\"+filename, text])

else:

subprocess.call(["espeak -v nl -w music/"+filename, text])

self.play_song(filename)

sleep(0.1)

def toggle_play(self):

x = self.mysoco.get_current_transport_info()

if x['current_transport_state'] == 'PLAYING':

self.mysoco.pause()

else:

self.mysoco.play()

sleep(0.1)

def change_volume(self, value):

self.mysoco.volume += value

sleep(0.25)

def previous(self):

self.mysoco.previous()

sleep(0.1)

def next(self):

self.mysoco.next()

sleep(0.1)

def get_current_track_info(self):

return self.mysoco.get_current_track_info()

def get_current_transport_info(self):

return self.mysoco.get_current_transport_info()

def get_current_song_string(self):

song = ''

transport_info = self.get_current_transport_info()

if transport_info['current_transport_state'] == 'PLAYING':

x = self.get_current_track_info()

song_uri = x['uri']

if 'mp3radio://' in x['uri'] or 'aac://' in x['uri']:

radio = ' "radio station" '

else:

radio = ''

song = 'Now playing' + radio

flag = 0

if len(x['title']) > 0:

song = ' "' + x['title'] + '"'

flag = 1

if len(x['artist']) > 0:

song = song + ' by "' + x['artist'] + '"'

if len(x['album']) > 0:

song = song + ' from album "' + x['album'] + '"'

if flag == 0 and len(x['uri']) > 0:

s = song_uri.split('://')

song = song + ' "' + s[1] + '"'

flag = 1

song = remove_accents(song)

return(song)

def get_current_song_artist_album(self):

song = ''

artist = ''

album = ''

uri = ''

transport_info = self.get_current_transport_info()

if transport_info['current_transport_state'] == 'PLAYING':

x = self.get_current_track_info()

if len(x['title']) > 0:

song = x['title']

song = remove_accents(song)

song = escape(song)

if len(x['artist']) > 0:

artist = x['artist']

artist = remove_accents(artist)

artist = escape(artist)

if len(x['album']) > 0:

album = x['album']

album = remove_accents(album)

album = escape(album)

if len(x['uri']) > 0:

uri = x['uri']

uri = remove_accents(uri)

uri = escape(uri)

return(song,artist,album,uri)

iot is a homebrew logger

import urllib, urllib2, os, traceback, time

def __iot_send(device_id, event = None):
   if os.name == 'nt':
      event = urllib.unquote(event).decode('utf8')
      print "%d: %s" % (device_id, event)
   else:
      try:
         url = 'https://www.picnicprojects.com/casa/log.php?d=' + str(device_id)
         if event <> None:
            url += '&v=' + event
         response = urllib2.urlopen(url)
      except:
         pass

def iot_send(device_id):
   __iot_send(device_id)

def iot_send_string(device_id, string):
   event = urllib.quote(string)
   __iot_send(device_id, event)

def iot_send_value(device_id, value):
   event = '%f'% value 
   __iot_send(device_id, event)

def iot_send_values(device_id, values):
   first = 1
   print values
   for v in values:
      print v
      if first == 1:
         event = '%f' % v
         first = 0
      else:
         event += ',%f' % v
   __iot_send(device_id, event)
   
def iot_send_trace():
   __iot_send(999, 'exception: ' + traceback.format_exc())

import urllib, urllib2, os, traceback, time

def __iot_send(device_id, event = None):

if os.name == 'nt':

event = urllib.unquote(event).decode('utf8')

print "%d: %s" % (device_id, event)

else:

try:

url = 'https://www.picnicprojects.com/casa/log.php?d=' + str(device_id)

if event <> None:

url += '&v=' + event

response = urllib2.urlopen(url)

except:

pass

def iot_send(device_id):

__iot_send(device_id)

def iot_send_string(device_id, string):

event = urllib.quote(string)

__iot_send(device_id, event)

def iot_send_value(device_id, value):

event = '%f'% value

__iot_send(device_id, event)

def iot_send_values(device_id, values):

first = 1

print values

for v in values:

print v

if first == 1:

event = '%f' % v

first = 0

else:

event += ',%f' % v

__iot_send(device_id, event)

def iot_send_trace():

__iot_send(999, 'exception: ' + traceback.format_exc())

October 2, 2018October 4, 2018

Python sunset timer for Philips Hue

The Philips Hue has a nice API. It took me a few picnics to figure out how it works. So for your convenience I give you a few snippets of Python code.

First you have to figure out what your username is:

response = requests.post(self.ip + '/api', data=json.dumps({"devicetype": "casa"}))
print response.content

1 2	response = requests.post(self.ip + '/api', data=json.dumps({"devicetype": "casa"})) print response.content

With this username you can check lights or turn them on and off:

r = requests.get('http://192.168.1.194/api/qKsJorcN2lYr-bFfmzyFNqrCe5sEV/lights/3')
print r.content

1 2	r = requests.get('http://192.168.1.194/api/qKsJorcN2lYr-bFfmzyFNqrCe5sEV/lights/3') print r.content

And now a wrapper function:

import requests
import time
import json
import math
import iot

def EnhanceColor(normalized):
    if normalized > 0.04045:
        return math.pow( (normalized + 0.055) / (1.0 + 0.055), 2.4)
    else:
        return normalized / 12.92

def xy(r, g, b):
    rNorm = r / 255.0
    gNorm = g / 255.0
    bNorm = b / 255.0

    rFinal = EnhanceColor(rNorm)
    gFinal = EnhanceColor(gNorm)
    bFinal = EnhanceColor(bNorm)
    
    X = rFinal * 0.649926 + gFinal * 0.103455 + bFinal * 0.197109
    Y = rFinal * 0.234327 + gFinal * 0.743075 + bFinal * 0.022598
    Z = rFinal * 0.000000 + gFinal * 0.053077 + bFinal * 1.035763

    if X + Y + Z == 0:
        return (0,0)
    else:
        xFinal = X / (X + Y + Z)
        yFinal = Y / (X + Y + Z)
        return (xFinal, yFinal)

class Bridge():
   def __init__(self):
      self.ip = '192.168.1.194'
      self.username = "qKsJorcBTlYr-bFfma30NqrCeprG5sEV"

   def put(self, action, d):
      s = 'http://' + self.ip + "/api/" + self.username + action
      print s
      r = requests.put(s,data=json.dumps(d))
      print r.content
      
   def kitchen_off(self):
      self.put('/groups/2/action/',{'on':False})
      iot.iot_send_string(50, 'Kitchen Off')

   def kitchen_night_light(self):
      self.put('/groups/2/action/',{'on':True, 'bri':1})
      iot.iot_send_string(50, 'Kitchen Night Light')
      
   def kitchen_dimmed(self):
      self.put('/groups/2/action/',{'on':True, 'bri':128})
      iot.iot_send_string(50, 'Kitchen Dimmed')

   def kitchen_bright(self):
      self.put('/groups/2/action/',{'on':True, 'bri':254})
      iot.iot_send_string(50, 'Kitchen Bright')

   def kitchen_blink(self):
      for i in range(2):
         self.put('/groups/2/action/',{'on':True, 'bri':254})
         time.sleep(1)
         self.put('/groups/2/action/',{'on':False})
         time.sleep(1)
      iot.iot_send_string(50, 'Kitchen Blink')

   def living_room_off(self):
      self.put('/groups/1/action/',{'on':False})
      iot.iot_send_string(50, 'Living Room Off')

   def living_room_night_light(self):
      self.put('/groups/1/action/',{'on':True, 'bri':1})
      iot.iot_send_string(50, 'Living Room Night Light')
      
   def living_room_dimmed(self):
      self.put('/groups/1/action/',{'on':True, 'bri':128})
      iot.iot_send_string(50, 'Living Room Dimmed')

   def living_room_bright(self):
      self.put('/groups/1/action/',{'on':True, 'bri':254})
      iot.iot_send_string(50, 'Living Room Bright')

   def bloom_color_loop(self):
      self.put('/lights/8/state',{'on':True, 'bri':254, 'effect':'colorloop'})
      iot.iot_send_string(50, 'Hue Bloom Color Loop')

   def bloom_set_rgb(self, r,g,b):
      self.put('/lights/8/state',{'on':True, 'bri':254, 'xy':xy(r,g,b)})
      iot.iot_send_string(50, 'Hue Bloom red=%d, green=%d, blue=%d' % (r,g,b))

import requests

import time

import json

import math

import iot

def EnhanceColor(normalized):

if normalized > 0.04045:

return math.pow( (normalized + 0.055) / (1.0 + 0.055), 2.4)

else:

return normalized / 12.92

def xy(r, g, b):

rNorm = r / 255.0

gNorm = g / 255.0

bNorm = b / 255.0

rFinal = EnhanceColor(rNorm)

gFinal = EnhanceColor(gNorm)

bFinal = EnhanceColor(bNorm)

X = rFinal * 0.649926 + gFinal * 0.103455 + bFinal * 0.197109

Y = rFinal * 0.234327 + gFinal * 0.743075 + bFinal * 0.022598

Z = rFinal * 0.000000 + gFinal * 0.053077 + bFinal * 1.035763

if X + Y + Z == 0:

return (0,0)

else:

xFinal = X / (X + Y + Z)

yFinal = Y / (X + Y + Z)

return (xFinal, yFinal)

class Bridge():

def __init__(self):

self.ip = '192.168.1.194'

self.username = "qKsJorcBTlYr-bFfma30NqrCeprG5sEV"

def put(self, action, d):

s = 'http://' + self.ip + "/api/" + self.username + action

print s

r = requests.put(s,data=json.dumps(d))

print r.content

def kitchen_off(self):

self.put('/groups/2/action/',{'on':False})

iot.iot_send_string(50, 'Kitchen Off')

def kitchen_night_light(self):

self.put('/groups/2/action/',{'on':True, 'bri':1})

iot.iot_send_string(50, 'Kitchen Night Light')

def kitchen_dimmed(self):

self.put('/groups/2/action/',{'on':True, 'bri':128})

iot.iot_send_string(50, 'Kitchen Dimmed')

def kitchen_bright(self):

self.put('/groups/2/action/',{'on':True, 'bri':254})

iot.iot_send_string(50, 'Kitchen Bright')

def kitchen_blink(self):

for i in range(2):

self.put('/groups/2/action/',{'on':True, 'bri':254})

time.sleep(1)

self.put('/groups/2/action/',{'on':False})

time.sleep(1)

iot.iot_send_string(50, 'Kitchen Blink')

def living_room_off(self):

self.put('/groups/1/action/',{'on':False})

iot.iot_send_string(50, 'Living Room Off')

def living_room_night_light(self):

self.put('/groups/1/action/',{'on':True, 'bri':1})

iot.iot_send_string(50, 'Living Room Night Light')

def living_room_dimmed(self):

self.put('/groups/1/action/',{'on':True, 'bri':128})

iot.iot_send_string(50, 'Living Room Dimmed')

def living_room_bright(self):

self.put('/groups/1/action/',{'on':True, 'bri':254})

iot.iot_send_string(50, 'Living Room Bright')

def bloom_color_loop(self):

self.put('/lights/8/state',{'on':True, 'bri':254, 'effect':'colorloop'})

iot.iot_send_string(50, 'Hue Bloom Color Loop')

def bloom_set_rgb(self, r,g,b):

self.put('/lights/8/state',{'on':True, 'bri':254, 'xy':xy(r,g,b)})

iot.iot_send_string(50, 'Hue Bloom red=%d, green=%d, blue=%d' % (r,g,b))

The iot class is a homebrew event logger:

import urllib, urllib2, os, traceback, time

def __iot_send(device_id, event = None):
   if os.name == 'nt':
      event = urllib.unquote(event).decode('utf8')
      print "%d: %s" % (device_id, event)
   else:
      try:
         url = 'https://www.picnicprojects.com/casa/log.php?d=' + str(device_id)
         if event <> None:
            url += '&v=' + event
         response = urllib2.urlopen(url)
      except:
         pass

def iot_send(device_id):
   __iot_send(device_id)

def iot_send_string(device_id, string):
   event = urllib.quote(string)
   __iot_send(device_id, event)

def iot_send_value(device_id, value):
   event = '%f'% value 
   __iot_send(device_id, event)

def iot_send_values(device_id, values):
   first = 1
   print values
   for v in values:
      print v
      if first == 1:
         event = '%f' % v
         first = 0
      else:
         event += ',%f' % v
   __iot_send(device_id, event)
   
def iot_send_trace():
   __iot_send(999, 'exception: ' + traceback.format_exc())

import urllib, urllib2, os, traceback, time

def __iot_send(device_id, event = None):

if os.name == 'nt':

event = urllib.unquote(event).decode('utf8')

print "%d: %s" % (device_id, event)

else:

try:

url = 'https://www.picnicprojects.com/casa/log.php?d=' + str(device_id)

if event <> None:

url += '&v=' + event

response = urllib2.urlopen(url)

except:

pass

def iot_send(device_id):

__iot_send(device_id)

def iot_send_string(device_id, string):

event = urllib.quote(string)

__iot_send(device_id, event)

def iot_send_value(device_id, value):

event = '%f'% value

__iot_send(device_id, event)

def iot_send_values(device_id, values):

first = 1

print values

for v in values:

print v

if first == 1:

event = '%f' % v

first = 0

else:

event += ',%f' % v

__iot_send(device_id, event)

def iot_send_trace():

__iot_send(999, 'exception: ' + traceback.format_exc())

You can use crontab to set timers. But why not make it yourself 🙂

import threading, datetime, time, os, traceback
import iot, sun, hue

def hue_set_lights(action, group):
   try:
      iot.iot_send_string(200, 'Hue: %s %s' % (group, action))
      if action == 'on':
         if group == 'kitchen':
            bridge.kitchen_night_light()
            bridge.bloom_set_rgb(255,165,0) #orange
         elif group == 'livingroom':
            bridge.living_room_night_light()
      else:
         if group == 'kitchen':
            bridge.kitchen_off()
         elif group == 'livingroom':
            bridge.living_room_off()
   except:
      iot.iot_send_trace()

def seconds_to_string(dt):
   h = int(dt/3600)
   m = (dt - h*3600) / 60
   s = "%d s = %02d:%05.2f" % (dt, h, m)
   return s
   
def seconds_until_end_of_day(now):
   dt = ((24 - now.hour - 1) * 60 * 60) + ((60 - now.minute - 1) * 60) + (60 - now.second)
   return dt

def seconds_until_time(now, h, m, delta):
   t2 = now.replace(hour = h, minute = m, second = 0, microsecond = 0) + datetime.timedelta(hours=delta)
   dt = t2 - now
   dt = dt.total_seconds()
   return(dt)

def seconds_until_sunset(now):
   t  = sun.get_local_sunset_time(now.date())
   t  = t.replace(tzinfo=None)
   dt = datetime.datetime.combine(datetime.date.today(), t.time()) - datetime.datetime.combine(datetime.date.today(), now.time())
   dt = dt.total_seconds()
   return(dt)

def seconds_until_sunrise(now):
   t  = sun.get_local_sunrise_time(now.date())
   t  = t.replace(tzinfo=None)
   dt = datetime.datetime.combine(datetime.date.today(), t.time()) - datetime.datetime.combine(datetime.date.today(), now.time())
   dt = dt.total_seconds()
   return(dt)

def calculate_seconds_until_event(now, days, h, m, duration, use_sun):
   # ignore days
   dt_on  = -1
   dt_off = -1
   d = now.weekday()
   if ((str(d+1) in str(days)) or ('*' in str(days))):
      dt_on  = seconds_until_time(now, h, m, 0)
      dt_off = seconds_until_time(now, h, m, duration)
      if use_sun:
         if h < 12: 
            dt_sun = seconds_until_sunrise(now)
            if dt_sun < dt_on:
               dt_on  = -1
               dt_off = -1
            elif dt_sun < dt_off:
               dt_off = dt_sun
         else:
            dt_sun = seconds_until_sunset(now) - 2700
            if dt_sun > dt_off:
               dt_on  = -1
               dt_off = -1
            elif dt_sun > dt_on:
               dt_on = dt_sun
   return(dt_on, dt_off)

   
def log_dt(now, dt, action, group):
   if action == 1:
      a = 'on'
   else:
      a = 'off'
   s = "timer programmed at: %s %s %s (%s)"  % ((now + datetime.timedelta(seconds = dt)), group, a, seconds_to_string(dt))

def schedule_timers(days, hour, minute, duration, use_sun, group):
   now = datetime.datetime.now()
   dt_on, dt_off = calculate_seconds_until_event(now, days, hour, minute, duration, use_sun)
   if dt_on > 0:
      log_dt(now, dt_on, 1, group)
      t = threading.Timer(dt_on, hue_set_lights, args=('on',group))
      t.setDaemon(True)
      t.start()
   if dt_off > 0:
      log_dt(now, dt_off, 0, group)
      t = threading.Timer(dt_off, hue_set_lights, args=('off',group))
      t.setDaemon(True)
      t.start()
      
bridge = hue.Bridge()
sun = sun.Sun()

if __name__ == '__main__':
   while 1:
      try:
         if os.name == 'nt':
            fname = 'timers.txt'
         else:
            fname = '/mnt/nas/data/timers.txt'
         fid = open(fname, 'r')
         for line in fid:
            if '#' not in line:
               a = line.replace('\n','').split(',')
               # days, hour, minute, duration, use_sun, group
               schedule_timers(a[0], int(a[1]), int(a[2]), float(a[3]), int(a[4]), a[5])
               # time.sleep(10)
      except:
         iot.iot_send_trace()
      dt = 60 + seconds_until_end_of_day(datetime.datetime.now())
      time.sleep(dt)

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

import threading, datetime, time, os, traceback

import iot, sun, hue

def hue_set_lights(action, group):

try:

iot.iot_send_string(200, 'Hue: %s %s' % (group, action))

if action == 'on':

if group == 'kitchen':

bridge.kitchen_night_light()

bridge.bloom_set_rgb(255,165,0) #orange

elif group == 'livingroom':

bridge.living_room_night_light()

else:

if group == 'kitchen':

bridge.kitchen_off()

elif group == 'livingroom':

bridge.living_room_off()

except:

iot.iot_send_trace()

def seconds_to_string(dt):

h = int(dt/3600)

m = (dt - h*3600) / 60

s = "%d s = %02d:%05.2f" % (dt, h, m)

return s

def seconds_until_end_of_day(now):

dt = ((24 - now.hour - 1) * 60 * 60) + ((60 - now.minute - 1) * 60) + (60 - now.second)

return dt

def seconds_until_time(now, h, m, delta):

t2 = now.replace(hour = h, minute = m, second = 0, microsecond = 0) + datetime.timedelta(hours=delta)

dt = t2 - now

dt = dt.total_seconds()

return(dt)

def seconds_until_sunset(now):

t = sun.get_local_sunset_time(now.date())

t = t.replace(tzinfo=None)

dt = datetime.datetime.combine(datetime.date.today(), t.time()) - datetime.datetime.combine(datetime.date.today(), now.time())

dt = dt.total_seconds()

return(dt)

def seconds_until_sunrise(now):

t = sun.get_local_sunrise_time(now.date())

t = t.replace(tzinfo=None)

dt = datetime.datetime.combine(datetime.date.today(), t.time()) - datetime.datetime.combine(datetime.date.today(), now.time())

dt = dt.total_seconds()

return(dt)

def calculate_seconds_until_event(now, days, h, m, duration, use_sun):

# ignore days

dt_on = -1

dt_off = -1

d = now.weekday()

if ((str(d+1) in str(days)) or ('*' in str(days))):

dt_on = seconds_until_time(now, h, m, 0)

dt_off = seconds_until_time(now, h, m, duration)

if use_sun:

if h < 12:

dt_sun = seconds_until_sunrise(now)

if dt_sun < dt_on:

dt_on = -1

dt_off = -1

elif dt_sun < dt_off:

dt_off = dt_sun

else:

dt_sun = seconds_until_sunset(now) - 2700

if dt_sun > dt_off:

dt_on = -1

dt_off = -1

elif dt_sun > dt_on:

dt_on = dt_sun

return(dt_on, dt_off)

def log_dt(now, dt, action, group):

if action == 1:

a = 'on'

else:

a = 'off'

s = "timer programmed at: %s %s %s (%s)" % ((now + datetime.timedelta(seconds = dt)), group, a, seconds_to_string(dt))

def schedule_timers(days, hour, minute, duration, use_sun, group):

now = datetime.datetime.now()

dt_on, dt_off = calculate_seconds_until_event(now, days, hour, minute, duration, use_sun)

if dt_on > 0:

log_dt(now, dt_on, 1, group)

t = threading.Timer(dt_on, hue_set_lights, args=('on',group))

t.setDaemon(True)

t.start()

if dt_off > 0:

log_dt(now, dt_off, 0, group)

t = threading.Timer(dt_off, hue_set_lights, args=('off',group))

t.setDaemon(True)

t.start()

bridge = hue.Bridge()

sun = sun.Sun()

if __name__ == '__main__':

while 1:

try:

if os.name == 'nt':

fname = 'timers.txt'

else:

fname = '/mnt/nas/data/timers.txt'

fid = open(fname, 'r')

for line in fid:

if '#' not in line:

a = line.replace('\n','').split(',')

# days, hour, minute, duration, use_sun, group

schedule_timers(a[0], int(a[1]), int(a[2]), float(a[3]), int(a[4]), a[5])

# time.sleep(10)

except:

iot.iot_send_trace()

dt = 60 + seconds_until_end_of_day(datetime.datetime.now())

time.sleep(dt)

You have to put all timers in timers.txt
Yes, I know, it looks a lot like crontab. This one is not better, but I made it myself 🙂

# days, hour, minute, duration, use_sun, group
1245,  5,30,  1,1,kitchen
12345, 7, 0,1.5,1,kitchen
67,    7, 5,3,  1,kitchen
*,    17,30,  6,1,kitchen
*,    18,15,  4,1,livingroom

# days, hour, minute, duration, use_sun, group

1245, 5,30, 1,1,kitchen

12345, 7, 0,1.5,1,kitchen

67, 7, 5,3, 1,kitchen

*, 17,30, 6,1,kitchen

*, 18,15, 4,1,livingroom

And you need to know when the sun rises and sets. I didn’t write this class myself but I also forgot where I found it. Credits to the anonymous maker!

import math
import datetime
from dateutil import tz

class SunTimeException(Exception):

    def __init__(self, message):
        super(SunTimeException, self).__init__(message)


class Sun:
    """
    Approximated calculation of sunrise and sunset datetimes. Adapted from:
    https://stackoverflow.com/questions/19615350/calculate-sunrise-and-sunset-times-for-a-given-gps-coordinate-within-postgresql
    """
    def __init__(self, lat=52.37,lon=4.90): # default Amsterdam
        self._lat = lat
        self._lon = lon

    def get_sunrise_time(self, date=datetime.date.today()):
        """
        Calculate the sunrise time for given date.
        :param lat: Latitude
        :param lon: Longitude
        :param date: Reference date
        :return: UTC sunrise datetime
        :raises: SunTimeException when there is no sunrise and sunset on given location and date
        """
        sr = self._calc_sun_time(date, True)
        if sr is None:
            raise SunTimeException('The sun never rises on this location (on the specified date)')
        else:
            return sr

    def get_local_sunrise_time(self, date=datetime.date.today(), local_time_zone=tz.tzlocal()):
        """
        Get sunrise time for local or custom time zone.
        :param date: Reference date
        :param local_time_zone: Local or custom time zone.
        :return: Local time zone sunrise datetime
        """
        sr = self._calc_sun_time(date, True)
        if sr is None:
            raise SunTimeException('The sun never rises on this location (on the specified date)')
        else:
            return sr.astimezone(local_time_zone)

    def get_sunset_time(self, date=datetime.date.today()):
        """
        Calculate the sunset time for given date.
        :param lat: Latitude
        :param lon: Longitude
        :param date: Reference date
        :return: UTC sunset datetime
        :raises: SunTimeException when there is no sunrise and sunset on given location and date.
        """
        ss = self._calc_sun_time(date, False)
        if ss is None:
            raise SunTimeException('The sun never sets on this location (on the specified date)')
        else:
            return ss

    def get_local_sunset_time(self, date=datetime.date.today(), local_time_zone=tz.tzlocal()):
        """
        Get sunset time for local or custom time zone.
        :param date: Reference date
        :param local_time_zone: Local or custom time zone.
        :return: Local time zone sunset datetime
        """
        ss = self._calc_sun_time(date, False)
        if ss is None:
            raise SunTimeException('The sun never sets on this location (on the specified date)')
        else:
            return ss.astimezone(local_time_zone)

    def _calc_sun_time(self, date=datetime.date.today(), isRiseTime=True, zenith=90.8):
        """
        Calculate sunrise or sunset date.
        :param date: Reference date
        :param isRiseTime: True if you want to calculate sunrise time.
        :param zenith: Sun reference zenith
        :return: UTC sunset or sunrise datetime
        :raises: SunTimeException when there is no sunrise and sunset on given location and date
        """
        # isRiseTime == False, returns sunsetTime
        day = date.day
        month = date.month
        year = date.year

        TO_RAD = math.pi/180.0

        # 1. first calculate the day of the year
        N1 = math.floor(275 * month / 9)
        N2 = math.floor((month + 9) / 12)
        N3 = (1 + math.floor((year - 4 * math.floor(year / 4) + 2) / 3))
        N = N1 - (N2 * N3) + day - 30

        # 2. convert the longitude to hour value and calculate an approximate time
        lngHour = self._lon / 15

        if isRiseTime:
            t = N + ((6 - lngHour) / 24)
        else: #sunset
            t = N + ((18 - lngHour) / 24)

        # 3. calculate the Sun's mean anomaly
        M = (0.9856 * t) - 3.289

        # 4. calculate the Sun's true longitude
        L = M + (1.916 * math.sin(TO_RAD*M)) + (0.020 * math.sin(TO_RAD * 2 * M)) + 282.634
        L = self._force_range(L, 360 ) #NOTE: L adjusted into the range [0,360)

        # 5a. calculate the Sun's right ascension

        RA = (1/TO_RAD) * math.atan(0.91764 * math.tan(TO_RAD*L))
        RA = self._force_range(RA, 360 ) #NOTE: RA adjusted into the range [0,360)

        # 5b. right ascension value needs to be in the same quadrant as L
        Lquadrant  = (math.floor( L/90)) * 90
        RAquadrant = (math.floor(RA/90)) * 90
        RA = RA + (Lquadrant - RAquadrant)

        # 5c. right ascension value needs to be converted into hours
        RA = RA / 15

        # 6. calculate the Sun's declination
        sinDec = 0.39782 * math.sin(TO_RAD*L)
        cosDec = math.cos(math.asin(sinDec))

        # 7a. calculate the Sun's local hour angle
        cosH = (math.cos(TO_RAD*zenith) - (sinDec * math.sin(TO_RAD*self._lat))) / (cosDec * math.cos(TO_RAD*self._lat))

        if cosH > 1:
            return None     # The sun never rises on this location (on the specified date)
        if cosH < -1:
            return None     # The sun never sets on this location (on the specified date)

        # 7b. finish calculating H and convert into hours

        if isRiseTime:
            H = 360 - (1/TO_RAD) * math.acos(cosH)
        else: #setting
            H = (1/TO_RAD) * math.acos(cosH)

        H = H / 15

        #8. calculate local mean time of rising/setting
        T = H + RA - (0.06571 * t) - 6.622

        #9. adjust back to UTC
        UT = T - lngHour
        UT = self._force_range(UT, 24)   # UTC time in decimal format (e.g. 23.23)

        #10. Return
        hr = self._force_range(int(UT), 24)
        min = round((UT - int(UT))*60, 0)
        if min == 60:
            hr += 1
            min = 0

        return datetime.datetime(year, month, day, hr, int(min), tzinfo=tz.tzutc())

    @staticmethod
    def _force_range(v, max):
        # force v to be >= 0 and < max
        if v < 0:
            return v + max
        elif v >= max:
            return v - max

        return v


if __name__ == '__main__':
    sun = Sun(85.0, 21.00)
    try:
        print(sun.get_local_sunrise_time())
        print(sun.get_local_sunset_time())

    # On a special date in UTC

        abd = datetime.date(2014, 1, 3)
        abd_sr = sun.get_local_sunrise_time(abd)
        abd_ss = sun.get_local_sunset_time(abd)
        print(abd_sr)
        print(abd_ss)
    except SunTimeException as e:
        print("Error: {0}".format(e))

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

import math

import datetime

from dateutil import tz

class SunTimeException(Exception):

def __init__(self, message):

super(SunTimeException, self).__init__(message)

class Sun:

"""

Approximated calculation of sunrise and sunset datetimes. Adapted from:

https://stackoverflow.com/questions/19615350/calculate-sunrise-and-sunset-times-for-a-given-gps-coordinate-within-postgresql

"""

def __init__(self, lat=52.37,lon=4.90): # default Amsterdam

self._lat = lat

self._lon = lon

def get_sunrise_time(self, date=datetime.date.today()):

"""

Calculate the sunrise time for given date.

:param lat: Latitude

:param lon: Longitude

:param date: Reference date

:return: UTC sunrise datetime

:raises: SunTimeException when there is no sunrise and sunset on given location and date

"""

sr = self._calc_sun_time(date, True)

if sr is None:

raise SunTimeException('The sun never rises on this location (on the specified date)')

else:

return sr

def get_local_sunrise_time(self, date=datetime.date.today(), local_time_zone=tz.tzlocal()):

"""

Get sunrise time for local or custom time zone.

:param date: Reference date

:param local_time_zone: Local or custom time zone.

:return: Local time zone sunrise datetime

"""

sr = self._calc_sun_time(date, True)

if sr is None:

raise SunTimeException('The sun never rises on this location (on the specified date)')

else:

return sr.astimezone(local_time_zone)

def get_sunset_time(self, date=datetime.date.today()):

"""

Calculate the sunset time for given date.

:param lat: Latitude

:param lon: Longitude

:param date: Reference date

:return: UTC sunset datetime

:raises: SunTimeException when there is no sunrise and sunset on given location and date.

"""

ss = self._calc_sun_time(date, False)

if ss is None:

raise SunTimeException('The sun never sets on this location (on the specified date)')

else:

return ss

def get_local_sunset_time(self, date=datetime.date.today(), local_time_zone=tz.tzlocal()):

"""

Get sunset time for local or custom time zone.

:param date: Reference date

:param local_time_zone: Local or custom time zone.

:return: Local time zone sunset datetime

"""

ss = self._calc_sun_time(date, False)

if ss is None:

raise SunTimeException('The sun never sets on this location (on the specified date)')

else:

return ss.astimezone(local_time_zone)

def _calc_sun_time(self, date=datetime.date.today(), isRiseTime=True, zenith=90.8):

"""

Calculate sunrise or sunset date.

:param date: Reference date

:param isRiseTime: True if you want to calculate sunrise time.

:param zenith: Sun reference zenith

:return: UTC sunset or sunrise datetime

:raises: SunTimeException when there is no sunrise and sunset on given location and date

"""

# isRiseTime == False, returns sunsetTime

day = date.day

month = date.month

year = date.year

TO_RAD = math.pi/180.0

# 1. first calculate the day of the year

N1 = math.floor(275 * month / 9)

N2 = math.floor((month + 9) / 12)

N3 = (1 + math.floor((year - 4 * math.floor(year / 4) + 2) / 3))

N = N1 - (N2 * N3) + day - 30

# 2. convert the longitude to hour value and calculate an approximate time

lngHour = self._lon / 15

if isRiseTime:

t = N + ((6 - lngHour) / 24)

else: #sunset

t = N + ((18 - lngHour) / 24)

# 3. calculate the Sun's mean anomaly

M = (0.9856 * t) - 3.289

# 4. calculate the Sun's true longitude

L = M + (1.916 * math.sin(TO_RAD*M)) + (0.020 * math.sin(TO_RAD * 2 * M)) + 282.634

L = self._force_range(L, 360 ) #NOTE: L adjusted into the range [0,360)

# 5a. calculate the Sun's right ascension

RA = (1/TO_RAD) * math.atan(0.91764 * math.tan(TO_RAD*L))

RA = self._force_range(RA, 360 ) #NOTE: RA adjusted into the range [0,360)

# 5b. right ascension value needs to be in the same quadrant as L

Lquadrant = (math.floor( L/90)) * 90

RAquadrant = (math.floor(RA/90)) * 90

RA = RA + (Lquadrant - RAquadrant)

# 5c. right ascension value needs to be converted into hours

RA = RA / 15

# 6. calculate the Sun's declination

sinDec = 0.39782 * math.sin(TO_RAD*L)

cosDec = math.cos(math.asin(sinDec))

# 7a. calculate the Sun's local hour angle

cosH = (math.cos(TO_RAD*zenith) - (sinDec * math.sin(TO_RAD*self._lat))) / (cosDec * math.cos(TO_RAD*self._lat))

if cosH > 1:

return None # The sun never rises on this location (on the specified date)

if cosH < -1:

return None # The sun never sets on this location (on the specified date)

# 7b. finish calculating H and convert into hours

if isRiseTime:

H = 360 - (1/TO_RAD) * math.acos(cosH)

else: #setting

H = (1/TO_RAD) * math.acos(cosH)

H = H / 15

#8. calculate local mean time of rising/setting

T = H + RA - (0.06571 * t) - 6.622

#9. adjust back to UTC

UT = T - lngHour

UT = self._force_range(UT, 24) # UTC time in decimal format (e.g. 23.23)

#10. Return

hr = self._force_range(int(UT), 24)

min = round((UT - int(UT))*60, 0)

if min == 60:

hr += 1

min = 0

return datetime.datetime(year, month, day, hr, int(min), tzinfo=tz.tzutc())

@staticmethod

def _force_range(v, max):

# force v to be >= 0 and < max

if v < 0:

return v + max

elif v >= max:

return v - max

return v

if __name__ == '__main__':

sun = Sun(85.0, 21.00)

try:

print(sun.get_local_sunrise_time())

print(sun.get_local_sunset_time())

# On a special date in UTC

abd = datetime.date(2014, 1, 3)

abd_sr = sun.get_local_sunrise_time(abd)

abd_ss = sun.get_local_sunset_time(abd)

print(abd_sr)

print(abd_ss)

except SunTimeException as e:

print("Error: {0}".format(e))