Whisker Python client library

Python package for Whisker clients.

• Whisker is a TCP/IP-based research control software suite. See http://www.whiskercontrol.com/
• Licensed under a permissive open-source license; see LICENSE.

TL;DR

• pip install whisker

• Fire up your Whisker server.

• Test with:

  whisker_test_twisted --server localhost
  
  whisker_test_rawsockets --server localhost
  

• Copy/paste the demo config file and demo task (see A complete simple task) and run it.

Usage

There are three styles of Whisker client available. Full worked examples are provided, along with a rationale for their use; see Rationale. The outlines, however, look like these:

Twisted client (preferred for simple interfaces)

from twisted.internet import reactor
from whisker.twistedclient import WhiskerTwistedTask

class MyWhiskerTask(WhiskerTwistedTask):
    # ...

w = MyWhiskerTask()
w.connect(...)
reactor.run()

Qt client (preferred for GUI use)

More complex; see the Starfeeder project example.

Raw socket client (deprecated)

from whisker.rawsocketclient import WhiskerRawSocketClient

w = WhiskerRawSocketClient()
w.connect_both_ports(...)
# ...
for line in w.getlines_mainsocket():
    # ...

Rationale

Approaches to sockets and message passing

Whisker allows a multitude of clients in a great many languages – anything that can “speak” down a TCP/IP port, such as C++, Visual Basic, Perl, and Python.

Whisker uses two sockets, a main socket, through which the Whisker server can send events unprompted, and an immediate socket, used for sending commands/queries and receiving immediate replies with a one-to-one relationship between commands (client → server) and responses (server → client). Consequently, the client must deal with unpredictable events coming from the server. It might also have to deal with some sort of user interface (UI) code (and other faster things, like data storage).

C++

In C++/MFC, sockets get their own thread anyway, and the framework tries to hide this from you. So the GUI and sockets coexists fairly happily. Many Whisker tasks use C++, but it’s not the easiest thing in the world.

Perl

In Perl, I’ve used only a very basic approach with a manual message loop, like this:

# Enter a loop to listen to messages from the server.
while (chomp($line = <$MAINSOCK>)) {
    if ($line =~ /^Ping/) {
        # If the server has sent us a Ping, acknowledge it.
        Send("PingAcknowledged");

    } elsif ($line =~ /^Event: (.+)/) {
        # The server has sent us an event.
        $event = $1;

        # Event handling for the behavioural task is dealt with here.

        if ($event =~ /^EndOfTask$/) { last; } # Exit the while loop.
    }
}

Python

In Python, I’ve used the following approaches:

Manual event loop

You can use base socket code, and poll the main socket for input regularly. Simple. But you can forget about simultaneous UI. Like this:

Non-threaded, event-driven

