Schinckel.net

One of my key tenets of Home Automation is that as much stuff should be done locally as possible. Whilst with HomeAssistant, that means “in the local network”, I feel that “in the device” is even better. However, sometimes values should still be configurable (from within Home Assistant), but the device should still work even if there is not (and has never been) a connection to Home Assistant since boot.

For instance, I have a sensor light. The device contains a PIR sensor, with configurable delay, sensitivity and lux settings. This in turn drives a relay connected to a pair of PAR bulbs, and additionally has a green LED. This device usually only has the ESP8266 controlling if the thing should ever turn on at all, but I hacked it so that it exposes the PIR sensor, and allows triggering the relay.

But the delay between the motion no longer being detected and the light turning off should be configurable. I don’t want to have to re-flash the device if I decide to change this value, so I wanted to make it that the value will be fetched from Home Assistant (if it is set).

This turned out to work really well. There are a few parts that need to be set up in the YAML (this is not the complete file):

esphome:
  name: ${device_name}
  platform: ESP8266
  board: esp01_1m
  on_boot:
    - logger.log:
        level: DEBUG
        format: 'Light ON time is set to %d'
        args: ['id(light_on_time)']

globals:
  - id: light_on_time
    type: int
    restore_value: true
    initial_value: '30'

sensor:
  - platform: homeassistant
    id: on_time
    entity_id: input_number.${device_name}
    on_value:
      then:
        - globals.set:
            id: light_on_time
            value: !lambda 'return int(x);'

It really is that simple. The first boot will set the light_on_time variable to 30. Then, when it connects to Home Assistant, it will look for an input_number.<device_name> (which matches the device name). If it finds one (or is ever told about this value changing), then it will commit that new value to the flash, and this will be be restored after a reboot.

There is one other thing we could do here, to make it so that we don’t write the value to the flash if it has not changed (and prevent wear to that, since it is limited to a number of writes):

sensor:
  - platform: homeassistant
    id: on_time
    entity_id: input_number.${device_name}
    on_value:
      then:
        if:
          condition:
            lambda: 'return id(light_on_time) != int(x);'
          then:
            - logger.log:
                level: DEBUG
                format: 'Light ON time changed from %d to %d seconds'
                args: ['id(light_on_time)', 'int(x)']
            - globals.set:
                id: light_on_time
                value: !lambda 'return int(x);'

With regards to a similar problem, detecting if it is dark enough to turn the light on should be something like “a bit before sunset to a bit after sunrise”. I could set the lux threshold on the device, but it would be nice to have motion detection work during the day too.

Here, we can have a global variable sun_is_down that defaults to being true, and is only set to false when the sun comes up. However, we would also want this to trigger when first connecting to Home Assistant and the sun is already up.

We can use a wait_until in our on_boot handler to trigger this:

globals:
  - id: sun_is_down
    type: bool
    restore_value: false
    initial_value: 'true'

esphome:
  ...
  on_boot:
    - wait_until:
        api.connected:
    - if:
        condition:
          sun.is_above_horizon:
        then:
          - logger.log:
              level: DEBUG
              format: 'Sun is up, light will not turn on.'
          - globals.set:
              id: sun_is_down
              value: 'false'
        else:
          - logger.log:
              level: DEBUG
              format: 'Sun is down, light will turn on.'

sun:
  on_sunrise:
    - elevation: 10°
      then:
        - logger.log:
            format: 'The sun is now down, light will turn on.'
        - globals.set:
            id: sun_is_down
            value: 'false'
  on_sunset:
    - elevation: 10°
      then:
        - logger.log:
            format: 'The sun is now up, light will not turn on.'
        - globals.set:
            id: sun_is_down
            value: 'true'

I’ve used a 10° elevation to trigger the light to turn on a bit before sunset and still turn on a bit after sunrise when motion is detected. I haven’t figured out a way to get the same check to apply in the sun.is_above_horizon: check.

Working from home means that I don’t use my AirPods as much as I used to: I’m happy to mostly listen to the music my partner (who has worked from home for over a decade) wants to listen to. Lately, that has been NonStop80s.

We play this through the IKEA Symphonisk stereo pair in the dining room, several metres away from the office. The sound is good, and it’s nice not having it come out of the computer speaker.

Because I’m a nut, I have all of this hooked up to Home Assistant, and have one nice automation that will turn off the music that is playing on that speaker when the TV in that room is turned on.

