CSP on pyboard with uasyncio

Not so far ago @pfalcon mentioned in microasync bugtracker about a port of asyncio for micropython – uasyncio. After that I ported asynchronous queue from asyncio to uasyncio, so now it can replace microasync.

So I conceived a little project: device which prints information from pyboard gyro sensor and ultrasonic sensor to OLED display. Sounds very simple, but it need to update information on display when data from one of sensors changed. So interaction with sensors should be non-blocking.

I found almost well done lib (fork which supports drawing text) for interacting with my OLED display and lib for working with ultrasonic sensor (non-blocking version).

First of all I created decorator for simplifying creating generators which returns a queue and make all interaction through it:

class OnlyChanged(Queue):
    def __init__(self, *args, **kwargs):
        self._last_val = None
        super().__init__(*args, **kwargs)

    def put(self, val):
        # Put in queue only if value changed
        if val != self._last_val:
            yield from super().put(val)
            self._last_val = val

def chan(fn):
    def wrapper(*args, **kwargs):
        q = OnlyChanged(1)
        get_event_loop().call_soon(fn(q, *args, **kwargs))
        return q
    return wrapper

So now it’s simple to write generator, which prints to display data received from the queue:

def get_display(q, *args, **kwargs):
    display = Display(*args, **kwargs)
    while True:
        lines = yield from q.get()

>>> display = get_display(pinout={'sda': 'Y10',
...                               'scl': 'Y9'},
...                       height=64,
...                       external_vcc=False)
>>> yield from display.put('Hello world!')

oled display

And generator for the ultrasonic sensor which puts values to the queue:

def get_ultrasonic(q, *args, **kwargs):
    ultrasonic = Ultrasonic(*args, **kwargs)
    while True:
        val = yield from ultrasonic.distance_in_cm()
        yield from q.put(val)

>>> ultrasonic = get_ultrasonic('X1', 'X2')
>>> yield from ultrasonic.get()

Similar generator for the pyboard gyro sensor:

def get_gyro(q):
    accel = pyb.Accel()
    while True:
        val = accel.filtered_xyz()
        yield from q.put(val)

>>> gyro = get_gyro()
>>> yield from gyro.get()
(12, 9, 72)

And by combining all of them it’s very similar to write program for expected device:

def main():
    display = get_display(pinout={'sda': 'Y10',
                                  'scl': 'Y9'},
    current = {'dist': 0, 'xyz': (0, 0, 0)}
    shared_q = alts(dist=get_ultrasonic('X1', 'X2'),
    while True:
        source, val = yield from shared_q.get()
        current[source] = val
        yield from display.put(
            'Distance: {:0.2f}cm\n'
            'x: {} y: {} z: {}'.format(current['dist'], *current['xyz']))

>>> loop = get_event_loop()
>>> loop.call_soon(main())
>>> loop.run_forever()

So the result code is very simple, all components are decoupled and it’s easy to test. Video of result:

Gist with source codes.

Getting total count of indexed documents in the GAE Search API

Around a month ago I was stuck with task – I had to get total count of indexed documents in the GAE Search API. Sounds simple, but not – this API doesn’t have a method for doing that, it has something similar – storage_usage, but this attribute contains size of index in bytes. But the API provides method for receiving ids of documents, so I’ve tried:

>>> len(index.get_range(ids_only=True))
>>> len(index.get_range(ids_only=True, limit=1000))
>>> len(index.get_range(ids_only=True, limit=1001))
ValueError: limit, 1001 must be <= 1000

And another pitfall – get_range couldn’t return more than 1000 ids. So I’ve tried to run it in cycle:

def calculate_count():
    # Starts with 1 because in other iteration new range will
    # start with last item from previous range:
    result = 1
    start_id = None
    while True:
        if start_id is None:
            index_range = index.get_range(ids_only=True, limit=1000)
            index_range = index.get_range(start_id=start_id,
                                          ids_only=True, limit=1000)
        if len(index_range) > 1:
            start_id = index_range[-1].doc_id
            result += len(index_range) - 1  # Ignore last item from previous range
            return result

>>> calculate_count()

Yep, It takes more than 60 seconds. But then I’ve tried to run each iteration in deferred and write result to memcahe:

def calculate_count(start_id=None, result=1):
    if start_id is None:
        index_range = index.get_range(ids_only=True, limit=1000)
        index_range = index.get_range(start_id=start_id,
                                      ids_only=True, limit=1000)
    if len(index_range) > 1:
                       len(index_range) - 1 + result)
        memcache.set('RESULT', result)

>>> calculate_count()
# Wait few minutes...
>>> memcache.get('RESULT')

And It works!

James Turnbull: The Docker Book

book cover Few days ago I finished reading The Docker Book by James Turnbull and probably it’s a good book for them who start using the Docker. This book contains explanation of patterns of usage of the Docker: using for development, for testing and for deploy. About how the Docker works inside. And a lot of information about tools which simplify work with the Docker, I learned about the good tool called Fig from this book.

Maybe some parts of this book duplicates official documentation, but for me this book been a great addition to it. And maybe it’s because I didn’t used the Docker a lot before.

Chrome extension in ClojureScript

Sometimes I need to write/change bunch of code in GAE interactive console and sometimes I need to change build scripts in jenkins tasks. That’s not comfortable to write code in simple browser textarea.

So I decided to create Chrome extension with which I can convert textarea to the code editor (and back) in a few clicks. As an editor I selected Ace because it simple to use and I’d worked with it before. As a language I selected ClojureScript.

For someone who eager: source code of the plugin on github, in the Chrome Web Store.

Developing this extension almost similar to extension in JavaScript, and nearly like ordinary ClojureScript application. But I found a few pitfalls and differences.

ClojureScript compilation

We can’t use :optimizations :none in the Chrome extension, because of goog.require way of loading dependencies. And we should to build separate compiled js files for each background/content/options/etc “pages”. So my cljs-build configuration:

{:builds {:background {:source-paths ["src/textarea_to_code_editor/background/"]
                       :compiler {:output-to "resources/background/main.js"
                                  :output-dir "resources/background/"
                                  :source-map "resources/background/main.js.map"
                                  :optimizations :whitespace
                                  :pretty-print true}}
          :content {:source-paths ["src/textarea_to_code_editor/content/"]
                    :compiler {:output-to "resources/content/main.js"
                               :output-dir "resources/content/"
                               :source-map "resources/content/main.js.map"
                               :optimizations :whitespace
                               :pretty-print true}}}}

If you want to use :optimizations :advanced, you can download externs for Chrome API from github.

Chrome API

From a first look using of Chrome API from ClojureScript is a bit uncomfortable, but with .. macro it looks not worse than in JavaScript. For example, adding listener to runtime messages in js:


And in ClojureScript:

(.. js/chrome -runtime -onMessage (addListener #(.log js/console %)))


Because we can’t use Chrome API in tests I created a little function for detecting if it available:

(defn available? [] (aget js/window "chrome"))

And run all extension bootstrapping code inside of (when (available?) ...). So now it’s simple to use with-redefs and with-reset (for mocking code inside of async tests) for mocking Chrome API.

For running test I used clojurescript.test, my config:

{:builds {:test {:source-paths ["src/" "test/"]
                 :compiler {:output-to "target/cljs-test.js"
                            :optimizations :whitespace
                            :pretty-print false}}}
 :test-commands {"test" ["phantomjs" :runner


Message passing between the extension background and content parts it’s a little pain, because it’s always turns into huge callback hell. But core.async (and a bit of core.match) can save us, for example, handling messages on content side:

(go-loop []
  (match (<! msg-chan)
    [:populate-context-menu data sender-chan] (h/populate-context-menu! data
                                                                        (:used-modes @storage)
    [:clear-context-menu _ _] (h/clear-context-menu!)
    [:update-used-modes mode _] (h/update-used-modes! storage mode)
    [& msg] (println "Unmatched message:" msg))

Sources of content side and backend side helpers for sending/receiving Chrome runtime messages using core.async channels.


Sources on github, extension in the Chrome Web Store.

Eric Evans: Domain-Driven Design: Tackling Complexity in the Heart of Software

book cover This week I finished reading “Domain-Driven Design: Tackling Complexity in the Heart of Software” by Eric Evans and I think it’s a great book, maybe — “must read”. It contains a good explanation of patterns with example situations where they should be used. Also this book contains interesting information about software design and software development on the whole.

From the opposite side, this is quite an old book, it was published in 2003. And some parts of the book are a bit outdated, some patterns and examples are useless now. Also some examples are too “enterprise”, and classes like CargoSpecificationRepositoryFactory scare me. And after reading some examples, I assume that it was painful to be a java developer in early 00.

Mocking clojurescript code written with core.async

When I write tests for the code in clojurescript and core.async I feel little pain — with-redefs doesn’t work correctly with go-blocks. For example I have a function:

(defn get-subtitles
  [sources limit]
  (go (-> (http/get (get-url sources limit))

And the test for it:

(deftest ^:async test-get-subtitles
  (with-redefs [http/get (constantly fixture)]
    (go (is (= (<! (get-subtitles const/all 100)) expected))

It didn’t work, it actually tries to make http request. Ok, I can try to put with-redefs inside go-block:

(deftest ^:async test-get-subtitles
  (go (with-redefs [http/get (constantly fixture)]
        (is (= (<! (get-subtitles const/all 100)) expected)))
      (with-redefs [http/get (constantly blank-result)]
        (is (= (<! (get-subtitles const/addicted 100)) [])))

The first assertion works, but in the second assertion I have previously redefined http/get and it’s incorrect and the assertion fails — with-redefs permanently changes var when applied in the go-block.

So I’ve developed a little macro which works like with-redefs and can be used inside of go-block and with code without core.async:

(defmacro with-reset
  [bindings & body]
  (let [names (take-nth 2 bindings)
        vals (take-nth 2 (drop 1 bindings))
        current-vals (map #(list 'identity %) names)
        tempnames (map (comp gensym name) names)
        binds (map vector names vals)
        resets (reverse (map vector names tempnames))
        bind-value (fn [[k v]] (list 'set! k v))]
    `(let [~@(interleave tempnames current-vals)]
         ~@(map bind-value binds)
           ~@(map bind-value resets))))))

And usage:

(deftest ^:async test-get-subtitles
  (go (with-reset [http/get (constantly fixture)]
        (is (= (<! (get-subtitles const/all 100)) expected)))
      (with-reset [http/get (constantly blank-result)]
        (is (= (<! (get-subtitles const/addicted 100)) [])))

I put this macro in clj-di for avoiding copying it between projects.

Tests for this macro on github.

UPD: example updated.

Destructuring arguments in CoffeeScript

CoffeeScript has a good and fairly undocumented feature — destructuring arguments. It’s very similar to binding forms from clojure and a bit like pattern matching like in scala and erlang. It has syntax similar to destructuring assignment but works with functions (and methods) arguments.

Simple examples:

getUppercaseName = ({name}) -> name.toUpperCase()
getUppercaseName name: 'test'
# 'TEST'

head = ([x, ...]) -> x
head [1, 2, 3, 4]
# 1

tail = ([_, xs...]) -> xs
tail [1, 2, 3, 4]
# [ 2, 3, 4 ]

prettifyUser = ({groups: {names: groupNames}, username}) ->
    "#{username}: #{groupNames.join(', ')}"
    groups: {names: ['a', 'b', 'c']}
    username: 'user'
# 'user: a, b, c'

Looks useful? But will be more if we combine it with underscore.js. Assume we need to transform result of $.serializeArray() like:

    {"name": "title", "value": "Test Title"},
    {"name": "text", "value": ""},
    {"name": "tags", "value": "Python"},
    {"name": "tags", "value": "CoffeeScript"},
    {"name": "tags", "value": "Clojure"},
    {"name": "csrfmiddlewaretoken", "value": "token"}

To something like (fields with blank value and service fields should be removed, values of fields with same name should be presented as a list):

    "title": "Test Title",
    "tags": ["Python", "CoffeeScript", "Clojure"]

Let’s start with imperative solution:

transform = (serializedForm) ->
    groupByName = ->
        result = {}
        for field in serializedForm
            if field.value != '' and field.name != 'csrfmiddlewaretoken'
                if result[field.name]
                    result[field.name].push field.value
                    result[field.name] = [field.value]
    flattenFieldsWithSingleValue = ->
        result = {}
        for name, value in groupByName()
            if value.length == 1
                result[name] = value[0]
                result[name] = value

It works, but looks ugly and complex.

Now look to implementation with underscore.js and destructuring arguments:

transform = (serializedForm) ->
    _.chain serializedForm
     .reject ({name, value}) ->
         value == '' or name == 'csrfmiddlewaretoken'
     .groupBy 'name'
     .map (vals, name) ->
         vals = _.map(vals, ({value}) -> value)
         [name, if vals.length > 1 then vals else vals[0]]

A lot better I think. Less lines of code, each transformation step can be simple separated from each other. And all this because of fluent interface presented by underscore.js _.chain and destructuring arguments.

Simple application for Google Glass written in scala

About a month ago I received Google Glass and only last weekend I found time for developing something for it. I don’t have a lot of experience in developing software for android, so I’ve started developing simple TODO application.

As a language I selected scala (but I think I should try to write similar application in clojure) and gradle as a build system (because it works out of the box with Android Studio).

For integrating scala and gradle I’ve used gradle-android-scala-plugin which works perfectly and requires adding only ~10 lines to build.gradle for configuration.

And I used scaloid — it’s a very great library, it makes work with android API less painful and makes code a bit more Scala-style.

Developing in scala for android is fun, but I stumbled over one not straightforward problem (perhaps it’s straightforward for mature android developers) — compilation fails with:

[info] Generating classes.dex
[warn] trouble writing output: Too many method references: 181232; max is 65536.
[warn] You may try using --multi-dex option.
[warn] References by package:

And the simple solution — change my proguard config to:

-dontwarn scala.**
-keep class scala.collection.SeqLike {
    public protected *;

Another minus it’s compile time. My simple application compiles around 1.5 minutes, tests — around 1 minute. But I found that incremental compilation will be available soon in gradle-android-scala-plugin, so this minus will disappear.

Source code of result, video of result:

Why I like writing proper docstrings while coding in python

Everyone knows that having proper docstrings in code is good, and most straightforward reason why — it’s in-place documentation for classes, functions and methods. So that’s another three reasons:

Sphinx and Read the Docs

For example, we have simple function with docstring:

def as_chan(create_chan):
    """Decorator which creates channel and coroutine. Passes channel as a
    first value to coroutine and returns that channel.


    .. code-block:: python

        def thermo(chan, unit):
            while True:
                yield chan.put(convert(thermo_get(), unit))

        def main():
            thermo_chan = thermo('C')
            while True:
                print((yield thermo_chan.get()))  # prints current temperature

    :param create_chan: Type of channel.
    :type create_chan: type[Channel]
    :returns: Created coroutine.

    def decorator(fn):
        def wrapped(*args, **kwargs):
            chan = create_chan()
            coroutine(fn)(chan, *args, **kwargs)
            return chan
        return wrapped
    return decorator

And using sphinx-apidoc and Read the Docs we can generate cool documentation like this. And it’ll be automatically generated after each push to github.

Quick help in IDE

I’ve been using pycharm, but I think there’s something similar in other IDEs.

Suppose we have function (full source code):

def get_servo(num):
    """Creates write and read channels for servo. Should be used
    only in coroutine.


    .. code-block:: python
        servo_set, servo_get = get_servo(1)
        yield servo_set.put(90)  # set servo to 90 degrees
        yield servo_get.get()  # prints current servo degree

    :param num: Number of servo.
    :type num: int
    :returns: Write and read channels.
    :rtype: (Channel, SlidingChannel)


It’s not simple to guess what it returns and how work with it, but if we just press ctrl+Q:


Warnings on wrong type and better autocomplete with type hinting

Assume we started writing simple function:

def get_user_groups(user):
    :type user: User 
    :rtype: list[Group]

With type hinting we can use proper autocomplete inside it:


And for result of the function:


And we can see warnings when pass value of wrong type to the function:


More about type hinting.

Steven F. Lott: Mastering Object-oriented Python

book cover Few days ago I finished reading “Mastering Object-oriented Python” by Steven F. Lott, and I think it’s nearly a good book. Maybe it’s longer than need (~600 pages) and sometimes I used to think that content repeats. Some code samples are controversial and not always follows PEP8.

But I think this book worth reading, they contains good explanation of magic methods, good examples of using abc and collections.abc, interesting chapter about logging and good manual for documenting python code.