The Twisted library is great for this (https://twistedmatrix.com/). However:

• Bits of it, like Tkinter integration, still don’t support Python 3 fully (as of 2015-12-23), though this is not a major problem (it’s easy to hack in relevant bits of Python 3 support).
• Though it will integrate its event loop (reactor) with several GUI toolkits, e.g. http://twistedmatrix.com/documents/current/core/howto/choosing-reactor.html this can still fail; e.g. with Tkinter, if you open a system dialogue (such as the standard “Open File…” or “Save As…” dialogues), the Twisted reactor will stop and wait, which is no good. This is a much bigger problem. (More detail on this problem in my dev_notes.txt for the starfeeder project.)
• So one could use Twisted with no user interaction during the task.

It looks, from the task writer’s perspective, like this:

#!/usr/bin/env python

"""
whisker/test_twisted.py

===============================================================================

    Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

    This file is part of the Whisker Python client library.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.

===============================================================================

**Command-line tool to test the Whisker Twisted client.**

"""

import argparse
import logging

from twisted.internet import reactor

from cardinal_pythonlib.logs import configure_logger_for_colour
from whisker.api import (
    Pen,
    PenStyle,
    BrushStyle,
    BrushHatchStyle,
    Brush,
    Rectangle,
)
from whisker.constants import DEFAULT_PORT
from whisker.twistedclient import WhiskerTwistedTask

log = logging.getLogger(__name__)

DEFAULT_DISPLAY_NUM = 0
DEFAULT_AUDIO_NUM = 0
DEFAULT_INPUT_LINE = 0
DEFAULT_OUTPUT_LINE = 64
DEFAULT_MEDIA_DIR = r"C:\Program Files\WhiskerControl\Server Test Media"
DEFAULT_BITMAP = "santa_fe.bmp"
DEFAULT_VIDEO = "mediaexample.wmv"
DEFAULT_WAV = "telephone.wav"

AUDIO = "audio"
DISPLAY = "display"
DOC = "doc"
VIDEO = "_video"


class MyWhiskerTwistedTask(WhiskerTwistedTask):
    """
    Class deriving from :class:`whisker.twistedclient.WhiskerTwistedTask` to
    demonstrate the Twisted Whisker client.
    """
    def __init__(self,
                 display_num: int,
                 audio_num: int,
                 input_line: str,
                 output_line: str,
                 media_dir: str,
                 bitmap: str,
                 video: str,
                 wav: str) -> None:
        super().__init__()  # call base class init
        self.display_num = display_num
        self.audio_num = audio_num
        self.input = input_line
        self.output = output_line
        self.media_dir = media_dir
        self.bitmap = bitmap
        self.video = video
        self.wav = wav
        # ... anything extra here

    def fully_connected(self) -> None:
        """
        Called when the server is fully connected. Set up the "task".
        """
        print("SENDING SOME TEST/DEMONSTRATION COMMANDS")
        self.whisker.get_network_latency_ms()
        self.whisker.report_name("Whisker Twisted client prototype")
        self.whisker.timestamps(True)
        self.whisker.timer_set_event("TimerFired", 1000, 9)
        self.whisker.timer_set_event("EndOfTask", 12000)

        # ---------------------------------------------------------------------
        # Audio
        # ---------------------------------------------------------------------

        # ---------------------------------------------------------------------
        # Display
        # ---------------------------------------------------------------------
        bg_col = (0, 0, 100)
        pen = Pen(width=3, colour=(255, 255, 150), style=PenStyle.solid)
        brush = Brush(
            colour=(255, 0, 0), bg_colour=(0, 255, 0),
            opaque=True, style=BrushStyle.hatched,
            hatch_style=BrushHatchStyle.bdiagonal)
        self.whisker.claim_display(number=self.display_num, alias=DISPLAY)
        display_size = self.whisker.display_get_size(DISPLAY)
        log.info("display_size: {}".format(display_size))
        self.whisker.display_scale_documents(DISPLAY, True)
        self.whisker.display_create_document(DOC)
        self.whisker.display_set_background_colour(DOC, bg_col)
        self.whisker.display_blank(DISPLAY)
        self.whisker.display_create_document("junk")
        self.whisker.display_delete_document("junk")
        self.whisker.display_show_document(DISPLAY, DOC)
        with self.whisker.display_cache_wrapper(DOC):
            self.whisker.display_add_obj_text(
                DOC, "_text", (50, 50), "hello there!",
                italic=True)
            self.whisker.display_add_obj_line(
                DOC, "_line", (25, 25), (200, 200), pen)
            self.whisker.display_add_obj_arc(
                DOC, "_arc",
                Rectangle(left=100, top=100, width=200, height=200),
                (25, 25), (200, 200), pen)
            self.whisker.display_add_obj_bezier(
                DOC, "_bezier",
                (100, 100), (150, 100),
                (150, 200), (100, 200),
                pen)
            self.whisker.display_add_obj_chord(
                DOC, "_chord",
                Rectangle(left=300, top=0, width=100, height=100),
                (100, 150), (400, 175),
                pen, brush)
            self.whisker.display_add_obj_ellipse(
                DOC, "_ellipse",
                Rectangle(left=0, top=200, width=200, height=100),
                pen, brush)
            self.whisker.display_add_obj_pie(
                DOC, "_pie",
                Rectangle(left=0, top=300, width=200, height=100),
                (10, 320), (180, 380),
                pen, brush)
            self.whisker.display_add_obj_polygon(
                DOC, "_polygon",
                [(400, 200), (450, 300), (400, 400), (300, 300)],
                pen, brush, alternate=True)
            self.whisker.display_add_obj_rectangle(
                DOC, "_rectangle",
                Rectangle(left=500, top=0, width=200, height=100),
                pen, brush)
            self.whisker.display_add_obj_roundrect(
                DOC, "_roundrect",
                Rectangle(left=500, top=200, width=100, height=200),
                150, 250,
                pen, brush)
            self.whisker.display_add_obj_camcogquadpattern(
                DOC, "_camcogquad",
                (500, 400),
                10, 10,
                [1, 2, 4, 8, 16, 32, 64, 128],
                [255, 254, 253, 252, 251, 250, 249, 248],
                [1, 2, 3, 4, 5, 6, 7, 8],
                [128, 64, 32, 16, 8, 4, 2, 1],
                (255, 0, 0),
                (0, 255, 0),
                (0, 0, 255),
                (255, 0, 255),
                (100, 100, 100))
        self.whisker.display_set_event(DOC, "_polygon", "poly_touched")
        self.whisker.display_clear_event(DOC, "_polygon")
        self.whisker.display_set_event(DOC, "_camcogquad", "play")
        self.whisker.display_set_event(DOC, "_roundrect", "pause")
        self.whisker.display_set_event(DOC, "_rectangle", "stop")
        self.whisker.display_set_obj_event_transparency(DOC, "_rectangle",
                                                        False)
        self.whisker.display_bring_to_front(DOC, "_chord")
        self.whisker.display_send_to_back(DOC, "_chord")
        self.whisker.display_keyboard_events(DOC)
        self.whisker.display_event_coords(False)
        self.whisker.display_set_audio_device(DISPLAY, AUDIO)
        self.whisker.display_add_obj_video(DOC, VIDEO, (600, 0), self.video)
        self.whisker.video_set_volume(DOC, VIDEO, 90)
        self.whisker.video_timestamps(True)
        video_dur_ms = self.whisker.video_get_duration_ms(DOC, VIDEO)
        log.info("video_dur_ms: {}".format(video_dur_ms))
        video_pos_ms = self.whisker.video_get_time_ms(DOC, VIDEO)
        log.info("video_pos_ms: {}".format(video_pos_ms))

    def incoming_event(self, event: str, timestamp: int = None) -> None:
        """
        Responds to incoming events from Whisker.
        """
        print("Event: {e} (timestamp {t})".format(e=event, t=timestamp))
        if event == "EndOfTask":
            # noinspection PyUnresolvedReferences
            reactor.stop()
        elif event == "play":
            self.whisker.video_play(DOC, VIDEO)
        elif event == "pause":
            self.whisker.video_pause(DOC, VIDEO)
        elif event == "stop":
            self.whisker.video_stop(DOC, VIDEO)
        elif event == "back":
            self.whisker.video_seek_relative(DOC, VIDEO, -1000)
        elif event == "forward":
            self.whisker.video_seek_relative(DOC, VIDEO, 1000)
        elif event == "start":
            self.whisker.video_seek_absolute(DOC, VIDEO, 0)


def main() -> None:
    """
    Command-line parser.
    See ``--help`` for details.
    """
    logging.basicConfig()
    logging.getLogger("whisker").setLevel(logging.DEBUG)
    configure_logger_for_colour(logging.getLogger())  # configure root logger
    # print_report_on_all_logs()

    parser = argparse.ArgumentParser(
        description="Test Whisker Twisted client",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument(
        '--server', default='localhost',
        help="Server")
    parser.add_argument(
        '--port', default=DEFAULT_PORT, type=int,
        help="Port")
    parser.add_argument(
        '--display_num', default=DEFAULT_DISPLAY_NUM, type=int,
        help="Display number to use")
    parser.add_argument(
        '--audio_num', default=DEFAULT_AUDIO_NUM, type=int,
        help="Audio device number to use")
    parser.add_argument(
        '--input', default=DEFAULT_INPUT_LINE, type=int,
        help="Input line number to use")
    parser.add_argument(
        '--output', default=DEFAULT_OUTPUT_LINE, type=int,
        help="Output line number to use")
    parser.add_argument(
        '--media_dir', default=DEFAULT_MEDIA_DIR, type=str,
        help="Media directory to use")
    parser.add_argument(
        '--bitmap', default=DEFAULT_BITMAP, type=str,
        help="Bitmap to use")
    parser.add_argument(
        '--video', default=DEFAULT_VIDEO, type=str,
        help="Video to use")
    parser.add_argument(
        '--wav', default=DEFAULT_WAV, type=str,
        help="WAV file to use")
    args = parser.parse_args()

    print("Module run explicitly. Running a Whisker test.")
    w = MyWhiskerTwistedTask(
        display_num=args.display_num,
        audio_num=args.audio_num,
        input_line=args.input,
        output_line=args.output,
        media_dir=args.media_dir,
        bitmap=args.bitmap,
        video=args.video,
        wav=args.wav,
    )
    w.connect(args.server, args.port)
    reactor.run()


if __name__ == '__main__':
    main()

Multithreading

For multithreading we can use Qt (with the PySide bindings). In this approach, the Whisker task runs in separate threads from the UI. This works well, though is not without some complexity. The Qt interface is nice, and can be fairly fancy. You have to be careful with database access if using SQLite (which is not always happy in a multithreaded context).

Verdict for simple uses

Use Twisted and avoid any UI code while the task is running.

Database access

Database backend

There are distinct advantages to making SQLite the default, namely:

• It comes with everything (i.e. no installation required);
• Database can be copied around as single files.

On the downside, it doesn’t cope with multithreading/multiuser access quite as well as “bigger” databases like MySQL.

Users will want simple textfile storage as well.

Front end

The options for SQLite access include direct access:

https://docs.python.org/3.4/library/sqlite3.html

and SQLAlchemy:

http://www.sqlalchemy.org/

Getting fancier, it’s possible to manage database structure migrations with tools like Alembic (for SQLAlchemy), but this may be getting too complicated for the target end user.

However, the other very pleasantly simple front-end is dataset:

https://dataset.readthedocs.org/en/latest/

User interface

A GUI can consume a lot of programmer effort. Let’s keep this minimal or absent as the general rule; for more advanced coding, the coder can do his/her own thing (a suggestion: Qt).

Task configuration

Much of the GUI is usually about configuration. So let’s get rid of all that, because we’re aiming at very simple programming here. Let’s just put config in a simple structure like JSON or YAML, and have the user edit it separately.

JSON

An example program:

#!/usr/bin/env python
# json_example.py

import json

# Demo starting values
# config = {
#     "session": 3,
#     "stimulus_order": [3, 2, 1],
#     "subject": "S1"
# }

# or, equivalently:
config = dict(
    session=3,
    stimulus_order=[3, 2, 1],
    subject='S1'
)

# Save
with open("json_config_demo.json", "w") as outfile:
    json.dump(config, outfile,
              sort_keys=True, indent=4, separators=(',', ': '))

# Load
with open("json_config_demo.json") as infile:
    config = json.load(infile)

The JSON looks like:

{
    "session": 3,
    "stimulus_order": [
        3,
        2,
        1
    ],
    "subject": "S1"
}

YAML with attrdict

This can be a bit fancier in terms of the object structure it can represent, a bit cleaner in terms of the simplicity of the config file, and safer in terms of security from dodgy config files.

Using an AttrDict allows a cleaner syntax for reading/writing the Python object.

#!/usr/bin/env python
# yaml_example.py

from attrdict import AttrDict
import yaml

# Demo starting values; lots of ways of creating an AttrDict
config = AttrDict()
config.session = 3
config.stimulus_order = [3, 2, 1]
config.subject = "S1"

# Save
with open("yaml_config_demo.yaml", "w") as outfile:
    outfile.write(yaml.dump(config.copy()))  # write a dict, not an AttrDict

# Load
with open("yaml_config_demo.yaml") as infile:
    config = AttrDict(yaml.safe_load(infile))

The YAML looks like this:

session: 3
stimulus_order: [3, 2, 1]
subject: S1

A quick YAML tutorial

A key:value pair looks like:

key: value

A list looks like:

list:
    - value1
    - value2
    # ...

A dictionary looks like:

dict:
    key1: value1
    key2: value2
    # ...

Verdict for simple uses

Use YAML with AttrDict.

Package distribution

This should be via PyPI, so users can just do:

pip3 install whisker

from whisker import *  # but better actual symbols than "*"!

A complete simple task

Having done pip install whisker, you should be able to do this:

# demo_config.yaml

# The URL is of the SQLAlchemy type.
# http://docs.sqlalchemy.org/en/latest/core/engines.html#database-urls
# Here we'll use a SQLite file. For bigger purposes, consider MySQL etc.

database_url: sqlite:///demo_task_results.db

server: localhost
# server: wombatvmxp

# Let's include a comma to check the CSV output
subject: Mr Potato, Head
session: 5

#!/usr/bin/env python
# demo_simple_whisker_client.py

# =============================================================================
# Imports, and configure logging
# =============================================================================

import logging
from attrdict import AttrDict
from datetime import datetime
from twisted.internet import reactor
from whisker.logging import configure_logger_for_colour
from whisker.constants import DEFAULT_PORT
from whisker.convenience import (load_config_or_die,
                                 connect_to_db_using_attrdict,
                                 insert_and_set_id,
                                 ask_user,
                                 save_data)
from whisker.twistedclient import WhiskerTwistedTask

log = logging.getLogger(__name__)
configure_logger_for_colour(log)

log.setLevel(logging.DEBUG)  # debug-level logging for this file...
logging.getLogger("whisker").setLevel(logging.DEBUG)  # ... and for Whisker

# =============================================================================
# Constants
# =============================================================================

TASKNAME_SHORT = "countpings"  # no spaces; we'll use it in a filename
TASKNAME_LONG = "Ping Counting Task"

# Our tables. They will be autocreated. (NOTE: do not store separate copies of
# table objects, as they can get out of sync as new columns area created.)
SESSION_TABLE = 'session'
TRIAL_TABLE = 'trial'
SUMMARY_TABLE = 'summary'


# =============================================================================
# The task itself
# =============================================================================

class MyWhiskerTwistedTask(WhiskerTwistedTask):
    def __init__(self, config, db, session):
        """Here, we initialize the task, and store any relevant variables."""
        super().__init__()  # call base class init
        self.config = config
        self.db = db
        self.session = session
        self.trial_num = 0

    def fully_connected(self):
        """At this point, we are fully connected to the Whisker server."""
        print("Task running.")
        period_ms = 1000
        self.whisker.timestamps(True)
        self.whisker.report_name(TASKNAME_LONG)
        self.whisker.get_network_latency_ms()
        self.whisker.timer_set_event("TimerFired", period_ms,
                                     self.session.num_pings - 1)
        self.whisker.timer_set_event("EndOfTask",
                                     period_ms * (self.session.num_pings + 1))

    def incoming_event(self, event, timestamp=None):
        """An event has arrived from the Whisker server."""
        # timestamp is the Whisker server's clock time in ms; we want real time
        now = datetime.utcnow()
        print("Event: {e} (timestamp {t}, real time {n})".format(
            e=event, t=timestamp, n=now.isoformat()))
        # We could do lots of things at this point. But let's keep it simple:
        if event == "EndOfTask":
            # noinspection PyUnresolvedReferences
            reactor.stop()  # stops Twisted and thus network processing
        else:
            trial = AttrDict(
                session_id=self.session.id,  # important foreign key
                trial_num=self.trial_num,
                event="Ping!",
                received=True,  # now we're just making things up...
                when=now,
            )
            insert_and_set_id(self.db[TRIAL_TABLE], trial)  # save to database
            self.trial_num += 1
            log.info("{} pings received so far".format(self.trial_num))


# =============================================================================
# Main execution sequence
# =============================================================================

def main():
    # -------------------------------------------------------------------------
    # Load config; establish database connection; ask the user for anything else
    # -------------------------------------------------------------------------

    log.info("Asking user for config filename")
    config = load_config_or_die(
        mandatory=['database_url'],
        defaults=dict(server='localhost', port=DEFAULT_PORT),
        log_config=True  # send to console (beware security of database URLs)
    )
    db = connect_to_db_using_attrdict(config.database_url)

    # Any additional user input required?
    num_pings = ask_user("Number of pings", default=10, type=int, min=1)
    ask_user("Irrelevant: Heads or tails", default='H', options=['H', 'T'])

    # -------------------------------------------------------------------------
    # Set up task and go
    # -------------------------------------------------------------------------

    session = AttrDict(start=datetime.now(),
                       subject=config.subject,
                       session=config.session,
                       num_pings=num_pings)
    insert_and_set_id(db[SESSION_TABLE], session)  # save to database
    log.info("Off we go...")
    task = MyWhiskerTwistedTask(config, db, session)
    task.connect(config.server, config.port)
    # noinspection PyUnresolvedReferences
    reactor.run()  # starts Twisted and thus network processing
    log.info("Finished.")

    # -------------------------------------------------------------------------
    # Done. Calculate summaries. Save data from this session to new CSV files.
    # -------------------------------------------------------------------------

    # Retrieve all our trials. (There may also be many others in the database.)
    # NOTE that find() returns an iterator (you get to iterate through it ONCE).
    # Since we want to use this more than once (below), use a list.
    trials = list(db[TRIAL_TABLE].find(session_id=session.id))

    # Calculate some summary measures
    summary = AttrDict(
        session_id=session.id,  # foreign key
        n_pings_received=sum(t.received for t in trials)
    )
    insert_and_set_id(db[SUMMARY_TABLE], summary)  # save to database

    # Save data. (Since the session and summary objects are single objects, we
    # encapsulate them in a list.)
    save_data("session", [session], timestamp=session.start,
              taskname=TASKNAME_SHORT)
    save_data("trial", trials, timestamp=session.start,
              taskname=TASKNAME_SHORT)
    save_data("summary", [summary], timestamp=session.start,
              taskname=TASKNAME_SHORT)


if __name__ == '__main__':
    main()

Change history

2016-02-10

• moved to package format

v0.2.0: 2016-02-25

• colourlog renamed logsupport.

v0.3.5: 2016-11-25

• Python type hints.
• Write exit_on_exception exceptions to log, not via print().
• whisker.qtclient.WhiskerOwner offers new pingack_received signal.

v0.3.6: 2016-12-01

• Changed from PySide to PyQt5 (fewer bugs).

v0.3.6: 2017-03-23

• Removed annotations from alembic.py; alembic uses inspect.getargspec(), which chokes with ValueError: Function has keyword-only arguments or annotations....
• Support PyQt 5.8, including removing calls to QHeaderView.setClickable, which has gone: https://doc.qt.io/qt-5/qheaderview.html

v0.1.10: 2016-06-22

• Updates for Starfeeder.

v0.1.11: 2016-06-23

• Further updates for Starfeeder; supporting structured handling of ClientMessage.

v1.0: 2017

• Update for new cardinal_pythonlib.

v1.0.3: 2017-09-07

• use SQLAlchemy 1.2 pool_pre_ping feature

v1.0.4: 2018-09-18

• PyQt5 and Twisted added as requirements
• Sphinx docs
• updated signal debugging for PyQt5 in debug_qt.py
• general tidy
• ValidationError removed from whisker.qt; is in whisker.exceptions (previously duplicated)
• additional randomization functions

v1.1.0: 2018-09-21 to 2018-09-22

• updated required version to cardinal_pythonlib>=1.0.26 in view of bugfix there
• there were two classes named WhiskerTask; renamed one to WhiskerTwistedTask and the other to WhiskerQtTask. Also renamed Whisker to WhiskerRawSocketClient for clarity.
• public docs at https://whiskerpythonclient.readthedocs.io/
• no [source] links; see https://github.com/rtfd/readthedocs.org/issues/2139
  • removed typing==3.5.2.2 dependency and restricted to Python 3.5 (as per cardinal_pythonlib) to see if that fixed the autodoc/RTD problem; no
  • no help from changing Sphinx theme either
  • no help from changing sys.path from conf.py
  • see https://github.com/rtfd/readthedocs.org/issues/2139
  • tried /readthedocs.yml as per that and https://docs.readthedocs.io/en/latest/yaml-config.html
• renamed layer_index to layer_key and changed its hint in whisker.random.ShuffleLayerMethod.

v1.2.0 to 1.3.0: 2020-02-09

• Added whisker.__version__.
• whisker.convenience.load_config_or_die() has a new config_filename argument.
• Requirement for Python 3.6+. (Because of cardinal_pythonlib.)
• New function whisker.convenience.update_record().

Things to do

Known problems

• 2016-11-25: Syntax check fails because PyCharm 2016.3 type hints go wrong for generators:
  • https://youtrack.jetbrains.com/issue/PY-20657#tab=Linked%20Issues
  • https://youtrack.jetbrains.com/issue/PY-20709

Automatic documentation of source code

whisker.api

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Whisker API constants and functions.

class whisker.api.Brush(colour: Tuple[int, int, int] = (255, 255, 255), bg_colour: Tuple[int, int, int] = (0, 0, 0), opaque: bool = True, style: whisker.api.BrushStyle = <BrushStyle.solid: 1>, hatch_style: whisker.api.BrushHatchStyle = <BrushHatchStyle.cross: 4>)

Represents a Whisker Brush graphics object.

class whisker.api.BrushHatchStyle

An enumeration.

class whisker.api.BrushStyle

An enumeration.

class whisker.api.DocEventType

An enumeration.

class whisker.api.HorizontalAlign

An enumeration.

class whisker.api.KeyEventType

An enumeration.

class whisker.api.LineEventType

An enumeration.

class whisker.api.Pen(width: int = 1, colour: Tuple[int, int, int] = (255, 255, 255), style: whisker.api.PenStyle = <PenStyle.solid: 0>)

Represents a Whisker Pen graphics object.

class whisker.api.PenStyle

An enumeration.

class whisker.api.Rectangle(left: int, top: int, width: int = None, height: int = None, right: int = None, bottom: int = None)

Represents a Whisker Rectangle graphics object.

The origin is at the top left, as per Whisker.

class whisker.api.ResetState

An enumeration.

class whisker.api.SafetyState

An enumeration.

class whisker.api.TextHorizontalAlign

An enumeration.

class whisker.api.TextVerticalAlign

An enumeration.

class whisker.api.ToneType

An enumeration.

class whisker.api.VerticalAlign

An enumeration.

class whisker.api.VideoPlayMode

An enumeration.

class whisker.api.WhiskerApi(whisker_immsend_get_reply_fn: Callable[..., str], sysevent_prefix: str = 'sys_', **kwargs)

Whisker API handler.

Distinct from any particular network/threading model, so all can use it (e.g. by inheritance), but hooks in to whichever you choose.

Parameters:

• whisker_immsend_get_reply_fn – A function that must take arguments *args, join stringified versions of them using a space as the separator, and send them to the Whisker server via the immediate socket, returning the string that the server sent back.
• sysevent_prefix – string to prefix system events with
• kwargs – used in case of mixin inheritance

audio_get_sound_duration_ms(device: str, sound: str) → Optional[int]

Whisker command: get the duration of a loaded sound.

Parameters:

• device – audio device number/name
• sound – sound buffer name

Returns:

duration in ms, or None upon failure

audio_load_tone(device: str, buffer_: str, frequency_hz: int, tone_type: whisker.api.ToneType, duration_ms: int) → bool

Whisker command: synthesize a tone ready to be played.

Parameters:

• device – audio device number/name
• buffer – buffer name
• frequency_hz – frequency (Hz)
• tone_type – tone type (e.g. sine, square, …)
• duration_ms – duration (ms)

Returns:

success?

audio_load_wav(device: str, sound: str, filename: str) → bool

Whisker command: load a WAV file, ready for playing.

Parameters:

• device – audio device number/name
• sound – sound buffer name to create
• filename – WAV filename on the server

Returns:

success?

audio_play_sound(device: str, sound: str, loop: bool = False) → bool

Whisker command: play a sound from a named buffer.

Parameters:

• device – audio device number/name
• sound – sound buffer name to play
• loop – play it on loop?

Returns:

success?

audio_play_wav(device: str, filename: str) → bool

Whisker command: play a WAV file on an audio device.

Parameters:

• device – audio device number/name
• filename – WAV filename on the server

Returns:

success?

audio_set_alias(from_: str, to: str) → bool

Whisker command: adds an alias for an audio device.

Parameters:

• from – audio device name/number
• to – new alias to apply

Returns:

success?

audio_set_sound_volume(device: str, sound: str, volume: int) → bool

Whisker command: set the volume for a sound.

Parameters:

• device – audio device number/name
• sound – sound buffer name to alter
• volume – from 0 (100 dB attentuation) to 100 (full volume)

Returns:

success

audio_silence_all_devices() → bool

Whisker comand: silence all audio devices.

audio_silence_device(device: str) → bool

Whisker command: silence an audio device.

audio_stop_sound(device: str, sound: str) → bool

Whisker command: stop playing a sound.

Parameters:

• device – audio device number/name
• sound – sound buffer name to stop

Returns:

success?

audio_unload_all(device: str) → bool

Whisker command: unload all sounds from an audio device.

audio_unload_sound(device: str, sound: str) → bool

Whisker command: unload a sound from an audio device.

Parameters:

• device – audio device number/name
• sound – sound buffer name to unload

Returns:

success?

authenticate_get_challenge(package: str, client_name: str) → Optional[str]

Whisker command: ask the server for an authentication challenge, using a particular software package name and client name.

authenticate_provide_response(response: str) → bool

Whisker command: respond to the server’s authentication challenge with an appropriate cryptographic answer.

broadcast(*args) → bool

Broadcasts a message to all other clients of this Whisker server that are listening.

call_after_delay(delay_ms: int, callback: Callable[..., None], args: List[Any] = None, kwargs: List[Any] = None, event: str = '') → None

Sets a Whisker timer so we’ll be called back after a delay, and then call a Python function.

Parameters:

• delay_ms – delay in ms
• callback – function to call back
• args – positional arguments to callback
• kwargs – keyword arguments to callback
• event – optional Whisker event name (a default will be usd if none is provided)

call_on_event(event: str, callback: Callable[..., None], args: List[Any] = None, kwargs: List[Any] = None, swallow_event: bool = False) → None

Tells the callback handler to call a function whenever we receive a particular event from Whisker.

Parameters:

• event – Whisker event name
• callback – function to call back
• args – positional arguments to callback
• kwargs – keyword arguments to callback
• swallow_event – make the API swallow the event, so it’s not seen by our client task?

claim_audio(number: int = None, group: str = None, device: str = None, alias: str = '') → bool

Whisker command: claim an audio device.

Parameters:

• number – device number, for numerical method
• group – group name, for named method
• device – device name within group, for named method
• alias – alias to apply

Returns:

success?

claim_display(number: int = None, group: str = None, device: str = None, alias: str = '') → bool

Whisker command: claim a dsplay device

Parameters:

• number – device number, for numerical method
• group – group name, for named method
• device – device name within group, for named method
• alias – alias to apply

Returns:

success?

Autocreating debug views is not supported (see C++ WhiskerClientLib).

claim_group(group: str, prefix: str = '', suffix: str = '') → bool

Whisker command: claim a device group

Parameters:

• group – group name
• prefix – prefix to apply to all device names in the group, when creating the alias
• suffix – suffix to apply to all device names in the group, when creating the alias

Returns:

success?

claim_line(number: int = None, group: str = None, device: str = None, output: bool = False, reset_state: whisker.api.ResetState = <ResetState.leave: 3>, alias: str = '') → bool

Whisker command: claim a digital I/O line.

Parameters:

• number – line number, for numerical method
• group – group name, for named method
• device – device name, for named method
• output – output line? (If not: input line.)
• reset_state – if/how to reset this line when the client disconnects
• alias – alias to apply

Returns:

success?

clear_all_callbacks() → None

Cancel all callbacks.

clear_event_callback(event: str, callback: Callable[..., None] = None) → None

Cancels a callback.

Parameters:

• event – Whisker event name
• callback – callback function to remove from callback list

command(*args) → bool

Return a boolean answer from an immediate socket command.

command_exc(*args) → None

Send a command; if it fails, raise WhiskerCommandFailed.

debug_callbacks() → None

Display our callback settings to the Python log.

display_add_obj(doc: str, obj: str, obj_type: str, *parameters) → bool

Whisker command: add a display object to a document.

This command has more specialized and helpful versions; don’t use it directly.

display_add_obj_arc(doc: str, obj: str, rect: whisker.api.Rectangle, start, end, pen) → bool

Whisker command: adds an Arc display object. See Whisker help for details.

The arc fits into the rect.

display_add_obj_bezier(doc: str, obj: str, start: Tuple[int, int], control1: Tuple[int, int], control2: Tuple[int, int], end: Tuple[int, int], pen: whisker.api.Pen) → bool

Whisker command: adds a Bezier display object. See Whisker help for details.

display_add_obj_bitmap(doc: str, obj: str, pos: Tuple[int, int], filename: str, stretch: bool = False, height: int = -1, width: int = -1, valign: whisker.api.VerticalAlign = <VerticalAlign.top: 0>, halign: whisker.api.HorizontalAlign = <HorizontalAlign.left: 0>) → bool

Whisker command: adds a Bitmap display object. See Whisker help for details.

display_add_obj_camcogquadpattern(doc: str, obj: str, pos: Tuple[int, int], pixel_width: int, pixel_height: int, top_left_patterns: List[int], top_right_patterns: List[int], bottom_left_patterns: List[int], bottom_right_patterns: List[int], top_left_colour: Tuple[int, int, int], top_right_colour: Tuple[int, int, int], bottom_left_colour: Tuple[int, int, int], bottom_right_colour: Tuple[int, int, int], bg_colour: Tuple[int, int, int]) → bool

Whisker command: adds a CamcogQuadPattern display object. See Whisker help for details.

Patterns are lists (of length 8) of bytes.

display_add_obj_chord(doc: str, obj: str, rect: whisker.api.Rectangle, line_start: Tuple[int, int], line_end: Tuple[int, int], pen: whisker.api.Pen, brush: whisker.api.Brush) → bool

Whisker command: adds a Text display object. See Whisker help for details.

The chord is the intersection of an ellipse (defined by the rect) and a line that intersects it.

display_add_obj_ellipse(doc: str, obj: str, rect: whisker.api.Rectangle, pen: whisker.api.Pen, brush: whisker.api.Brush) → bool

Whisker command: adds an Ellipse display object. See Whisker help for details.

The ellipse fits into the rectangle (and its centre is at the centre of the rectangle).

display_add_obj_line(doc: str, obj: str, start: Tuple[int, int], end: Tuple[int, int], pen: whisker.api.Pen) → bool

Whisker command: adds a Line display object. See Whisker help for details.

display_add_obj_pie(doc: str, obj: str, rect: whisker.api.Rectangle, arc_start: Tuple[int, int], arc_end: Tuple[int, int], pen: whisker.api.Pen, brush: whisker.api.Brush) → bool

Whisker command: adds a Pie display object. See Whisker help for details.

display_add_obj_polygon(doc: str, obj: str, points: List[Tuple[int, int]], pen: whisker.api.Pen, brush: whisker.api.Brush, alternate: bool = False) → bool

Whisker command: adds a Polygon display object. See Whisker help for details.

display_add_obj_rectangle(doc: str, obj: str, rect: whisker.api.Rectangle, pen: whisker.api.Pen, brush: whisker.api.Brush) → bool

Whisker command: adds a Rectangle display object. See Whisker help for details.

display_add_obj_roundrect(doc: str, obj: str, rect: whisker.api.Rectangle, ellipse_height: int, ellipse_width: int, pen: whisker.api.Pen, brush: whisker.api.Brush) → bool

Whisker command: adds a RoundRect display object. See Whisker help for details.

display_add_obj_text(doc: str, obj: str, pos: Tuple[int, int], text: str, height: int = 0, font: str = '', italic: bool = False, underline: bool = False, weight: int = 0, colour: Tuple[int, int, int] = (255, 255, 255), opaque: bool = False, bg_colour: Tuple[int, int, int] = (0, 0, 0), valign: whisker.api.TextVerticalAlign = <TextVerticalAlign.top: 0>, halign: whisker.api.TextHorizontalAlign = <TextHorizontalAlign.left: 0>) → bool

Whisker command: adds a Text display object. See Whisker help for details.

display_add_obj_video(doc: str, video: str, pos: Tuple[int, int], filename: str, loop: bool = False, playmode: whisker.api.VideoPlayMode = <VideoPlayMode.wait: 0>, width: int = -1, height: int = -1, play_audio: bool = True, valign: whisker.api.VerticalAlign = <VerticalAlign.top: 0>, halign: whisker.api.HorizontalAlign = <HorizontalAlign.left: 0>, bg_colour: Tuple[int, int, int] = (0, 0, 0)) → bool

Whisker command: adds a Video display object. See Whisker help (http://www.whiskercontrol.com) for details.

display_blank(device: str) → bool

Whisker command: blank a display device (remove any document).

display_bring_to_front(doc: str, obj: str) → bool

Whisker command: bring an object to the front of its document (so it appears in front of, or on top of, other objects).

display_cache_changes(doc: str) → bool

Whisker command: start caching changes to a document (to reduce flicker). When you’ve made all the changes, call display_show_changes().

display_cache_wrapper(doc: str) → Generator[[None, None], None]

Context manager to wrap display calls in a “cache changes”, [do stuff], “show changes” sequence.

Use like:

with something.display_cache_wrapper(doc):
    # do some display-related things

display_clear_background_event(doc: str, event_type: whisker.api.DocEventType = <DocEventType.touch_down: 5>) → bool

Clears a document event

Parameters:

• doc – document name
• event_type – touched, touch released, mouse click, etc.

Returns:

success?

display_clear_event(doc: str, obj: str, event_type: whisker.api.DocEventType = <DocEventType.touch_down: 5>) → bool

Whisker command: clears an event from a display object.

Parameters:

• doc – document name
• obj – object name
• event_type – event type (e.g. touched, released, mouse click, …)

Returns:

success?

display_create_device(name: str, resize: bool = True, directdraw: bool = True, rectangle: whisker.api.Rectangle = None, debug_touches: bool = False) → bool

Whisker command: create a new display device (window).

Parameters:

• name – display name
• resize – may the window be resized?
• directdraw – use DirectDraw for speed?
• rectangle – initial size of the window
• debug_touches – if True, touches will be shown on all documents displayed on the device

Returns:

success?

display_create_document(doc: str) → bool

Whisker command: create a display document.

display_delete_device(device: str) → bool

Whisker command: deletes a display device that you’ve created with display_create_device().

display_delete_document(doc: str) → bool

Whisker command: delete a display document.

display_delete_obj(doc: str, obj: str) → bool

Whisker command: delete a display object from a document.

display_event_coords(on: bool) → bool

Whisker command: enable/disable x/y coordinates being sent with relevant touchscreen/mouse events.

display_get_document_size(doc: str) → Optional[Tuple[int, int]]

Whisker command: get the logical size of a display document.

Returns a (width, height) tuple in pixels, or None.

display_get_object_extent(doc: str, obj: str) → Optional[whisker.api.Rectangle]

Whisker command: get the rectangle describing the extend of a display object (in logical coordinates).

Returns a Rectangle, or None.

display_get_size(device: str) → Optional[Tuple[int, int]]

Whisker command: get the size of a display device.

Parameters:device – display device number/name Returns:(width, height) in pixels, or None Return type:tuple

display_keyboard_events(doc: str, key_event_type: whisker.api.KeyEventType = <KeyEventType.down: 1>) → bool

Whisker command: enable/disable keyboard events.

Parameters:

• doc – display document
• key_event_type – choose “key down” (pressed), “key up” (released), both, or no events.

Returns:

success?

display_scale_documents(device: str, scale: bool = True) → bool

Whisker command: set document scaling for a display device.

display_send_to_back(doc: str, obj: str) → bool

Whisker command: send an object to the back of its document (so it appears behind, or underneath, other objects).

display_set_alias(from_: str, to: str) → bool

Whisker command: adds an alias for a display device.

Parameters:

• from – audio device name/number
• to – new alias to apply

Returns:

success?

display_set_audio_device(display_device: str, audio_device: str) → bool

Whisker command: sets the audio device associated with a display device, for video playback.

Parameters:

• display_device – display device number/name
• audio_device – audio device number/name

Returns:

success?

display_set_background_colour(doc: str, colour: Tuple[int, int, int] = (0, 0, 0)) → bool

Whisker command: set the background colour of a display document.

display_set_background_event(doc: str, event: str, event_type: whisker.api.DocEventType = <DocEventType.touch_down: 5>) → bool

Whisker command: set an event for when the background of a document if douched (or released).

Parameters:

• doc – document name
• event – event name to create
• event_type – touched, touch released, mouse click, etc.

Returns:

success?

display_set_document_size(doc: str, width: int, height: int) → bool

Whisker command: set the logical size in pixels of a display document.

display_set_event(doc: str, obj: str, event: str, event_type: whisker.api.DocEventType = <DocEventType.touch_down: 5>) → bool

Whisker command: sets an event (e.g. responding to touchscreen or mouse events) on a display object within a document.

Parameters:

• doc – document name
• obj – object name
• event – event to set
• event_type – event type (e.g. touched, released, mouse click, …)

Returns:

success?

display_set_obj_event_transparency(doc: str, obj: str, transparent: bool = False) → bool

Whisker command: set object event transparency. Transparency means: an object might respond (or not) to an event, but will it also let objects behind it respond too?

Parameters:

• doc – document name
• obj – object name
• transparent – transparent?

Returns:

success?

display_show_changes(doc: str) → bool

Whisker command: show any pending changes (see display_cache_changes()).

display_show_document(device: str, doc: str) → bool

Whisker command: show a document on a display device.

flash_line_ping_pong(line: str, on_now: bool, timing_sequence: List[int]) → None

Internal function used by flash_line_pulses().

Parameters:

• line – Whisker line number/name
• on_now – switch it on now (otherwise off)?
• timing_sequence – sequence of [off]/on/off/… times still to do, in ms

flash_line_pulses(line: str, count: int, on_ms: int, off_ms: int, on_at_rest: bool = False) → int

Flash a line for a specified number of pulses.

Parameters:

• line – Whisker line number/name
• count – number of flashes
• on_ms – time spent in the “on” phase of the cycle, in ms
• off_ms – time spent in the “off” phase of the cycle, in ms
• on_at_rest – is the line on at rest, so we count “off flashes”? The default, with this False, is to assume the line is off at rest and we count “on” flashes.

Returns:

the total duration of the flashing sequence, in ms

get_client_number() → int

Whisker command: get the client number for our client.

get_command_boolean(*args) → bool

Return a boolean answer from an immediate socket command.

get_network_latency_ms() → Optional[int]

Whisker command: test the network latency.

get_new_sysevent(*args) → str

Make a new system event name.

get_response(*args) → str

Return the answer from an immediate socket command.

get_response_with_timestamp(*args) → Tuple[str, Optional[int]]

Return the answer from an immediate socket command, with a timestamp.

get_server_time_ms() → int

Whisker command: get the time on the server, from its millisecond clock.

get_server_version() → str

Whisker command: get the server version, as a string.

get_server_version_numeric() → float

Whisker command: get the server version, as a float.

line_clear_all_events() → bool

Whisker command: clear (cancel) all line events.

line_clear_event(event: str) → bool

Whisker command: clear (cancel) a line event or events by name.

line_clear_event_by_line(line: str, event_type: whisker.api.LineEventType) → bool

Whisker command: clear (cancel) a line event by line number/name and transition type.

line_clear_safety_timer(line: str) → bool

Whisker command: clears a safety timer for a line.

line_off(line: str) → None

Turns a line off.

line_on(line: str) → None

Turns a line on.

line_read_state(line: str) → Optional[bool]

Whisker command: reads the state of a digital I/O line.

Parameters:line – line number/name Returns:True for on, False for off, None for “problem”

line_set_alias(line: str, alias: str) → bool

Whisker command: adds an alias for a line.

Parameters:

• line – line name/number
• alias – new alias to apply

Returns:

success?

line_set_event(line: str, event: str, event_type: whisker.api.LineEventType = <LineEventType.on: 1>) → bool

Whisker command: sets an event for a digital I/O line transition.

Parameters:

• line – line number/name
• event – event name to create
• event_type – fire event upon on transitions, off transitions, or both?

Returns:

success?

line_set_safety_timer(line: str, time_ms: int, safety_state: whisker.api.SafetyState) → bool

Whisker command: set a safety timer on a line. If the client doesn’t use this line for a while, the safety system will kick in.

Parameters:

• line – line number/name
• time_ms – how long after the client’s last action with this line before we consider the line to be being neglected (ms)?
• safety_state – turn it off or on after it’s been neglected?

Returns:

success?

line_set_state(line: str, on: bool) → bool

Whisker command: turns a digital output line on/off.

log_close() → bool

Whisker command: close the log.

log_open(filename: str) → bool

Whisker command: open a log file.

log_pause() → bool

Whisker command: pause logging.

log_resume() → bool

Whisker command: resume logging.

log_set_options(events: bool = True, key_events: bool = True, client_client: bool = True, comms: bool = False, signature: bool = True) → bool

Whisker command: set log file options.

Parameters:

• events – log events?
• key_events – log keyboard events?
• client_client – log client/client communications?
• comms – log comms?
• signature – sign the log digitally?

log_write(*args) → bool

Whisker command: write a message to the log.

permit_client_messages(permit: bool) → bool

Whisker command: enable/disable client messages.

ping() → bool

Whisker command: ping the server.

process_backend_event(event: str) → bool

Process an through the system event handler.

Returns True if the backend API has dealt with the event and it doesn’t need to go to the main behavioural task.

relinquish_all_audio() → bool

Whisker command: relinquish all audio devices.

relinquish_all_displays() → bool

Whisker command: relinquish all display devices.

relinquish_all_lines() → bool

Whisker command: relinquish all lines.

report_comment(*args) → bool

Whisker command: tell the server our comment.

report_name(*args) → bool

Whisker command: tell the server our name.

report_status(*args) → bool

Whisker command: tell the server our status.

reset_clock() → bool

Whisker command: reset server clock.

send_after_delay(delay_ms: int, msg: str, event: str = '') → None

Sets a Whisker timer so we’ll be called back after a delay, and then send the command.

Parameters:

• delay_ms – delay in ms
• msg – message to send after the delay
• event – optional Whisker event name (a default will be usd if none is provided)

send_to_client(client_num: int, *args) → bool

Whisker command: send a message to another client.

set_media_directory(directory: str) → bool

Whisker command: set the server’s media directory.

shutdown() → bool

Whisker command: shutdown.

timer_clear_all_events() → bool

Whisker command: clear (cancel) all timers.

timer_clear_event(event: str) → bool

Whisker command: clear (cancel) a named timer.

timer_set_event(event: str, duration_ms: int, reload_count: int = 0) → bool

Whisker command: set a timer.

Parameters:

• event – event name
• duration_ms – timer duration (ms)
• reload_count – 0 for no reloads, a positive integer for a specified number of reloads, -1 for infinite reloads.

Returns:

success?

timestamps(on: bool) → bool

Whisker command: turn timestamps on/off.

video_get_duration_ms(doc: str, video: str) → Optional[int]

Whisker command: gets a video’s duration.

Parameters:

• doc – display document name
• video – video name

Returns:

duration in ms, or None

video_get_time_ms(doc: str, video: str) → Optional[int]

Whisker command: gets the current video time (since the start).

Parameters:

• doc – display document name
• video – video name

Returns:

time in ms, or None

video_pause(doc: str, video: str) → bool

Whisker command: pauses a video.

video_play(doc: str, video: str) → bool

Whisker command: plays a video.

video_seek_absolute(doc: str, video: str, time_ms: int) → bool

Whisker command: execute a “seek” command on a video, using absolute time (milliseconds since the video’s start).

video_seek_relative(doc: str, video: str, time_ms: int) → bool

Whisker command: execute a “seek” command on a video, using relative time (e.g. +5000 ms for 5 seconds forwards; -5000 ms for 5 seconds backwards).

video_set_volume(doc: str, video: str, volume: int) → bool

Parameters:

• doc – display document name
• video – video name
• volume – from 0 (minimum) to 100 (full)

Returns:

success?

video_stop(doc: str, video: str) → bool

Whisker command: stops a video.

video_timestamps(on: bool) → bool

Whisker command: enables/disables the sending of video times (in ms) being sent along with “video touched” events.

whisker.api.assert_ducktype_colour(colour: Any) → None

If the parameter does not pass the test is_ducktype_colour(), raise ValueError.

whisker.api.assert_ducktype_pos(pos: Any) → None

If the parameter does not pass the test is_ducktype_pos(), raise ValueError.

whisker.api.is_ducktype_colour(colour: Any) → bool

Does the parameter look like a colour (red, green blue) tuple, with values for each in the range 0-255?

whisker.api.is_ducktype_int(x: Any) → bool

Does the parameter look like an integer?

whisker.api.is_ducktype_nonnegative_int(x: Any) → bool

Does the parameter look like an integer that is 0 or greater?

whisker.api.is_ducktype_pos(pos: Any) → bool

Does the parameter look like a position (x, y) tuple, with values that can be made numeric?

whisker.api.min_to_ms(time_minutes: float) → int

Converts minutes to milliseconds.

whisker.api.msg_from_args(*args) → str

Converts its arguments, which may be of any type, to strings (via str()). Then joins the truthy ones them with spaces to make a command.

whisker.api.on_off_to_boolean(msg: str) → bool

Convert’s Whisker’s on/off value strings to boolean.

whisker.api.quote(string: str) → str

Quotes a string. Suboptimal: doesn’t escape quotes.

whisker.api.s_to_ms(time_seconds: float) → int

Converts seconds to milliseconds.

whisker.api.split_timestamp(msg: str) → Tuple[str, Optional[int]]

Splits a Whisker message into the message proper and a server timestamp.

Parameters:msg – a Whisker message that potentially has a timestamp Returns:(msg, timestamp) where timestamp may be None Return type:tuple

whisker.callback

whisker/callback.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Classes to implement callbacks via the Whisker API.

class whisker.callback.CallbackDefinition(event: str, callback: Callable[..., Any], args: List[Any] = None, kwargs: Dict[str, Any] = None, target_n_calls: int = 0, swallow_event: bool = False)

Event callback handler. Distinct from any particular network/threading model, so all can use it.

Parameters:

• event – Whisker event name
• callback – user-supplied callback function
• args – positional arguments to callback
• kwargs – keyword arguments to callback
• target_n_calls – 0 to keep calling indefinitely, os a positive integer to make that many calls (upon receiving the relevant event) but then to forget about the callback and mark it as defunct
• swallow_event – make the API swallow the event, so it’s not seen by our client task?

call() → None

Calls the callback function.

is_defunct() → bool

Is this callback defunct, by virtue of having been called already as many times as the user asked?

class whisker.callback.CallbackHandler

Implements callbacks based on Whisker events.

add(target_n_calls: int, event: str, callback: Callable[..., Any], args: List[Any] = None, kwargs: Dict[str, Any] = None, swallow_event: bool = True) → None

Adds a callback to the handler.

See CallbackDefinition.

add_persistent(event: str, callback: Callable[..., Any], args: List[Any] = None, kwargs: Dict[str, Any] = None, swallow_event: bool = True) → None

Adds a persistent callback for the specified event.

add_single(event: str, callback: Callable[..., Any], args: List[Any] = None, kwargs: Dict[str, Any] = None, swallow_event: bool = True) → None

Adds a single-shot callback for the specified event.

clear() → None

Removes all callbacks.

debug() → None

Write a description of our callbacks to the debug log.

process_event(event: str) → Tuple[int, bool]

Calls any callbacks registered for the event. Returns the number of callbacks called.

remove(event: str, callback: Callable[..., None] = None) → None

Removes a callback (either by event/callback pair, or all callbacks for an event.

whisker.constants

whisker/constants.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Basic constants.

whisker.convenience

whisker/convenience.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Convenience functions for simple Whisker clients.

whisker.convenience.ask_user(prompt: str, default: Any = None, type: Type[Any] = <class 'str'>, min: Any = None, max: Any = None, options: List[Any] = None, allow_none: bool = True) → Any

Prompts the user via the command line, optionally with a default, range or set of options. Asks for user input. Coerces the return type.

Parameters:

• prompt – prompt to display
• default – default value if the user just presses Enter
• type – type to coerce return value to (default str)
• min – if not None, enforce this minimum value
• max – if not None, enforce this maximum value
• options – permitted options (if this is truthy and the user enters an option that’s not among then, raise ValueError)
• allow_none – permit the return of None values (otherwise, if a None value would be returned, raise ValueError)

Returns:

user-supplied value

whisker.convenience.connect_to_db_using_attrdict(database_url: str, show_url: bool = False, engine_kwargs: Dict[str, Any] = None)

Connects to a dataset database, and uses AttrDict as the row type, so AttrDict objects come back out again.

whisker.convenience.insert_and_set_id(table: dataset.table.Table, obj: Dict[str, Any], idfield: str = 'id') → Any

Inserts an object into a dataset.Table and ensures that the primary key (PK) is written back to the object, in its idfield.

(The dataset.Table.insert() command returns the primary key. However, it doesn’t store that back, and we want users to do that consistently.)

Parameters:

• table – the database table in which to insert
• obj – the dict-like record object to be added to the database
• idfield – the name of the primary key (PK) to be created within obj, containing the record’s PK in the database

Returns:

the primary key (typically integer)

whisker.convenience.load_config_or_die(config_filename: str = None, mandatory: Iterable[Union[str, List[Any]]] = None, defaults: Dict[str, Any] = None, log_config: bool = False) → attrdict.dictionary.AttrDict

Offers a GUI file prompt; loads a YAML config from it; or exits.

Parameters:

• config_filename – Optional filename. If not supplied, a GUI interface will be used to ask the user.
• mandatory – List of mandatory items; if one of these is itself a list, that list is used as a hierarchy of attributes.
• defaults – A dict-like object of defaults.
• log_config – Report the config to the Python log?

Returns:

AttrDict containing the config

Return type:

an class

whisker.convenience.save_data(tablename: str, results: List[Dict[str, Any]], taskname: str, timestamp: Union[arrow.arrow.Arrow, datetime.datetime] = None, output_format: str = 'csv') → None

Saves a dataset result set to a suitable output file.

Parameters:

• tablename – table name, used only for creating the filename
• results – results to save
• taskname – task name, used only for creating the filename
• timestamp – timestamp, used only for creating the filename; if None, the current date/time is used
• output_format – one of: "csv", "json", "tabson" (see https://dataset.readthedocs.org/en/latest/api.html#dataset.freeze)

whisker.convenience.update_record(table: dataset.table.Table, obj: Dict[str, Any], newvalues: Dict[str, Any], idfield: str = 'id') → Optional[int]

Updates an existing record, using its primary key.

Parameters:

• table – the database table to update
• obj – the dict-like record object to be updated
• newvalues – a dictionary of new values to write to obj and the database
• idfield – the name of the primary key (PK, ID) within obj, used for the SQL WHERE clause to determine which record to update

Returns:

the number of rows updated, or None

whisker.debug_qt

whisker/debug_qt.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Functions to debug Qt objects and signals under PyQt5.

• See http://stackoverflow.com/questions/2045352

whisker.debug_qt.debug_object(obj: SimpleClass) → None

Writes a debug log message within information about the QObject.

whisker.debug_qt.debug_thread(thread: <MagicMock id='140405796558440'>) → None

Writes a debug log message about the QThread.

whisker.debug_qt.enable_signal_debugging(connect_call: Callable[Any, None] = None, disconnect_call: Callable[Any, None] = None, emit_call: Callable[Any, None] = None) → None

Call this to enable PySide/Qt signal debugging. This will trap all connect() and disconnect() calls, calling the user-supplied functions first and then the real things.

whisker.debug_qt.enable_signal_debugging_simply() → None

Enables Qt signal debugging for connect() and emit() calls.

whisker.debug_qt.simple_connect_debugger(*args) → None

Function to report on connect() calls.

whisker.debug_qt.simple_emit_debugger(*args) → None

Function to report on emit() calls.

whisker.exceptions

whisker/exceptions.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Whisker exceptions.

exception whisker.exceptions.ImproperlyConfigured

Whisker is improperly configured; normally due to a missing library.

exception whisker.exceptions.ValidationError(message: str)

Exception to represent failure to validate user input.

exception whisker.exceptions.WhiskerCommandFailed

Raised by whisker.api.WhiskerApi if a Whisker immediate-socket command fails.

whisker.qt

whisker/qt.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Qt classes for Whisker GUI tasks.

class whisker.qt.DatabaseModelMixin(session: sqlalchemy.orm.session.Session, listdata: List[Any], **kwargs)

Mixin to provide functions that operate on items within a Qt Model and talk to an underlying SQLAlchemy database session.

Typically mixed in like this:

class MyListModel(QAbstractListModel, DatabaseModelMixin):
    pass

Parameters:

• session – SQLAlchemy Session
• listdata – data; a collection of SQLAlchemy ORM objects in a format that behaves like a list
• kwargs – additional parameters for superclass (required for mixins)

delete_item(row_index: bool, delete_from_session: bool = True) → None

Deletes an item from the collection (and optionally from the SQLAlchemy session).

Parameters:

• row_index – index
• delete_from_session – also delete the ORM object from the SQLAlchemy session? Default is True.

Raises:

ValueError – if index invalid

get_object(index: int) → Any

Parameters:index – integer index Returns:ORM object at the index, or None if the index is out of bounds

insert_at_index(obj: object, index: int = None, add_to_session: bool = True, flush: bool = True) → None

Inserts an ORM object into the data list.

Parameters:

• obj – SQLAlchemy ORM object to insert
• index – add it at this index location; specifying None adds it at the end of the list
• add_to_session – also add the object to the SQLAlchemy session? Default is True.
• flush – flush the SQLAlchemy session after adding the object? Only applicable if add_to_session is true.

Raises:

ValueError – if index invalid

item_deletable(rowindex: int) → bool

Override this if you need to prevent rows being deleted.

move_down(index: int) → None

Moves an object down one place in the list (from index to index + 1).

Parameters:index – index of the object to move Raises:ValueError – if index invalid

move_up(index: int) → None

Moves an object up one place in the list (from index to index - 1).

Parameters:index – index of the object to move Raises:ValueError – if index invalid

exception whisker.qt.EditCancelledException

Exception to represent that the user cancelled editing.

class whisker.qt.GarbageCollector(parent: SimpleClass, debug: bool = False, interval_ms=10000)

Disable automatic garbage collection and instead collect manually every INTERVAL milliseconds.

This is done to ensure that garbage collection only happens in the GUI thread, as otherwise Qt can crash.

See:

• https://riverbankcomputing.com/pipermail/pyqt/2011-August/030378.html
• http://pydev.blogspot.co.uk/2014/03/should-python-garbage-collector-be.html
• https://bugreports.qt.io/browse/PYSIDE-79

Parameters:

• parent – parent QObject
• debug – be verbose about garbage collection
• interval_ms – period/interval (in ms) at which to perform garbage collection

check() → None

Check if garbage collection should happen, and if it should, do it.

static debug_cycles() → None

Collect garbage and print what was collected.

class whisker.qt.GenericAttrTableModel(data: List[Any], header: List[Tuple[str, str]], session: sqlalchemy.orm.session.Session, default_sort_column_name: str = None, default_sort_order: int = <MagicMock name='mock.AscendingOrder' id='140405871650800'>, deletable: bool = True, parent: SimpleClass = None, **kwargs)

Takes a list of objects and a list of column headers; provides a view on it using func:str. That is, for a list of objects [a, b, c] and a list of headers ["h1", "h2", "h3"], it provides a table view like

h1	h2	h3
str(a.h1)	str(a.h2)	str(a.h3)
str(b.h1)	str(b.h2)	str(b.h3)
str(c.h1)	str(c.h2)	str(c.h3)

• Note that it MODIFIES THE LIST PASSED TO IT.
• Sorting: consider QSortFilterProxyModel… but not clear that can do arbitrary column sorts, since its sorting is via its lessThan() function.
• The tricky part is keeping selections persistent after sorting. Not achieved yet. Simpler to wipe the selections.

Parameters:

• data – data; a collection of SQLAlchemy ORM objects in a format that behaves like a list
• header – list of (colname, attr_or_func) tuples. The first part, colname, is displayed to the user. The second part, attr_or_func, is used to retrieve data. For each object obj, the view will call thing = getattr(obj, attr_or_func). If thing is callable (i.e. is a member function), the view will display str(thing()); otherwise, it will display str(thing).
• session – SQLAlchemy Session
• default_sort_column_name – column to sort by to begin with, or None to respect the original order of data
• default_sort_order – default sort order (default is ascending order)
• deletable – may the user delete objects from the collection?
• parent – parent (owning) QObject
• kwargs – additional parameters for superclass (required for mixins)

columnCount(parent: <MagicMock id='140405793012872'> = <MagicMock name='mock()' id='140405871621792'>, *args, **kwargs)

Qt override.

Parameters:

• parent – parent object, specified by QModelIndex
• args – unnecessary? Certainly unused.
• kwargs – unnecessary? Certainly unused.

Returns:

number of columns

data(index: <MagicMock id='140405793012872'>, role: int = <MagicMock name='mock.DisplayRole' id='140405871638344'>) → Optional[str]

Returns the data for a given row/column (in this case as a string).

Overrides QAbstractTableModel.data(); see http://doc.qt.io/qt-5/qabstractitemmodel.html.

Parameters:

• index – QModelIndex, which specifies the row/column number
• role – a way of specifying the type of view on the data; here, we only accept Qt.DisplayRole

Returns:

string representation of the relevant attribute (or member function result) for the relevant SQLAlchemy ORM object, or None

get_default_sort() → Tuple[int, int]

Returns:(default_sort_column_number, default_sort_order) where the column number is zero-based and the sort order is as per __init__()

headerData(col: int, orientation: int, role: int = <MagicMock name='mock.DisplayRole' id='140405871638344'>) → Optional[str]

Qt override.

Returns the column heading for a particular column.

Parameters:

• col – column number
• orientation – required to be Qt.Horizontal
• role – required to be Qt.DisplayRole

Returns:

column header string, or None

rowCount(parent: <MagicMock id='140405793012872'> = <MagicMock name='mock()' id='140405871621792'>, *args, **kwargs)

Counts the number of rows.

Overrides QAbstractTableModel.rowCount(); see http://doc.qt.io/qt-5/qabstractitemmodel.html.

Parameters:

• parent – parent object, specified by QModelIndex
• args – unnecessary? Certainly unused.
• kwargs – unnecessary? Certainly unused.

Returns:

“number of rows under the given parent. When the parent is valid… the number of children of parent.” Here: the number of objects in the list.

Return type:

In the Qt original

sort(col: int, order: int = <MagicMock name='mock.AscendingOrder' id='140405871650800'>) → None

Sort table on a specified column.

Parameters:

• col – column number
• order – sort order (e.g. ascending order)

class whisker.qt.GenericAttrTableView(session: sqlalchemy.orm.session.Session, modal_dialog_class, parent: SimpleClass = None, sortable: bool = True, stretch_last_section: bool = True, readonly: bool = False, **kwargs)

Provides a QTableView view on SQLAlchemy data.

Signals:

• selection_changed(QItemSelection, QItemSelection)

Parameters:

• session – SQLAlchemy Session
• modal_dialog_class – class that is a subclass of TransactionalEditDialogMixin
• parent – parent (owning) QObject
• sortable – should we sort the data?
• stretch_last_section – stretch the last column (section) horizontally?
• readonly – read-only view?
• kwargs – additional parameters for superclass (required for mixins)

get_selected_modelindex() → Optional[<MagicMock id='140405793012872'>]

Gets the model index of the selected item.

Returns:a QModelIndex or None

refresh_sizing() → None

Ask the view to resize its rows/columns to fit its data.

resize() → None

Resize all rows to have the correct height, and its columns to the correct width.

setModel(model: SimpleClass) → None

Qt override to set the model used by this view.

Parameters:model – instance of the model

class whisker.qt.GenericListModel(data, session: sqlalchemy.orm.session.Session, parent: SimpleClass = None, **kwargs)

Takes a list and provides a view on it using str(). That is, for a list of object [a, b, c], it displays a list view with items (rows) like this:

str(a)

str(b)

str(c)

• Note that it MODIFIES THE LIST PASSED TO IT.

data(index: <MagicMock id='140405793012872'>, role: int = <MagicMock name='mock.DisplayRole' id='140405871638344'>) → Optional[str]

Returns the data for a given row (in this case as a string).

Overrides QAbstractListModel.data(); see http://doc.qt.io/qt-5/qabstractitemmodel.html.

Parameters:

• index – QModelIndex, which specifies the row number
• role – a way of specifying the type of view on the data; here, we only accept Qt.DisplayRole

Returns:

string representation of the object, or None

rowCount(parent: <MagicMock id='140405793012872'> = <MagicMock name='mock()' id='140405871621792'>, *args, **kwargs) → int

Counts the number of rows.

Overrides QAbstractListModel.rowCount(); see http://doc.qt.io/qt-5/qabstractitemmodel.html.

Parameters:

• parent – parent object, specified by QModelIndex
• args – unnecessary? Certainly unused.
• kwargs – unnecessary? Certainly unused.

Returns:

“number of rows under the given parent. When the parent is valid… the number of children of parent.” Here: the number of objects in the list.

Return type:

In the Qt original

class whisker.qt.ModalEditListView(session: sqlalchemy.orm.session.Session, modal_dialog_class: Type[whisker.qt.TransactionalEditDialogMixin], *args, readonly: bool = False, **kwargs)

A version of QListView that supports SQLAlchemy handling of its objects.

Parameters:

• session – SQLAlchemy Session
• modal_dialog_class – class that is a subclass of TransactionalEditDialogMixin
• readonly – read-only view?
• args – positional arguments to superclass
• kwargs – keyword arguments to superclass

get_selected_modelindex() → Optional[<MagicMock id='140405793012872'>]

Gets the model index of the selected item.

Returns:a QModelIndex or None

setModel(model: SimpleClass) → None

Qt override to set the model used by this view.

Parameters:model – instance of the model

class whisker.qt.RadioGroup(value_text_tuples: Iterable[Tuple[str, Any]], default: Any = None)

Framework for radio buttons.

Parameters:

• value_text_tuples – list of (value, text) tuples, where value is the value stored to Python object and text is the text shown to the user
• default – default value

add_buttons_to_layout(layout: <MagicMock id='140405794232360'>) → None

Adds its button widgets to the specified Qt layout.

get_value() → Any

Returns:the value of the currently selected radio button, or None if none is selected

set_value(value: Any) → None

Set the radio group to the specified value.

class whisker.qt.StatusMixin(name: str, logger: logging.Logger, thread_info: bool = True, caller_info: bool = True, **kwargs)

Add this to a QObject to provide easy Python logging and Qt signal-based status/error messaging.

It emits status messages to a Python log and to Qt signals.

Uses the same function names as Python logging, for predictability.

Signals:

• status_sent(str, str)
• error_sent(str, str)

Parameters:

• name – name to add to log output
• logger – Python logger to send output to
• thread_info – show thread information as part of the output?
• caller_info – add information about the function that sent the message?
• kwargs – additional parameters for superclass (required for mixins)

class whisker.qt.TextLogElement(maximum_block_count: int = 1000, font_size_pt: int = 10, font_family: str = 'Courier', title: str = 'Log')

Add a text log to your dialog box.

Encapsulates a QWidget representing a textual log in a group box.

Use this to embed a log within another Qt dialogue.

(This is nothing to do with the Python logging logs.)

Parameters:

• maximum_block_count – the maximum number of blocks (log entries) to show in this window
• font_size_pt – font size, in points
• font_family – font family
• title – title for group box

add(msg: str) → None

Adds a message to the log.

copy_whole_log() → None

Copies the log to the clipboard.

get_widget() → <MagicMock id='140405793951928'>

Returns:a QWidget that is the group box containing the log and its controls

scroll_to_end_of_log() → None

Scrolls the log so that the last entry is visible.

class whisker.qt.TransactionalEditDialogMixin(session: sqlalchemy.orm.session.Session, obj: object, layout: <MagicMock id='140405794232360'>, readonly: bool = False, **kwargs)

Mixin for a config-editing dialogue. Use it as:

class MyEditingDialog(TransactionalEditDialogMixin, QDialog):
    pass

Wraps the editing in a SAVEPOINT transaction. The caller must still commit() afterwards, but any rollbacks are automatic.

See also http://pyqt.sourceforge.net/Docs/PyQt5/multiinheritance.html for the super().__init__(..., **kwargs) chain.

Signals:

• ok()

Parameters:

• session – SQLAlchemy Session
• obj – SQLAlchemy ORM object being edited
• layout – Qt class`QLayout` to encapsulate within the class:QDialog’s main layout. This should contain all the editing widgets. The mixin wraps that with OK/cancel buttons.
• readonly – make this a read-only view
• kwargs – additional parameters to superclass

dialog_to_object(obj: object) → None

The class using the mixin must override this to write information from the dialogue box to the SQLAlchemy ORM object. It should raise an exception if validation fails. (This class isn’t fussy about which exception that is; it checks for Exception.)

edit_in_nested_transaction() → int

Pops up the dialog, allowing editing.

• Does so within a database transaction.
• If the user clicks OK and the data validates, commits the transaction.
• If the user cancels, rolls back the transaction.
• We want it nestable, so that the config dialog box can edit part of the config, reversibly, without too much faffing around.

Development notes:

• We could nest using SQLAlchemy’s support for nested transactions, which works whether or not the database itself supports nested transactions via the SAVEPOINT method.

• With sessions, one must use autocommit=True and the subtransactions flag; these are virtual transactions handled by SQLAlchemy.

• Alternatively one can use begin_nested() or begin(nested=True), which uses SAVEPOINT.

• The following databases support the SAVEPOINT method:

  • MySQL with InnoDB
  • SQLite, from v3.6.8 (2009)
  • PostgreSQL

• Which is better? The author suggests SAVEPOINT for most applications (https://groups.google.com/forum/#!msg/sqlalchemy/CaZyyMx7_8Y/otM0BzDyaigJ). Regarding subtransactions: “When a rollback is issued, the subtransaction will directly roll back the innermost real transaction, however each subtransaction still must be explicitly rolled back to maintain proper stacking of subtransactions.” So it’s not as simple as you might guess.

• See:

  • http://docs.sqlalchemy.org/en/latest/core/connections.html
  • http://docs.sqlalchemy.org/en/latest/orm/session_transaction.html
  • http://stackoverflow.com/questions/2336950/transaction-within-transaction
  • http://stackoverflow.com/questions/1306869/are-nested-transactions-allowed-in-mysql
  • https://en.wikipedia.org/wiki/Savepoint
  • http://www.sqlite.org/lang_savepoint.html
  • http://stackoverflow.com/questions/1654857/nested-transactions-with-sqlalchemy-and-sqlite

• Let’s use the SAVEPOINT technique.

• No. Even this fails:

  with self.session.begin_nested():
      self.config.port = 5000
  

• We were aiming for this:

  try:
      with self.session.begin_nested():
          result = self.exec_()  # enforces modal
          if result == QDialog.Accepted:
              logger.debug("Config changes accepted;  will be committed")
          else:
              logger.debug("Config changes cancelled")
              raise EditCancelledException()
  except EditCancelledException:
      logger.debug("Config changes rolled back.")
  except:
      logger.debug("Exception within nested transaction. "
                   "Config changes will be rolled back.")
      raise
      # Other exceptions will be handled as normal.
  

• No… the commit fails, and this SQL is emitted:

  SAVEPOINT sa_savepoint_1
  UPDATE table SET field=?
  RELEASE SAVEPOINT sa_savepoint_1  -- sensible
  ROLLBACK TO SAVEPOINT sa_savepoint_1  -- not sensible
  -- raises sqlite3.OperationalError: no such savepoint: sa_savepoint_1
  

• May be this bug:

  • https://www.mail-archive.com/sqlalchemy@googlegroups.com/msg28381.html
  • http://bugs.python.org/issue10740
  • https://groups.google.com/forum/#!topic/sqlalchemy/1QelhQ19QsE

• The bugs are detailed in sqlalchemy/dialects/sqlite/pysqlite.py;

  see “Serializable isolation / Savepoints / Transactional DDL”

• We work around it by adding hooks to the engine as per that advice; see db.py

class whisker.qt.ViewAssistMixin(session: sqlalchemy.orm.session.Session, modal_dialog_class: Type[whisker.qt.TransactionalEditDialogMixin], readonly: bool = False, *args, **kwargs)

Mixin to add SQLAlchemy database-handling functions to a ViewAssistMixin.

Typically used like this:

class MyListView(QListView, ViewAssistMixin):
    pass

Signals:

• selection_changed(QItemSelection, QItemSelection)
• selected_maydelete(bool, bool)

Parameters:

• session – SQLAlchemy Session
• modal_dialog_class – class that is a subclass of TransactionalEditDialogMixin
• readonly – read-only view?
• kwargs – additional parameters for superclass (required for mixins)

add_in_nested_transaction(new_object: object, at_index: int = None) → Optional[int]

Starts an SQLAlchemy nested transaction (which uses SAVEPOINT SQL); adds the new object into the session; edits the object. If the editing is cancelled, cancel the addition. Otherwise, the object is added to the session (in a nested transaction that might be cancelled by a caller) and to the model/list.

Parameters:

• new_object – SQLAlchemy ORM object to insert
• at_index – add it at this index location; specifying None (the default) adds it at the end of the list; 0 is the start

clear_selection() → None

Clears any selection.

edit(index: <MagicMock id='140405793012872'>, trigger: <MagicMock name='mock.EditTrigger' id='140405793084752'>, event: <MagicMock id='140405796557320'>) → bool

Edits the specified object.

Overrides QAbstractItemView::edit(); see http://doc.qt.io/qt-5/qabstractitemview.html#edit-1.

Parameters:

• index – QModelIndex of the object to edit
• trigger – action that caused the editing process
• event – associated QEvent

Returns:

True if the view’s State is now EditingState; otherwise False

edit_by_modelindex(index: <MagicMock id='140405793012872'>, readonly: bool = None) → None

Edits the specified object.

Parameters:

• index – QModelIndex of the object to edit
• readonly – read-only view? If None (the default), uses the setting from self.

edit_selected(readonly: bool = None) → None

Edits the currently selected object.

Parameters:readonly – read-only view? If None (the default), uses the setting from self.

get_n_rows() → int

Returns:the number of rows (items) present

get_selected_modelindex() → Optional[<MagicMock id='140405793012872'>]

Returns the QModelIndex of the currently selected row, or None.

Should be overridden in derived classes.

get_selected_object() → object

Returns the SQLAlchemy ORM object that’s currently selected, or None.

get_selected_row_index() → Optional[int]

Gets the index of the currently selected row.

Returns:index, or None

go_to(row: Optional[int]) → None

Makes a specific row (by index) the currently selected row.

Parameters:row – row index, or None to select the last row if there is one

insert_at_end(obj: object, add_to_session: bool = True, flush: bool = True) → None

Insert a SQLAlchemy ORM object into the model at the end of the list (and, optionally, into the SQLAlchemy session).

Parameters:

• obj – SQLAlchemy ORM object to insert
• add_to_session – also add the object to the SQLAlchemy session? Default is True.
• flush – flush the SQLAlchemy session after adding the object? Only applicable if add_to_session is true.

insert_at_index(obj: object, index: int = None, add_to_session: bool = True, flush: bool = True) → None

Insert an SQLAlchemy ORM object into the model at a specific location (and, optionally, into the SQLAlchemy session).

Parameters:

• obj – SQLAlchemy ORM object to insert
• index – index to insert at; None for end, 0 for start
• add_to_session – also add the object to the SQLAlchemy session? Default is True.
• flush – flush the SQLAlchemy session after adding the object? Only applicable if add_to_session is true.

insert_at_start(obj: object, add_to_session: bool = True, flush: bool = True) → None

Insert a SQLAlchemy ORM object into the model at the start of the list (and, optionally, into the SQLAlchemy session).

Parameters:

• obj – SQLAlchemy ORM object to insert
• add_to_session – also add the object to the SQLAlchemy session? Default is True.
• flush – flush the SQLAlchemy session after adding the object? Only applicable if add_to_session is true.

is_selected() → bool

Is a row currently selected?

move_selected_down() → None

Moves the selected object down one place in the list.

move_selected_up() → None

Moves the selected object up one place in the list.

remove_by_index(row_index: int, delete_from_session: bool = True) → None

Remove a specified SQLAlchemy ORM object from the list.

Parameters:

• row_index – index of object to remove
• delete_from_session – also delete it from the SQLAlchemy session? (default: True)

remove_selected(delete_from_session: bool = True) → None

Remove the currently selected SQLAlchemy ORM object from the list.

Parameters:delete_from_session – also delete it from the SQLAlchemy session? (default: True)

set_model_common(model: SimpleClass, listview_base_class: Type[SimpleClass]) → None

Helper function to set up a Qt model.

Parameters:

• model – instance of the Qt model
• listview_base_class – class being used as the Qt view

whisker.qt.exit_on_exception(func)

Decorator to stop whole program on exceptions (use for threaded slots).

See

• http://stackoverflow.com/questions/18740884
• http://stackoverflow.com/questions/308999/what-does-functools-wraps-do

Parameters:func – the function to decorate Returns:the decorated function

whisker.qt.run_gui(qt_app: <MagicMock id='140405794403664'>, win: <MagicMock id='140405794016952'>) → int

Run a GUI, given a base window.

Parameters:

• qt_app – Qt application
• win – Qt dialogue window

Returns:

exit code

whisker.qtclient

whisker/qtclient.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Multithreaded framework for Whisker Python clients using Qt.

• Created: late 2016
• Last update: 20 Sep 2018
• Note funny bug: data sometimes sent twice.
  • Looks like it might be this: http://www.qtcentre.org/threads/29462-QTcpSocket-sends-data-twice-with-flush()
  • Attempted solution:
    • change QTcpSocket() to QTcpSocket(parent=self), in case the socket wasn’t getting moved between threads properly – didn’t fix
    • disable flush() – didn’t fix.
    • send function is only being called once, according to log
    • adding thread ID information to the log shows that whisker_controller events are coming from two threads…
  • Anyway, bug was this:
    • http://stackoverflow.com/questions/34125065
    • https://bugreports.qt.io/browse/PYSIDE-249
• Source:
  • http://code.qt.io/cgit/qt/qtbase.git/tree/src/corelib/kernel/qobject.h?h=5.4
  • http://code.qt.io/cgit/qt/qtbase.git/tree/src/corelib/kernel/qobject.cpp?h=5.4

class whisker.qtclient.ThreadOwnerState

An enumeration.

class whisker.qtclient.WhiskerController(server: str, parent: SimpleClass = None, connect_timeout_ms: int = 5000, read_timeout_ms: int = 500, name: str = 'whisker_controller', sysevent_prefix: str = 'sys_', **kwargs)

Controls the Whisker immediate socket.

• Encapsulates the Whisker API.
• Lives in Whisker thread (B) (see WhiskerOwner).
• Emits signals that can be processed by the Whisker task (derived from WhiskerQtTask).

Signals:

• finished()
• connected()
• disconnected()
• message_received(str, arrow.Arrow, int)
• event_received(str, arrow.Arrow, int)
• key_event_received(str, arrow.Arrow, int)
• client_message_received(int, str, arrow.Arrow, int)
• warning_received(str, arrow.Arrow, int)
• syntax_error_received(str, arrow.Arrow, int)
• error_received(str, arrow.Arrow, int)
• pingack_received(arrow.Arrow, int)

Parameters:

• server – host name or IP address of Whisker server
• parent – optional parent :QObject
• connect_timeout_ms – time to wait for successful connection (ms) before considering it a failure
• read_timeout_ms – time to wait for each read (primary effect is to determine the application’s responsivity when asked to finish; see WhiskerMainSocketListener)
• name – (name for StatusMixin)
• sysevent_prefix – default system event prefix (for WhiskerController)
• kwargs – parameters to superclass

is_connected() → bool

Is our immediate socket connected?

(If the immediate socket is running, the main socket should be too.)

ping() → None

Ping the server.

Override WhiskerApi.ping() so we can emit a signal on success.

class whisker.qtclient.WhiskerMainSocketListener(server: str, port: int, parent: SimpleClass = None, connect_timeout_ms: int = 5000, read_timeout_ms: int = 100, name: str = 'whisker_mainsocket', **kwargs)

Whisker thread (A) (see WhiskerOwner). Listens to the main socket.

Signals:

• finished()
• disconnected()
• line_received(msg: str, timestamp: arrow.Arrow)

Parameters:

• server – host name or IP address of Whisker server
• port – main port number for Whisker server
• parent – optional parent :QObject
• connect_timeout_ms – time to wait for successful connection (ms) before considering it a failure
• read_timeout_ms – time to wait for each read (primary effect is to determine the application’s responsivity when asked to finish; see WhiskerMainSocketListener)
• name – (name for StatusMixin)
• kwargs – parameters to superclass

class whisker.qtclient.WhiskerOwner(task: whisker.qtclient.WhiskerQtTask, server: str, main_port: int = 3233, parent: SimpleClass = None, connect_timeout_ms: int = 5000, read_timeout_ms: int = 500, name: str = 'whisker_owner', sysevent_prefix: str = 'sys_', **kwargs)

Object to own and manage communication with a Whisker server.

This object is owned by the GUI thread. It devolves work to two other threads:

1. main socket listener (WhiskerMainSocketListener, as the instance self.mainsock, in the thread self.mainsockthread);
2. task (WhiskerQtTask, as the instance self.task in the thread self.taskthread) + immediate socket blocking handler (WhiskerController, as the instance self.controller in the thread self.taskthread).

References to “main” here just refers to the main socket (as opposed to the immediate socket), not the thread that’s doing most of the processing.

Signals of relevance to users:

• connected()
• disconnected()
• finished()
• message_received(str, arrow.Arrow, int)
• event_received(str, arrow.Arrow, int)
• pingack_received(arrow.Arrow, int)

Parameters:

• task – instance of something derived from WhiskerQtTask
• server – host name or IP address of Whisker server
• main_port – main port number for Whisker server
• parent – optional parent :QObject
• connect_timeout_ms – time to wait for successful connection (ms) before considering it a failure
• read_timeout_ms – time to wait for each read (primary effect is to determine the application’s responsivity when asked to finish; see WhiskerMainSocketListener)
• name – (name for StatusMixin)
• sysevent_prefix – default system event prefix (for WhiskerController)
• kwargs – parameters to superclass

is_running() → bool

Are any of our worker threads starting/running/stopping?

ping() → None

Pings the Whisker server.

report_status() → None

Print current thread/server state to the log.

start() → None

Start our worker threads.

stop() → None

Called by the GUI when we want to stop.

class whisker.qtclient.WhiskerQtTask(parent: SimpleClass = None, name: str = 'whisker_task', **kwargs)

Base class for building a Whisker task itself. You should derive from this.

• Lives in Whisker thread (B) (see WhiskerOwner).

Signals:

• finished()

Parameters:

• parent – optional parent :QObject
• name – (name for StatusMixin)
• kwargs – parameters to superclass

on_connect() → None

The WhiskerOwner makes this slot get called when the WhiskerController is connected.

You should override this.

set_controller(controller: whisker.qtclient.WhiskerController) → None

Called by WhiskerOwner. No need to override.

whisker.qtclient.disable_nagle(socket: <MagicMock id='140405852664832'>) → None

Disable the Nagle algorithm on the Qt socket. (This makes it send any outbound data immediately, rather than waiting and packaging it up with potential subsequent data. So this function chooses a low-latency mode over a high-efficiency mode.)

whisker.qtclient.get_socket_error(socket: <MagicMock id='140405852664832'>) → str

Returns a textual description of the last error on the specified Qt socket.

whisker.qtclient.is_socket_connected(socket: <MagicMock id='140405852664832'>) → bool

Is the Qt socket connected?

whisker.qtclient.quote(msg: str) → str

Return its argument suitably quoted for transmission to Whisker.

• Whisker has quite a primitive quoting system…
• Check with strings that actually include quotes.

whisker.random

whisker/random.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Randomization functions that may be used by Whisker tasks.

class whisker.random.ShuffleLayerMethod(flat: bool = False, layer_key: int = None, layer_attr: str = None, layer_func: Callable[Any, Any] = None, shuffle_func: Callable[Sequence[Any], List[int]] = None)

Class to representing instructions to layered_shuffle() (q.v.).

Parameters:

• flat – take data as x[index]?
• layer_key – take data as x[index][layer_key]?
• layer_attr – take data as getattr(x[index], layer_attr)?
• layer_func – take data as layer_func(x[index])?
• shuffle_func – function (N.B. may be a lambda function with parameters attached) that takes a list of objects and returns a list of INDEXES, suitably shuffled.

Typical values of shuffle_func:

• None: no shuffle
• random_shuffle_indexes: plain shuffle; see random_shuffle_indexes()
• dwor_shuffle_indexes: DWOR-style shuffle (will need a lambda for its multiplier parameter; see dwor_shuffle_indexes()
• block_shuffle_indexes_by_value: aggregate items into blocks, defined by value, and shuffle those blocks; see block_shuffle_indexes_by_value()
• sort_indexes: not exactly shuffling! See sort_indexes().
• reverse_sort_indexes: not exactly shuffling! See reverse_sort_indexes().

get_indexes_for_value(x: List[Any], value: Any) → List[int]

Returns a list of indexes of x where its value (as defined by this layer) is value.

get_unique_values(x: List[Any]) → List[Any]

Returns all the unique values of x for this layer.

get_values(x: List[Any]) → List[Any]

Returns all the values of interest from x for this layer.

whisker.random.block_shuffle_by_attr(x: List[Any], attrorder: List[str], start: int = None, end: int = None) → None

DEPRECATED: layered_shuffle() is more powerful.

Exactly as for block_shuffle_by_item(), but by item attribute rather than item index number.

For example:

from collections import namedtuple
import itertools
from whisker.random import block_shuffle_by_attr

p = list(itertools.product("ABC", "xyz", "123"))
Trio = namedtuple("Trio", ["upper", "lower", "digit"])
q = [Trio(*x) for x in p]
block_shuffle_by_attr(q, ['upper', 'lower', 'digit'])

q started off as:

[
    Trio(upper='A', lower='x', digit='1'),
    Trio(upper='A', lower='x', digit='2'),
    Trio(upper='A', lower='x', digit='3'),
    Trio(upper='A', lower='y', digit='1'),
    Trio(upper='A', lower='y', digit='2'),
    Trio(upper='A', lower='y', digit='3'),
    Trio(upper='A', lower='z', digit='1'),
    Trio(upper='A', lower='z', digit='2'),
    Trio(upper='A', lower='z', digit='3'),
    Trio(upper='B', lower='x', digit='1'),
    Trio(upper='B', lower='x', digit='2'),
    Trio(upper='B', lower='x', digit='3'),
    Trio(upper='B', lower='y', digit='1'),
    Trio(upper='B', lower='y', digit='2'),
    Trio(upper='B', lower='y', digit='3'),
    Trio(upper='B', lower='z', digit='1'),
    Trio(upper='B', lower='z', digit='2'),
    Trio(upper='B', lower='z', digit='3'),
    Trio(upper='C', lower='x', digit='1'),
    Trio(upper='C', lower='x', digit='2'),
    Trio(upper='C', lower='x', digit='3'),
    Trio(upper='C', lower='y', digit='1'),
    Trio(upper='C', lower='y', digit='2'),
    Trio(upper='C', lower='y', digit='3'),
    Trio(upper='C', lower='z', digit='1'),
    Trio(upper='C', lower='z', digit='2'),
    Trio(upper='C', lower='z', digit='3')
]

but after the shuffle q might now be:

[
    Trio(upper='B', lower='z', digit='1'),
    Trio(upper='B', lower='z', digit='3'),
    Trio(upper='B', lower='z', digit='2'),
    Trio(upper='B', lower='x', digit='1'),
    Trio(upper='B', lower='x', digit='3'),
    Trio(upper='B', lower='x', digit='2'),
    Trio(upper='B', lower='y', digit='3'),
    Trio(upper='B', lower='y', digit='2'),
    Trio(upper='B', lower='y', digit='1'),
    Trio(upper='A', lower='z', digit='2'),
    Trio(upper='A', lower='z', digit='1'),
    Trio(upper='A', lower='z', digit='3'),
    Trio(upper='A', lower='x', digit='1'),
    Trio(upper='A', lower='x', digit='2'),
    Trio(upper='A', lower='x', digit='3'),
    Trio(upper='A', lower='y', digit='3'),
    Trio(upper='A', lower='y', digit='1'),
    Trio(upper='A', lower='y', digit='2'),
    Trio(upper='C', lower='x', digit='2'),
    Trio(upper='C', lower='x', digit='3'),
    Trio(upper='C', lower='x', digit='1'),
    Trio(upper='C', lower='y', digit='2'),
    Trio(upper='C', lower='y', digit='1'),
    Trio(upper='C', lower='y', digit='3'),
    Trio(upper='C', lower='z', digit='1'),
    Trio(upper='C', lower='z', digit='2'),
    Trio(upper='C', lower='z', digit='3')
]

You can see that the A/B/C group has been shuffled as blocks. Then, within B, the x/y/z groups have been shuffled (and so on for A and C). Then, within B.z, the 1/2/3 values have been shuffled (and so on).

whisker.random.block_shuffle_by_item(x: List[Any], indexorder: List[int], start: int = None, end: int = None) → None

DEPRECATED: layered_shuffle() is more powerful.

Shuffles the list x[start:end] hierarchically, in place.

Parameters:

• x – list to shuffle
• indexorder – a list of indexes of each item of x The first index varies slowest; the last varies fastest.
• start – start index of x
• end – end index of x

For example:

p = list(itertools.product("ABC", "xyz", "123"))

x is now a list of tuples looking like ('A', 'x', '1').

block_shuffle_by_item(p, [0, 1, 2])

p might now look like:

C z 1 } all values of "123" appear  } first "xyz" block
C z 3 } once, but randomized        }
C z 2 }                             }
                                    }