Sometimes the tracks that are played on NonStop80s are really bad. For instance, Snooker Loopy, by Chas and Dave is an absolute piece of shit song. If it weren’t a streaming station, I’d just hit skip.

What would be nice is to be able to have the speakers stop playing, and start playing again when the next track comes on.

Enter “Mute the rest of this track”.

alias: Mute the rest of this track
sequence:
  - condition: state
    entity_id: media_player.dining_room_sonos
    state: playing
  - condition: template
    value_template: '{{ not state_attr("media_player.dining_room_sonos", "is_volume_muted") }}'
  - service: media_player.volume_mute
    data:
      is_volume_muted: true
    entity_id: media_player.dining_room_sonos
  - delay: '1'
  - wait_for_trigger:
      - platform: state
        entity_id: media_player.dining_room_sonos
    timeout: '{{ timeout or 300 }}'
  - service: media_player.volume_mute
    data:
      is_volume_muted: false
    entity_id: media_player.dining_room_sonos
mode: single
icon: 'mdi:volume-mute'
description: Mute the media player until the media_title attribute changes.
fields:
  timeout:
    description: Amount of time to stay muted if the track does not change
    example: 300

This is a script that you can use in Home Assistant to do just that. It will (iff the dining_room_sonos is unmuted, and currently playing), mute that speaker until something else on the speaker changes.

This could be a track change, or a volume change, or anything else. It’s deliberately loose, so that if it was muted, and someone changes the volume, it unmutes immediately.

After a configurable (default 300 seconds) timeout, it will unmute.

I was trying to make it so that you could provide the named media_player entity, but there still seem to be places where you can’t use templates (but need to provide the string directly). I was able to get the conditions to use the variable, but was not able to figure out how to use them in the service calls…and to be honest, I don’t really need it to apply to other players.

One of my team implemented RobotFramework support for automated browser testing of our platform a while ago. At the time, we were using Codeship Basic, and I built a helper to run a robot test suite within a tox environment. It was all good, because chromedriver and all it’s dependencies were already installed.

But time passes, and we needed to move to Codeship Pro. Which has some neater features, but required me to build docker images for everything. We already use docker for deployment, but I didn’t really want to build a bunch of distinct images just for testing that re-implemented the same stuff that we have in our deployment images. Even just appending new stuff to them means that things could turn out to be a pain in the arse to manage.

And getting chromedriver installed into a docker image is not neat.

I did find a docker image that just has an instance of chromedriver, and exposes that. But getting that to work with robot was still a bunch of work. After much experimentation, I was able to get the connections between everything to work.

First, we need to have the chromedriver container running:

$ docker run -p 4444:4444 CHROMEDRIVER_WHITELISTED_IPS='' robcherry/docker-chromedriver:latest

Then, there are a few moving parts that need to be in place to get things to work. Using my djangobot management command (which I had to extend a bit here), a single command can be used to spin up a Django runserver command, apply migrations (if necessary), and then run the robot commands. The trick is you need to teach Robot to speak to the remote WebDriver instance, which then in turn speaks to the running django webserver.

First, the RobotFramework commands; my resource.robot file which is referenced by all of my robot test suites contains:

*** Variables ***

${HOSTNAME}         127.0.0.1
${PORT}             8000
${SCHEME}           http
${SERVER}           ${SCHEME}://${HOSTNAME}:${PORT}
${BROWSER}          headlesschrome
${TIMEOUT}          30
${REMOTE_URL}

*** Settings ***

Documentation   A resource file with reusable keywords and variables.
Library         SeleniumLibrary             timeout=${TIMEOUT}      implicit_wait=1
Library         Collections
Library         DebugLibrary
Library         DateTime
Library         String
Library         djangobot.DjangoLibrary     ${HOSTNAME}     ${PORT}

*** Keywords ***

Create Remote Webdriver
    ${chrome_options} =     Evaluate    sys.modules['selenium.webdriver'].ChromeOptions()    sys, selenium.webdriver
    Call Method    ${chrome_options}   add_argument    headless
    Call Method    ${chrome_options}   add_argument    disable-gpu
    Call Method    ${chrome_options}   add_argument    no-sandbox
    ${options}=     Call Method     ${chrome_options}    to_capabilities

    Create Webdriver    Remote   command_executor=${REMOTE_URL}    desired_capabilities=${options}
    Open Browser    ${SERVER}   ${BROWSER}  remote_url=${REMOTE_URL}    desired_capabilities=${options}

