Merge branch 'master' of rvb.name:openhab-process
[openhab-process.git] / mqtt-pulse / mqtt_pulse.py
1 #!/usr/bin/python
2
3 from pulsectl import Pulse,PulseLoopStop
4
5 import threading
6 from time import sleep
7
8 import sys
9 from ConfigParser import ConfigParser
10 import paho.mqtt.client as paho
11
12 # ====================== PulseAudio-related part ========================
13
14 sink_name=None
15 muted=None
16 volume=None
17
18 callback_changed=None
19
20 paLock1=threading.RLock()
21 paLock2=threading.RLock()
22 stopFlag=False
23
24 debug=False
25
26 def Init():
27   global pulse
28   pulse = Pulse("mqtt-pa")
29
30 def Reconnect():
31   pulse.close()
32   Init()
33
34 def GetDefaultOut():
35   return pulse.server_info().default_sink_name
36
37 def GetDefaultSink():
38   sinkname=GetDefaultOut()
39   for sink in pulse.sink_list():
40     if sink.name==sinkname:
41       return sink
42
43 def GetDefaultVolume():
44   return pulse.volume_get_all_chans(GetDefaultSink())
45
46 def SetDefaultVolume(volume):
47   pulse.volume_set_all_chans(GetDefaultSink(), volume)
48
49 def IsDefaultMuted():
50   return GetDefaultSink().mute<>0
51   
52 def MuteDefault(mute = True):
53   return pulse.mute(GetDefaultSink(),mute)
54
55 def AcquirePALock():
56   tname=threading.current_thread().name
57   if debug:
58     print tname," aquiring action..."
59   paLock1.acquire()
60   if debug:
61     print tname," aquired action..."
62   pulse.event_listen_stop()
63   if debug:
64     print tname," event_listener stop command sent..."
65     print tname," aquiring loop..."
66   paLock2.acquire()
67   if debug:
68     print tname," aquired loop..."
69
70 def ReleasePALock():
71   tname=threading.current_thread().name
72   if debug:
73     print tname," releasing loop..."
74   paLock2.release()
75   if debug:
76     print tname," released loop..."
77     print tname," releasing action..."
78   paLock1.release()
79   if debug:
80     print tname," released action..."
81
82 def AquireLoopLock():
83   tname=threading.current_thread().name
84   if debug:
85     print tname," aquiring loop..."
86   paLock2.acquire()
87   if debug:
88     print tname," aquired loop..."
89
90 def ReleaseLoopLock():
91   tname=threading.current_thread().name
92   if debug:
93     print tname," releasing loop..."
94   paLock2.release()
95   if debug:
96     print tname," released loop..."
97
98 def EventListener(callback):
99   pulse.event_mask_set('all')
100   pulse.event_callback_set(callback)
101   AquireLoopLock()
102   try:
103     pulse.event_listen()
104   finally:
105     ReleaseLoopLock()
106
107 def EventProcess(ev):
108   raise PulseLoopStop
109
110 def StateProcess():
111   global sink_name,muted,volume
112   AcquirePALock()
113   try:
114     tname=threading.current_thread().name
115     current_sink=GetDefaultOut()
116     current_vol=round(GetDefaultVolume(),2)
117     current_muted="ON" if IsDefaultMuted() else "OFF"
118     if current_sink<>sink_name:
119       sink_name=current_sink
120       if debug:
121         print 'sink='+sink_name
122       if callback_changed:
123         if debug:
124           print 'callback...'
125         callback_changed('sink',sink_name)
126     if current_vol<>volume:
127       volume=current_vol
128       if debug:
129         print 'volume='+str(volume)
130       if callback_changed:
131         if debug:
132           print 'callback...'
133         callback_changed('volume',str(volume))
134     if current_muted<>muted:
135       muted=current_muted
136       if debug:
137         print 'muted='+muted
138       if callback_changed:
139         if debug:
140           print 'callback...'
141         callback_changed('muted',muted)
142   finally:
143     ReleasePALock()
144
145 def PAListener():
146   while not stopFlag:    
147     EventListener(EventProcess)
148     StateProcess()
149
150 def RunBackground(process):
151   stopFlag=False
152   thread = threading.Thread(target=process,name="Background")
153   thread.start()
154
155 def StopBackground():
156   global stopFlag
157   stopFlag=True
158   pulse.event_listen_stop()
159
160 def CommandGetDefaultOut():
161   AcquirePALock()
162   try:
163     result=GetDefaultOut()
164   finally:
165     ReleasePALock()
166   return result
167
168 def CommandGetDefaultVolume():
169   AcquirePALock()
170   try:
171     result=GetDefaultVolume()
172   finally:
173     ReleasePALock()
174   return result
175
176 def CommandIsDefaultMuted():
177   AcquirePALock()
178   try:
179     result=IsDefaultMuted()
180   finally:
181     ReleasePALock()
182   return result
183
184 def CommandSetDefaultVolume(volume):
185   AcquirePALock()
186   try:
187     SetDefaultVolume(volume)    
188   finally:
189     ReleasePALock()
190
191 def CommandMuteDefault(mute=True):
192   AcquirePALock()
193   try:
194     MuteDefault(mute)    
195   finally:
196     ReleasePALock()
197
198 # ====================== MQTT-related part ========================
199
200 lockMQTT = threading.RLock()
201
202 def on_message(mosq, obj, msg):
203   if debug:
204     print("Received: " + msg.topic + " " + str(msg.payload))  
205   try:
206     subtopic=msg.topic[len(mqtt_topic_in)+1:]
207     payload=msg.payload
208     if subtopic=="volume":
209       CommandSetDefaultVolume(float(payload))
210     elif subtopic=="muted":
211       if payload=="ON":
212         payload=True
213       elif payload=="OFF":
214         payload=False
215       else:
216         payload=int(payload)  
217       CommandMuteDefault(payload)
218     else:
219       if debug:
220         print "Unknown command"  
221   except:
222     if debug:
223       print "Command failed"    
224   
225 def InitMQTT():
226
227   global client,mqtt_topic_in,mqtt_topic_out
228
229   conffile = sys.argv[1]
230
231   config = ConfigParser()
232   config.add_section('mqtt')
233   # set defaults for anonymous auth
234   config.set('mqtt', 'username', '')
235   config.set('mqtt', 'password', '')
236   config.set('mqtt', 'port', '1883')
237   config.set('mqtt', 'in', 'pulse/in')
238   config.set('mqtt', 'out', 'pulse/out')
239   config.read(conffile)
240
241   mqtt_server = config.get('mqtt', 'server')
242   mqtt_port = config.getint('mqtt', 'port')
243   mqtt_username = config.get('mqtt', 'username')
244   mqtt_password = config.get('mqtt', 'password')
245   mqtt_topic_in = config.get('mqtt', 'in')
246   mqtt_topic_out = config.get('mqtt', 'out')
247
248   client = paho.Client('pulse')
249   client.username_pw_set(mqtt_username, mqtt_password)
250   client.on_message=on_message
251   client.connect(mqtt_server, port=mqtt_port)
252   client.subscribe(mqtt_topic_in+'/#',1)
253
254 def MQTTCallback(param,value):
255
256   lockMQTT.acquire()
257   try:
258     client.publish(mqtt_topic_out+'/'+param, payload=value)
259     if debug:
260       print "Sent "+param+"="+value
261   finally:
262     lockMQTT.release()
263
264 def StartPulseListener():
265   global callback_changed
266   Init()
267   InitMQTT()
268   callback_changed=MQTTCallback
269   StateProcess()
270   RunBackground(PAListener)
271
272 StartPulseListener()
273
274 try:
275   while True:
276     try:
277       client.loop()
278     except KeyboardInterrupt:
279       break
280 finally:
281   StopBackground()