C y 2 } next "123" block            }
C y 1 }                             }
C y 3 }                             }
                                    }
C x 3                               }
C x 2                               }
C x 1                               }

A y 3                               } second "xyz" block
...                                 } ...

A clearer explanation is in block_shuffle_by_attr().

whisker.random.block_shuffle_indexes_by_value(x: List[Any]) → List[int]

Returns a list of indexes of x, block-shuffled by value.

That is: we aggregate items into blocks, defined by value, and shuffle those blocks, returning the corresponding indexes of the original list.

whisker.random.dwor_shuffle_indexes(x: List[Any], multiplier: int = 1) → List[int]

Returns a list of indexes of x, DWOR-shuffled by value.

This is a bit tricky as we don’t have a guarantee of equal numbers. It does sensible things in those circumstances.

whisker.random.gen_dwor(values: Iterable[Any], multiplier: int = 1) → Generator[[Any, None], None]

Generates values using a draw-without-replacement (DWOR) system.

Parameters:

• values – values to generate
• multiplier – DWOR multiplier; see below.

Yields:

successive values

Here’s how it works.

• Suppose values == [A, B, C].

• We’ll call the number of values (here, 3), and the “multiplier” parameter.

• If you iterate through gen_dwor(values, multiplier=1), you will get a sequence that might look like this (with spaces added for clarity):

  CAB ABC BCA BAC BAC ACB CBA ...
  

  That is, individual are drawn randomly from a “hat” of size , containing one of each thing from values. When the hat is empty, it is refilled with more.