Start Session
    Run Keyword If  '${REMOTE_URL}'    Create Remote Webdriver
    Run Keyword If  '${REMOTE_URL}' == ''    Open Browser    ${SERVER}   ${BROWSER}

    Set Window Size     2048  2048
    Fetch Url       login
    Add Cookie      robot   true

    Register Keyword To Run On Failure    djangobot.DjangoLibrary.Dump Error Data

End Session
    Close Browser

Logout
    Fetch Url     logout

Notice that the Start Session keyword determines which type of browser to open - either a local or remote one.

Thus, each *.robot file starts with:

*** Settings ***
Resource                resource.robot
Suite Setup             Start Session
Suite Teardown          End Session
Test Setup              Logout

Because the requests will no longer be coming from localhost, you need to ensure that your runserver is listening on the interface the requests will be coming from. If you can’t detect this, and your machine is not exposed to an insecure network, then you can use 0.0.0.0 to get the django devserver to listen on all interfaces. You will also need to supply the hostname that you will be using for the requests (which won’t be localhost anymore), and ensure this is in your Django settings.ALLOWED_HOSTS.

In my case, I needed to make my robot command allow all this, but ultimately I can now do:

$ ./manage.py robot --runserver 0 \
                    --listen 0.0.0.0 \
                    --hostname mymachine.local \
                    --remote-url http://localhost:4444 \
                    --include tag

This runs against the database I already have prepared, but in my codeship-steps.yml I needed to do a bit more, and hook it up to the other containers:

coverage run --branch --parallel \
    /app/manage.py robot --migrate \
                         --server-url=http://web.8000 \
                         --remote-url=http://chromedriver:4444 \
                         --tests-dir=/app/robot_tests/ --output-dir=/coverage/robot_results/ \
                         --exclude skip  --exclude expected-failure

Now, if only Codeship’s jet tool actually cached multi-stage builds correctly.

Maybe I neeed to try this.

One of my colleagues has spent a bunch of time investigating and then implementing some testing using Robot Framework. Whilst at times the command line feels like it was written by someone who hasn’t used unix much, it’s pretty powerful. There are also some nice tools, like several Google Chrome plugins that will record what you are doing and generate a script based upon that. There are also other tools to help build testing scripts.

There is also an existing DjangoLibrary for integrating with Django.

It’s an interesting approach: you install some extra middleware that allows you to perform requests directly to the server to create instances using Factory Boy, or fetch data from Querysets. However, it requires that the data is serialised before sending to the django server, and the same the other way. This means, for instance, that you cannot follow object references to get a related object without a bunch of legwork: usually you end up doing another Query Set query.

There are some things in it that I do not like:

• A new instance of the django runserver command is started for each Test Suite. In our case, this takes over 10 seconds to start as all imports are processed.
• The database is flushed between Test Suites. We have data that is added through migrations that is required for the system to operate correctly, and in some cases for tests to execute. This is the same problem I’ve seen with TransactionTestCase.
• Migrations are applied before running each Test Suite. This is unnecessary, and just takes more time.
• Migrations are created automatically before running each Test Suite. This is just the wrong approach: at worst you’d want to warn that migrations are not up to date - otherwise you are testing migrations that may not have been committed: your CI would pass because the migrations were generated, but your system would fail in reality because those migrations do not really exist. Unless you are also making migrations directly on your production server and not committing them at all, in which case you really should stop that.

That’s in addition to having to install extra middleware.

But, back onto the initial issue: interacting with Django models.

What would be much nicer is if you could just call the python code directly. You’d get python objects back, which means you can follow references, and not have to deal with serialisation.

It’s fairly easy to write a Library for Robot Framework, as it already runs under Python. The tricky bit is that to access Django models (or Factory Boy factories), you’ll want to have the Django infrastructure all managed for you.

Let’s look at what the DjangoLibrary might look like if you are able to assume that django is already available and configured:

import importlib

from django.apps import apps
from django.core.urlresolvers import reverse

from robot.libraries.BuiltIn import BuiltIn


