There a few moving parts (described in the following posts).
- Daemon to proxy lirc port
- Run XBMC with custom lirc socket
- XBMC Lircmap.xml
- XBMC keymap.xml
This script is the guts of it. Save it somewhere and remember to make it executable. I have it in /usr/local/bin/lircmangled
Code:
#!/usr/bin/env python
#
# proxy the lirc socket to distinguish between short and long presses
# on apple remote buttons
#
# long press occurs when > n repeat events occur within t time of eachother.
# short press occurs when there is >= t time after an event.
#
# holding apple remote keys generates lirc events with increasing repeat counts.
# clicks on the apple remote keys can generate a few repeat lirc events, and
# if they are fast enough, can also have increasing repeat counts.
# this makes it a bit tricky to find a couple of short clicks followed by a
# long click as this looks like a continuous stream of repeat lirc events for
# the same button. can track this by monitoring the time beteen events and
# resequencing the repeat counts where appropriate
#
# Julian Neil
#
# this software is released under the Do It Yourself License.
# If it doesn't work for you, then fix it.
import logging
import asyncore
import asynchat
import socket
import os
import sys
import signal
import time
import threading
logging.basicConfig(level=logging.ERROR)
log = logging.getLogger(sys.argv[0])
hold_repeats = 3 # number of repeat lirc events for long press (3 or 4 seem ok)
repeat_threshold = 0.13 # time threshold for repeat events. (> 0.12 . might need more on a loaded machine)
proxy_socket = '/var/run/lirc/lircmangled' # socket for mangled output (and pass-through input)
proxied_socket = '/var/run/lirc/lircd' # lirc socket
# connect to lirc socket, mangle its output to show long presses, and forward to the proxy
class Mangler(asynchat.async_chat):
def __init__(self, mangle_proxy, proxied_socket):
asynchat.async_chat.__init__(self)
self.set_terminator('\n')
self.create_socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.connect(proxied_socket)
self.mangle_proxy = mangle_proxy
self.ibuffer = '' # buffer for incoming lirc data
self.last_repeats = 0 # repeat count from last lirc event
self.last_time = 0 # time of last lirc event
self.last_line = None # last lirc event split into strings
self.hold_repeats = 0 # number of repeats the key has been held
self.lock = threading.Lock()
self.timer = None
def collect_incoming_data(self, data):
if (data):
self.ibuffer = self.ibuffer + data
def found_terminator(self):
with self.lock:
now = time.time()
self.cancel_timer()
bits = self.ibuffer.split(None, 5)
# check for apple remote button presses
if (len(bits) == 4 and 'apple' in bits[3].lower()):
repeats = int(bits[1], 16)
# if this is a repeat event and the time since the last press was < t we need to check for a long press
if (repeats == self.last_repeats + 1 and now - self.last_time < repeat_threshold):
self.hold_repeats += 1
if (self.hold_repeats >= hold_repeats):
# if we get >= n repeats all within time t of eachother consider it a long press
bits[2] += '_HOLD'
bits[1] = "%0.2X" % (self.hold_repeats - hold_repeats)
self.mangle_proxy.send(' '.join(bits) + '\n')
self.last_line = None
else:
# not enough repeats yet to distinguish between short and long press
self.last_line = bits
self.start_timer()
else:
# if we have a buffered line, it must have been a short press.. so send it
if (self.last_line):
self.last_line[1]='00'
self.mangle_proxy.send(' '.join(self.last_line) + '\n')
# new button press.. dont know if it is short or long press yet
self.hold_repeats = 0
self.last_line = bits
self.start_timer()
# remember the repeat count and time to check against when we get the next event
self.last_repeats = repeats
self.last_time = now
else:
# not an apple button press.. just forward it
self.mangle_proxy.send(self.ibuffer + '\n')
self.ibuffer = ''
# start a timer so we know when to send a short press
def start_timer(self):
self.timer = threading.Timer(repeat_threshold, self.timeout)
self.timer.start()
def cancel_timer(self):
if (self.timer):
self.timer.cancel()
# timer expired.. must have a short press.. replace its repeat count with zero
def timeout(self):
with self.lock:
if (self.last_line and time.time() - self.last_time > repeat_threshold):
self.last_line[1]='00'
self.mangle_proxy.send(' '.join(self.last_line) + '\n')
self.last_line = None
class MangleProxy(asyncore.dispatcher_with_send):
handler_id = 1
def __init__(self, socket, proxied_socket):
asyncore.dispatcher_with_send.__init__(self, socket)
self.handler_id = MangleProxy.handler_id
MangleProxy.handler_id += 1
log.info("Creating handler %d", self.handler_id)
self.mangler = Mangler(self, proxied_socket)
def handle_read(self):
data = self.recv(8192)
if data:
self.mangler.push(data)
def handle_close(self):
log.info("Closing handler %d", self.handler_id)
self.mangler.close();
self.close();
# listen for and accept connects to the proxy socket
class MangleServer(asyncore.dispatcher):
def __init__(self, proxy_socket, proxied_socket):
asyncore.dispatcher.__init__(self)
log.info('Starting server on %s', proxy_socket)
self.proxy_socket = proxy_socket
self.proxied_socket = proxied_socket
self.remove_proxy_socket()
self.create_socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.set_reuse_addr()
self.bind(self.proxy_socket)
self.listen(5)
os.chmod(self.proxy_socket, 0666)
self.closed = False
def remove_proxy_socket(self):
try:
if (os.path.exists(self.proxy_socket)):
os.unlink(self.proxy_socket)
except OSError, error:
log.error('Unable to remove socket file %s : %s', self.proxy_socket, str(error))
raise
def handle_accept(self):
pair = self.accept()
if pair is None:
pass
else:
sock, addr = pair
log.info('Incoming connection from %s', repr(addr))
handler = MangleProxy(sock, self.proxied_socket)
def handle_close(self):
if (not self.closed):
log.info('Stopping server on %s', self.proxy_socket)
self.close()
self.remove_proxy_socket()
self.closed = True
# fix for asyncore bug causing 100% cpu when listening on unix domain sockets
def writable(self):
return False
server = None
try:
# ignore HUPs.. like a good daemon
signal.signal(signal.SIGHUP, lambda x,y: None)
server = MangleServer(proxy_socket, proxied_socket)
asyncore.loop()
finally:
if (server):
server.handle_close()You will need some sort of init script to start it at boot time. I'm using ubuntu 10.04, so here is a simple upstart script. I have it saved as /etc/init/lircmangled.conf
Code:
description "Lirc Mangle Daemon"
start on filesystem
stop on runlevel [!2345]
respawn
respawn limit 10 5
pre-start script
mkdir -p -m0755 /var/run/lirc
end script
exec /usr/local/bin/lircmangled
Search
Help