• If you iterate through gen_dwor(values, multiplier=2), however, you might get this:

  AACBBC CABBAC BAACCB ...
  

  The computer has put copies of each value in the hat, and then draws one each time at random (so the hat starts with values in it). When the hat is exhausted, it re-populates.

The general idea is to provide randomness, but randomness that is constrained to prevent unlikely but awkward sequences like

AAAAAAAAAAAAAAAA ... unlikely but possible with full randomness!

yet also have the option to avoid predictability. With , then a clever subject could infer exactly what’s coming up on every th trial. So a low value of brings very few “runs” but some predictability; as approaches infinity, it’s equivalent to full randomness; some reasonably low value of in between may be a useful experimental sweet spot.

See also, for example:

• http://www.whiskercontrol.com/help/FiveChoice/index.html?fivechoice_dwor.htm

whisker.random.get_dwor_list(values: Iterable[Any], length: int, multiplier: int = 1) → List[Any]

Makes a fixed-length list via gen_dwor().

Parameters:

• values – values to pick from
• length – list length
• multiplier – DWOR multiplier

Returns:

list of length length

Example:

from whisker.random import get_dwor_list
values = ["a", "b", "c"]
print(get_dwor_list(values, length=24, multiplier=1))
print(get_dwor_list(values, length=24, multiplier=2))
print(get_dwor_list(values, length=24, multiplier=3))