class DjangoLibrary:
    """

    Tools for making interaction with Django easier.

    Installation: ensure that in your `resource.robot` or test file, you have the
    following in your "***Settings***" section:

        Library         djangobot.DjangoLibrary     ${HOSTNAME}     ${PORT}

    The following keywords are provided:


    Factory:        execute the named factory with the args and kwargs. You may omit
                    the 'factories' module from the path to reduce the amount of code
                    required.

        ${obj}=     Factory     app_label.FactoryName       arg  kwarg=value
        ${obj}=     Factory     app_label.factories.FactoryName     arg  kwarg=value


    Queryset:       return a queryset of the installed model, using the default manager
                    and filtering according to any keyword arguments.

        ${qs}=      Queryset    auth.User       pk=1


    Method Call:    Execute the callable with tha args/kwargs provided. This differs
                    from the Builtin "Call Method" in that it expects a callable, rather
                    than an instance and a method name.

        ${x}=       Method Call     ${foo.bar}      arg  kwargs=value


    Relative Url:   Resolve the named url and args/kwargs, and return the path. Not
                    quite as useful as the "Url", since it has no hostname, but may be
                    useful when dealing with `?next=/path/` values, for instance.

        ${url}=     Relative Url        foo:bar     baz=qux


    Url:            Resolve the named url with args/kwargs, and return the fully qualified url.

        ${url}=     Url                 foo:bar     baz=qux


    Fetch Url:      Resolve the named url with args/kwargs, and then using SeleniumLibrary,
                    navigate to that URL. This should be used instead of the "Go To" command,
                    as it allows using named urls instead of manually specifying urls.

        Fetch Url   foo:bar     baz=qux


    Url Should Match:   Assert that the current page matches the named url with args/kwargs.

        Url Should Match        foo:bar     baz=qux

    """

    def __init__(self, hostname, port, **kwargs):
        self.hostname = hostname
        self.port = port
        self.protocol = kwargs.pop('protocol', 'http')

    @property
    def selenium(self):
        return BuiltIn().get_library_instance('SeleniumLibrary')

    def factory(self, factory, **kwargs):
        module, name = factory.rsplit('.', 1)
        factory = getattr(importlib.import_module(module), name)
        return factory(**kwargs)

    def queryset(self, dotted_path, **kwargs):
        return apps.get_model(dotted_path.split('.'))._default_manager.filter(**kwargs)

    def method_call(self, method, *args, **kwargs):
        return method(*args, **kwargs)

    def fetch_url(self, name, *args, **kwargs):
        return self.selenium.go_to(self.url(name, *args, **kwargs))

    def relative_url(self, name, *args, **kwargs):
        return reverse(name, args=args, kwargs=kwargs)

    def url(self, name, *args, **kwargs):
        return '{}://{}:{}'.format(
            self.protocol,
            self.hostname,
            self.port,
        ) + reverse(name, args=args, kwargs=kwargs)

    def url_should_match(self, name, *args, **kwargs):
        self.selenium.location_should_be(self.url(name, *args, **kwargs))

You can write a management command: this allows you to hook in to Django’s existing infrastructure. Then, instead of calling robot directly, you use ./manage.py robot

What’s even nicer about using a management command is that you can have that (optionally, because in development you probably will already have a devserver running) start runserver, and kill it when it’s finished. This is the same philosophy as robotframework-DjangoLibrary already does, but we can start it once before running out tests, and kill it at the end.

So, what could our management command look like? Omitting the code for starting runserver, it’s quite neat:

from __future__ import absolute_import

from django.core.management import BaseCommand, CommandError

import robot


class Command(BaseCommand):
    def add_arguments(self, parser):
        parser.add_argument('tests', nargs='?', action='append')
        parser.add_argument('--variable', action='append')
        parser.add_argument('--include', action='append')

    def handle(self, **options):
        robot_options = {
            'outputdir': 'robot_results',
            'variable': options.get('variable') or []
        }
        if options.get('include'):
            robot_options['include'] = options['include']

        args = [
            'robot_tests/{}_test.robot'.format(arg)
            for arg in options['tests'] or ()
            if arg
        ] or ['robot_tests']

        result = robot.run(*args, **robot_options)

        if result:
            raise CommandError('Robot tests failed: {}'.format(result))

I think I’d like to do a bit more work on finding tests, but this works as a starting point. We can call this like:

./manage.py robot foo --variable BROWSER:firefox --variable PORT:8000

This will find a test called robot_tests/foo_test.robot, and execute that. If you omit the test argument, it will run on all tests in the robot_tests/ directory.

I’ve still got a bit to do on cleaning up the code that starts/stops the server, but I think this is useful even without that.