9 02 2019

Let me give you a little bit of context. I’m doing remote working since quite a long time now. That’s mean I don’t take the train as often as I used to. I really enjoy it but, as a side effect, I lost the habit of taking a crowded train, which make it even more pain the few times I have to take it today.

My home is located around 7 to 10 min walk to the train station. Depending if you walk fast or slow. We also have to take consideration that the train schedule could vary from every 20 min to 1 h based on the hour of the day.

As I’m not quite lucky, usually when I’m ready to go to catch my train, it’s usually the worst timing to be able to get a train. So I have to wait the next one, grab my laptop, start working on something and miss the optimal timing to not wait at the station again :p

I decided to find a geeky solution. The first thing to look at is how can I access to train schedule data.

The most common and easy way to do geeky things is to use API. So I start to look if there is one API available for RATP (Autonomous Operator of Parisian Transports). And … what a really good news, they decided to make their API public (https://data.ratp.fr/explore/dataset/horaires-temps-reel/).

I followed docs to know how access to work with this public API and learned that I need to request access to it. Fine it’s quite usual to create an account on the software and ask for a token to access to it.

Unfortunately, I had some little surprise to work with this API. After I read the procedure I was confused, I felt like doing my paperwork. To request an access to the API I had to print paper form and filled it with my personal information, a static IP to access to the API and sign it. Then scan it and send it back.

Then, as all good old software, don’t expect a nice and clean REST API, no no no, enjoy the SOAP API and the huge wsdl file and burn your eyes on big blocks of xml. To be honest it was nearly as painful to use then wait the train.

But those things apart, that’s really nice to see those companies moving forward, allowing people play their API.

At this point I wrote a little dumb script to get from the API the remaining time for my next train.

import zeep
from datetime import datetime

def cache(func):
    cache = {}
    def wrapper(*args, **kwargs):
        key = "%s:%s" % (args,kwargs)
        try:
            ret = cache[key]
        except KeyError:
            ret = func(*args, **kwargs)
            cache[key] = ret 
        return ret 
    return wrapper

LINE = {    'code': 'J',
            'id': '112382',
            'reseau': {
                'code': 'sncf',
                'name': 'SNCF'
            }
        }

def gap_in_min(t1, t2):
    gap = t1 - t2
    return round(gap.total_seconds()/60)

wsdl = 'Wsiv.wsdl'
client = zeep.Client(wsdl=wsdl)

@cache
def get_stations(line):
    return client.service.getStations(station={'line': line})

@cache
def get_directions(line):
    return client.service.getDirections(line)

@cache
def get_lines(line):
    return client.service.getLines(line)

def get_next_mission_time():
    now = datetime.now()
    
    lines = get_lines(LINE)
    for line in lines:
        stations = get_stations(line)
        for station in stations['stations']:
            if 'Houilles Carrieres-sur-Seine' in station['name']:
                directions = get_directions(line)
                for direction in directions['directions']:
                    if direction['name'] != "Gare Saint-Lazare": continue
                    missions = client.service.getMissionsNext(station=station, direction=direction, limit=3)
                    if not missions['missions']: continue
                    _station = missions['missions'][0]['stations'][0]['name']
                    _date = datetime.strptime(missions['missions'][0]['stationsDates'][0], "%Y%m%d%H%M")
                    return gap_in_min(_date,  now)
    return -127

if __name__ == "__main__":
    get_next_mission_time()

What can I do with those datas ? I firstly thought create a simple web page with this information and use it on my phone. But that was not convenient enough.

Searching for idea on internet, I found a real USB traffic light (http://www.cleware-shop.de/USB-MiniTrafficLight-EN), it looked really fun so I goes for it …

Official sample of code :

• http://www.cleware-shop.de/epages/63698188.sf/en_US/?ObjectPath=/Shops/63698188/Categories/Downloads
• https://github.com/flok99/clewarecontrol

Official example are mainly in cpp, python examples are kind of binding on those cpp library. Having to compile some cpp code to interact with an USB device is kind of boring. I was looking for a python script to directly communicate with my USB device using the python standard USB library. I decided to search what people did on internet.

I saw lots of peoples playing with it in different languages (ruby, python, node, …). I tried several Python codes but I struggle a little bit with it. Some were not working or had strange behavior with my traffic light. The best sample of code I found is https://github.com/rbaier/python-clewareampel

So I decided to debug it and tried to make it work with my traffic light. The code was very straightforward, I found only two potential line which could fail in my case.

        self.device = usb.core.find(idVendor=0x0d50, idProduct=0x0008)

Which I simplify verified using lsusb command

lsusb
Bus 001 Device 045: ID 0d50:0008 Cleware GmbH

And the following one :

    def _set_led(self, color, value):
        self.device.ctrl_transfer(0x21, 0x09, 0x200, 0x00, [0x00, color, value])

To be sure of the hexa code, I ended reading USB docs and adding debug on the cpp code. It was really fun to dig into the cpp code and convert decimal to hexa and compare results with my Python code.

• http://www.usbmadesimple.co.uk/ums_4.htm
• https://www.beyondlogic.org/usbnutshell/usb6.shtml

The hexa code of all my sample was similar. As I’m not an USB expect, those hexa values sound legit as it’s described a SET_CONFIGURATION request from my host to the USB device. I assumed it was ok :

e.ctrl_transfer( 0x21, 0x09, 0x200, 0x00, [0x00, color, value])

# ctrl_transfer(self, bmRequestType, bRequest, wValue=0, wIndex=0, data_or_wLength=None, timeout=None)

Following the doc, we have quite a lot of information in the Setup Packet for bmRequestType

In our case which give :

And about bRequest = 0x09 If we read the Standard Device Requests section it’s SET_CONFIGURATION (0x09)

From this part I started to look one layer above and read some docs about pyusb.ctrl_transfer. Sadly the first sentence was :

I don’t intend to explain the purpose of each transfer and the differences among them. Therefore, I assume you know at least the basics of the USB transfers.

Which is not my case. I decided to give a random try at the way described in the starting guide https://github.com/pyusb/pyusb/blob/master/docs/tutorial.rst#lets-get-it-started

The example in the starting guide does not use ctrl_transfert. It gives me the following code : https://github.com/gaelL/python-clewareampel

Magically it worked. I felt quite lucky, but anyway it’s only a hacky project, I don’t really have big plan for it.

Some implementation later I had this nice fully working traffic light following this color code.

•  Blinking, Unexpected error. I’m broken, help me 🙂
•  No train for the next one hour.
•  Don’t go now
•  Be prepared to leave soon
•  Go, optimal time to catch the train
•  Hurry up, if you walk quite fast, you will be able to catch the train.

My last work was to setup it on a raspberry and that’s it, I had a nice traffic light at home to tell me when I to take the train.

gaelL

Concourse lazy pipeline templating for free ?