whisker.random.get_indexes_for_value(x: List[Any], value: Any) → List[int]

Returns a list of indexes of x where its value is value.

whisker.random.get_unique_values(iterable: Iterable[Any]) → List[Any]

Gets the unique values of its input. See https://stackoverflow.com/questions/12897374/get-unique-values-from-a-list-in-python.

(We don’t use list(set(x)), because if the elements of x are themselves lists (perfectly common!), that gives TypeError: unhashable type: 'list'.)

whisker.random.last_index_of(x: List[Any], value: Any) → int

Gets the index of the last occurrence of value in the list x.

whisker.random.layered_shuffle(x: List[Any], layers: List[whisker.random.ShuffleLayerMethod]) → None

Most powerful hierarchical shuffle command here.

Shuffles x in place in a layered way as specified by the sequence of methods.

In more detail:

• for each layer, it shuffles values of x as defined by the ShuffleLayerMethod (for example: “shuffle x in blocks based on the value of x.someattr”, or “shuffle x randomly”)
• it then proceeds to deeper layers within sub-lists defined by each unique value from the previous layer.

Parameters:

• x – sequence (e.g. list) to shuffle
• layers – list of ShuffleLayerMethod instructions

Examples:

from collections import namedtuple
import itertools
import logging
import random
from whisker.random import *
logging.basicConfig(level=logging.DEBUG)

startlist = ["a", "b", "c", "d", "a", "b", "c", "d", "a", "b", "c", "d"]
x1 = startlist[:]
x2 = startlist[:]
x3 = startlist[:]
x4 = startlist[:]

do_nothing_method = ShuffleLayerMethod(flat=True, shuffle_func=None)
do_nothing_method.get_unique_values(x1)
do_nothing_method.get_indexes_for_value(x1, "b")

layered_shuffle(x1, [do_nothing_method])
print(x1)

flat_randomshuffle_method = ShuffleLayerMethod(
    flat=True, shuffle_func=random_shuffle_indexes)
flat_randomshuffle_method.get_unique_values(x1)
flat_randomshuffle_method.get_indexes_for_value(x1, "b")
layered_shuffle(x1, [flat_randomshuffle_method])
print(x1)

flat_blockshuffle_method = ShuffleLayerMethod(
    flat=True, shuffle_func=block_shuffle_indexes_by_value)
layered_shuffle(x2, [flat_blockshuffle_method])
print(x2)

flat_dworshuffle_method = ShuffleLayerMethod(
    flat=True, shuffle_func=dwor_shuffle_indexes)
layered_shuffle(x3, [flat_dworshuffle_method])
print(x3)

flat_dworshuffle2_method = ShuffleLayerMethod(
    flat=True, shuffle_func=lambda x: dwor_shuffle_indexes(x, multiplier=2))
layered_shuffle(x4, [flat_dworshuffle2_method])
print(x4)

p = list(itertools.product("ABC", "xyz", "123"))
Trio = namedtuple("Trio", ["upper", "lower", "digit"])
q = [Trio(*x) for x in p]
print("\n".join(str(x) for x in q))

upper_method = ShuffleLayerMethod(
    layer_attr="upper", shuffle_func=block_shuffle_indexes_by_value)
lower_method = ShuffleLayerMethod(
    layer_attr="lower", shuffle_func=reverse_sort_indexes)
digit_method = ShuffleLayerMethod(
    layer_attr="digit", shuffle_func=random_shuffle_indexes)

layered_shuffle(q, [upper_method, lower_method, digit_method])
print("\n".join(str(x) for x in q))

whisker.random.make_dwor_hat(values: Iterable[Any], multiplier: int = 1) → List[Any]

Makes a “hat” to draw values from. See gen_dwor(). Does not modify the starting list; returns a copy.

whisker.random.random_shuffle_indexes(x: List[Any]) → List[int]

Returns a list of indexes of x, randomly shuffled.

whisker.random.reverse_sort_indexes(x: List[Any]) → List[int]

Returns the indexes of x in an order that would reverse-sort x by value.

whisker.random.shuffle_list_chunks(x: List[Any], chunksize: int) → None

Divides a list into chunks and shuffles the chunks themselves (in place). For example:

x = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
shuffle_list_chunks(x, 4)

x might now be

[5, 6, 7, 8, 1, 2, 3, 4, 9, 10, 11, 12]
 ^^^^^^^^^^  ^^^^^^^^^^  ^^^^^^^^^^^^^

Uses cardinal_pythonlib.lists.flatten_list() and cardinal_pythonlib.lists.sort_list_by_index_list(). (I say that mainly to test Intersphinx, when it is enabled.)

whisker.random.shuffle_list_slice(x: List[Any], start: int = None, end: int = None) → None

Shuffles a segment of a list, x[start:end], in place.

Note that start=None means “from the beginning” and end=None means “to the end”.

whisker.random.shuffle_list_subset(x: List[Any], indexes: List[int]) → None

Shuffles some elements of a list (in place). The elements to interchange (shuffle) as specified by indexes.

whisker.random.shuffle_list_within_chunks(x: List[Any], chunksize: int) → None

Divides a list into chunks and shuffles WITHIN each chunk (in place). For example:

x = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]
shuffle_list_within_chunks(x, 4)

x might now be:

[4, 1, 3, 2, 7, 5, 6, 8, 9, 12, 11, 10]
 ^^^^^^^^^^  ^^^^^^^^^^  ^^^^^^^^^^^^^

whisker.random.shuffle_where_equal_by_attr(x: List[Any], attrname: str) → None

DEPRECATED: layered_shuffle() is more powerful.

Shuffles a list x, in place, where list members are equal as judged by the attribute attrname.

This is easiest to show by example:

from collections import namedtuple
import itertools
from whisker.random import shuffle_where_equal_by_attr

p = list(itertools.product("ABC", "xyz", "123"))
Trio = namedtuple("Trio", ["upper", "lower", "digit"])
q = [Trio(*x) for x in p]
shuffle_where_equal_by_attr(q, 'digit')

q started off as:

[
    Trio(upper='A', lower='x', digit='1'),
    Trio(upper='A', lower='x', digit='2'),
    Trio(upper='A', lower='x', digit='3'),
    Trio(upper='A', lower='y', digit='1'),
    Trio(upper='A', lower='y', digit='2'),
    Trio(upper='A', lower='y', digit='3'),
    Trio(upper='A', lower='z', digit='1'),
    Trio(upper='A', lower='z', digit='2'),
    Trio(upper='A', lower='z', digit='3'),
    Trio(upper='B', lower='x', digit='1'),
    Trio(upper='B', lower='x', digit='2'),
    Trio(upper='B', lower='x', digit='3'),
    Trio(upper='B', lower='y', digit='1'),
    Trio(upper='B', lower='y', digit='2'),
    Trio(upper='B', lower='y', digit='3'),
    Trio(upper='B', lower='z', digit='1'),
    Trio(upper='B', lower='z', digit='2'),
    Trio(upper='B', lower='z', digit='3'),
    Trio(upper='C', lower='x', digit='1'),
    Trio(upper='C', lower='x', digit='2'),
    Trio(upper='C', lower='x', digit='3'),
    Trio(upper='C', lower='y', digit='1'),
    Trio(upper='C', lower='y', digit='2'),
    Trio(upper='C', lower='y', digit='3'),
    Trio(upper='C', lower='z', digit='1'),
    Trio(upper='C', lower='z', digit='2'),
    Trio(upper='C', lower='z', digit='3')
]

but after the shuffle q might now be:

[
    Trio(upper='A', lower='x', digit='1'),
    Trio(upper='A', lower='y', digit='2'),
    Trio(upper='A', lower='z', digit='3'),
    Trio(upper='B', lower='z', digit='1'),
    Trio(upper='A', lower='z', digit='2'),
    Trio(upper='C', lower='x', digit='3'),
    Trio(upper='B', lower='y', digit='1'),
    Trio(upper='A', lower='x', digit='2'),
    Trio(upper='C', lower='y', digit='3'),
    Trio(upper='A', lower='y', digit='1'),
    Trio(upper='C', lower='y', digit='2'),
    Trio(upper='C', lower='z', digit='3'),
    Trio(upper='C', lower='y', digit='1'),
    Trio(upper='C', lower='z', digit='2'),
    Trio(upper='A', lower='y', digit='3'),
    Trio(upper='B', lower='x', digit='1'),
    Trio(upper='B', lower='z', digit='2'),
    Trio(upper='B', lower='y', digit='3'),
    Trio(upper='C', lower='z', digit='1'),
    Trio(upper='C', lower='x', digit='2'),
    Trio(upper='B', lower='z', digit='3'),
    Trio(upper='C', lower='x', digit='1'),
    Trio(upper='B', lower='x', digit='2'),
    Trio(upper='A', lower='x', digit='3'),
    Trio(upper='A', lower='z', digit='1'),
    Trio(upper='B', lower='y', digit='2'),
    Trio(upper='B', lower='x', digit='3')
]

As you can see, the digit attribute seems to have stayed frozen and everything else has jumbled. What has actually happened is that everything with digit == 1 has been shuffled among themselves, and similarly for digit == 2 and digit == 3.

whisker.random.sort_indexes(x: List[Any]) → List[int]

Returns the indexes of x in an order that would sort x by value.

See https://stackoverflow.com/questions/7851077/how-to-return-index-of-a-sorted-list

whisker.rawsocketclient

whisker/rawsocketclient.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Framework for Whisker Python clients using raw sockets.

CONSIDER USING THE TWISTED OR QT FRAMEWORKS INSTEAD.

• Created: 18 Aug 2011.
• Last update: 10 Feb 2016

class whisker.rawsocketclient.WhiskerRawSocketClient

Basic Whisker class, in which clients do all the work via raw network sockets.

(Not sophisticated. Use whisker.twistedclient.WhiskerTwistedTask instead.)

connect_both_ports(server: str, mainport: Union[str, int]) → bool

Connect the main and immediate ports to the server.

connect_immediate(server: str, portstring: Union[str, int], code: str) → bool

Connect the immediate port to the server.

connect_main(server: str, portstring: Union[str, int]) → bool

Connect the main port to the server.

getlines_immsock() → Generator[[str, None], None]

Yield a set of lines from the immediate socket.

getlines_mainsock() → Generator[[str, None], None]

Yield a set of lines from the main socket.

log_out() → None

Shut down the connection to Whisker.

send(s: str) → None

Send something to the server on the main socket, with a trailing newline.

send_immediate(s: str) → str

Send a command to the server on the immediate socket, and retrieve its reply.

classmethod set_verbose_logging(verbose: bool) → None

Set the Python log level.

whisker.socket

whisker/socket.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Low-level network socket functions.

whisker.socket.get_port(x: Union[str, int]) → int

Works out an integer TCP/IP port number.

Parameters:

x – port number or name

Returns:

port number

Raises:

• ValueError – bad value
• TypeError – bad type

whisker.socket.socket_receive(sock: socket.socket, bufsize: int = 4096) → str

Receives ASCII data from a socket and returns a string.

Parameters:

• sock – TCP/IP socket
• bufsize – buffer size

Returns:

data as a string

whisker.socket.socket_send(sock: socket.socket, data: str) → int

Sends some of the data to a network socket, encoded via ASCII, and returns the number of bytes actually sent.

See https://stackoverflow.com/questions/34252273.

whisker.socket.socket_sendall(sock: socket.socket, data: str) → None

Sends all the data specified to a network socket, encoded via ASCII.

See https://stackoverflow.com/questions/34252273.

whisker.sqlalchemy

whisker/sqlalchemy.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

SQLAlchemy helper functions for Whisker tasks.

whisker.sqlalchemy.database_is_mysql(dbsettings: Dict[str, str]) → bool

Checks the URL in dbsettings['url']: is it a MySQL database?

whisker.sqlalchemy.database_is_postgresql(dbsettings: Dict[str, str]) → bool

Checks the URL in dbsettings['url']: is it a PostgreSQL database?

whisker.sqlalchemy.database_is_sqlite(dbsettings: Dict[str, str]) → bool

Checks the URL in dbsettings['url']: is it an SQLite database?

whisker.sqlalchemy.get_database_engine(settings: Dict[str, Any], unbreak_sqlite_transactions: bool = True, pool_pre_ping: bool = True) → sqlalchemy.engine.base.Engine

Get an SQLAlchemy database Engine from a simple definition.

Parameters:

• settings –

  a dictionary with the following keys:

  • url: a string
  • echo: a boolean
  • connect_args: a dictionary

  All are passed to SQLAlchemy’s create_engine() function.

• unbreak_sqlite_transactions – hook in events to unbreak SQLite transaction support? (Detailed in sqlalchemy/dialects/sqlite/pysqlite.py; see “Serializable isolation / Savepoints / Transactional DDL”.)
• pool_pre_ping – boolean; requires SQLAlchemy 1.2

Returns:

an SQLAlchemy Engine

whisker.sqlalchemy.get_database_engine_session_thread_scope(settings: Dict[str, Any], readonly: bool = False, autoflush: bool = True) → Tuple[sqlalchemy.engine.base.Engine, sqlalchemy.orm.session.Session]

Gets a thread-scoped SQLAlchemy Engine and Session.

Parameters:

• settings – passed to get_database_engine()
• readonly – make the session read-only?
• autoflush – passed to sessionmaker()

Returns:

(engine, session)

Return type:

tuple

whisker.sqlalchemy.get_database_session_thread_scope(*args, **kwargs) → sqlalchemy.orm.session.Session

Gets a thread-scoped SQLAlchemy Session.

Parameters:

• args – positional arguments to get_database_engine_session_thread_scope()
• kwargs – keyword arguments to get_database_engine_session_thread_scope()

Returns:

the session

whisker.sqlalchemy.get_database_session_thread_unaware(settings: Dict[str, Any]) → sqlalchemy.orm.session.Session

Returns an SQLAlchemy database session.

Warning

DEPRECATED: this function is not thread-aware.

Parameters:settings – passed to get_database_engine() Returns:an SQLAlchemy Session

whisker.sqlalchemy.noflush_readonly(*args, **kwargs) → None

Does nothing, and is thereby used to block a database session flush.

whisker.sqlalchemy.session_scope_thread_unaware(settings: Dict[str, Any]) → Generator[[sqlalchemy.orm.session.Session, None], None]

Context manager to provide an SQLAlchemy database session (which executes a COMMIT on success or a ROLLBACK on failure).

Warning

DEPRECATED: this function is not thread-aware.

Parameters:settings – passed to get_database_session_thread_unaware() Yields:an SQLAlchemy Session

whisker.sqlalchemy.session_thread_scope(settings: Dict[str, Any], readonly: bool = False) → Generator[[sqlalchemy.orm.session.Session, None], None]

Context manager to provide a thread-safe SQLAlchemy database session (which executes a COMMIT on success or a ROLLBACK on failure).

Parameters:

• settings – passed to get_database_session_thread_scope()
• readonly – passed to get_database_session_thread_scope()

Yields:

an SQLAlchemy Session

whisker.test_rawsockets

whisker/test_rawsockets.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Command-line tool to test the Whisker raw-socket client.

whisker.test_rawsockets.main() → None

Command-line parser. See --help for details.

whisker.test_rawsockets.test_whisker(server: str, port: int = 3233, verbose_network: bool = True) → None

Tests the Whisker raw-socket client.

Parameters:

• server – Whisker server IP address or hostname
• port – main Whisker port number
• verbose_network – be verbose about network traffic

whisker.test_twisted

whisker/test_twisted.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Command-line tool to test the Whisker Twisted client.

class whisker.test_twisted.MyWhiskerTwistedTask(display_num: int, audio_num: int, input_line: str, output_line: str, media_dir: str, bitmap: str, video: str, wav: str)

Class deriving from whisker.twistedclient.WhiskerTwistedTask to demonstrate the Twisted Whisker client.

fully_connected() → None

Called when the server is fully connected. Set up the “task”.

incoming_event(event: str, timestamp: int = None) → None

Responds to incoming events from Whisker.

whisker.test_twisted.main() → None

Command-line parser. See --help for details.

whisker.twistedclient

whisker/twistedclient.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Event-driven framework for Whisker Python clients using Twisted.

• Created: 18 Aug 2011
• Last update: 10 Feb 2016

class whisker.twistedclient.WhiskerImmSocket(task: whisker.twistedclient.WhiskerTwistedTask)

Whisker Twisted immediate socket handler.

Uses raw sockets.

Parameters:task – instance of WhiskerTwistedTask

connect(server: str, port: int) → None

Connects the Whisker immediate socket.

Parameters:

• server – server hostname/IP address
• port – immediate port number

getlines_immsock() → Generator[[str, None], None]

Generates lines from the immediate socket.

Yields:lines from the socket

send_and_get_reply(*args) → str

Builds its arguments into a string; sends it to Whisker via the immediate socket; gets the reply; returns it.

class whisker.twistedclient.WhiskerMainPortFactory(task: whisker.twistedclient.WhiskerTwistedTask)

A Protocol factory for the Whisker main port.

Parameters:task – instance of WhiskerTwistedTask

buildProtocol(addr: str) → Optional[whisker.twistedclient.WhiskerMainPortProtocol]

Build and return the protocol.

clientConnectionFailed(connector: twisted.internet.tcp.Connector, reason: str) → None

Client connection failed. Stop the reactor.

clientConnectionLost(connector: twisted.internet.tcp.Connector, reason: str) → None

If we get disconnected, reconnect to server.

class whisker.twistedclient.WhiskerMainPortProtocol(task: whisker.twistedclient.WhiskerTwistedTask, encoding: str = 'ascii')

Line-based Twisted protocol for the Whisker main port.

Parameters:

• task – instance of WhiskerTwistedTask
• encoding – encoding to use; normally "ascii"

connectionMade() → None

Called when the main port is connected.

lineReceived(data: bytes) → None

Called when data is received on the main port. Sends it to WhiskerTwistedTask.incoming_message() via self.task.

Parameters:data – bytes

rawDataReceived(data: bytes) → None

Raw data received. Unused; we use lineReceived() instead.

send(data: str) → None

Encodes and sends data to the main port.

class whisker.twistedclient.WhiskerTwistedTask

Base class for Whisker clients using the Twisted socket system.

• Contains self.whisker, an instance of whisker.api.WhiskerApi.
• Specimen usage: see test_twisted.py.

connect(server: str, port: Union[str, int]) → None

Connects to the Whisker server.

Parameters:

• server – Whisker server hostname/IP address
• port – Whisker main TCP/IP port number

fully_connected() → None

The Whisker server is now fully connected.

Override this, e.g. to start the task.

incoming_client_message(fromclientnum: int, msg: str, timestamp: int = None) → None

Client message (from another client) received.

Override this function.

Parameters:

• fromclientnum – source client number
• msg – message
• timestamp – server timestamp (ms)

incoming_error(msg: str) → None

Error report received from Whisker server.

Parameters:msg – message

incoming_event(event: str, timestamp: int = None) → None

General Whisker event received.

Override this function.

Parameters:

• event – event
• timestamp – server timestamp (ms)

incoming_info(msg: str) → None

Information message received from Whisker server.

Parameters:msg – message

incoming_key_event(key: str, depressed: bool, document: str, timestamp: int = None) → None

Keyboard event received.

Override this function.

Parameters:

• key – which key?
• depressed – was it depressed (or released)?
• document – source display document
• timestamp – server timestamp (ms)

incoming_message(msg: str) → None

Processes an incoming message from the Whisker server (via the main socket).

incoming_syntax_error(msg: str) → None

Syntax error report received from Whisker server.

Parameters:msg – message

incoming_warning(msg: str) → None

Warning received from Whisker server.

Parameters:msg – message

send(*args) → None

Builds its arguments into a string and sends that as a command to the Whisker server via the main socket.

send_and_get_reply(*args) → Optional[str]

Builds its arguments into a string, sends that as a command to the Whisker server via the immediate socket, blocks and waits for the reply, and returns that reply. (Returns None if not connected.)

classmethod set_verbose_logging(verbose: bool) → None

Sets the Python log level for this module.

Parameters:verbose – be verbose?

whisker.version

whisker/version.py

---

Copyright © 2011-2020 Rudolf Cardinal (rudolf@pobox.com).

This file is part of the Whisker Python client library.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

---

Library version constants.

External libraries

This project also depends on:

• arrow: https://arrow.readthedocs.io/
• attrdict: https://pypi.org/project/attrdict/
• cardinal_pythonlib: https://cardinalpythonlib.readthedocs.io/
• colorama: https://pypi.org/project/colorama/
• colorlog: https://pypi.org/project/colorlog/
• dataset: https://dataset.readthedocs.io/
• PyQt5: https://www.riverbankcomputing.com/software/pyqt/intro
• Twisted: https://twistedmatrix.com/trac/
• pyyaml: https://pyyaml.org/
• SQLAlchemy: https://www.sqlalchemy.org/

For development of the library itself, it also uses:

• sphinx: http://www.sphinx-doc.org/
• sphinx_rtd_theme: https://sphinx-rtd-theme.readthedocs.io/
• twine: https://github.com/pypa/twine

Indices and tables

• Index
• Module Index
• Search Page