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:
and SQLAlchemy:
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:
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
renamedlogsupport
.
v0.3.5: 2016-11-25
- Python type hints.
- Write
exit_on_exception
exceptions to log, not viaprint()
. whisker.qtclient.WhiskerOwner
offers newpingack_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 usesinspect.getargspec()
, which chokes withValueError: 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
andTwisted
added as requirements- Sphinx docs
- updated signal debugging for PyQt5 in
debug_qt.py
- general tidy
ValidationError
removed fromwhisker.qt
; is inwhisker.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 toWhiskerTwistedTask
and the other toWhiskerQtTask
. Also renamedWhisker
toWhiskerRawSocketClient
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 percardinal_pythonlib
) to see if that fixed the autodoc/RTD problem; no - no help from changing Sphinx theme either
- no help from changing
sys.path
fromconf.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
- removed
- renamed
layer_index
tolayer_key
and changed its hint inwhisker.random.ShuffleLayerMethod
.
v1.2.0 to 1.3.0: 2020-02-09
- Added
whisker.__version__
. whisker.convenience.load_config_or_die()
has a newconfig_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:
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
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>)[source]¶ Represents a Whisker Brush graphics object.
-
class
whisker.api.
Pen
(width: int = 1, colour: Tuple[int, int, int] = (255, 255, 255), style: whisker.api.PenStyle = <PenStyle.solid: 0>)[source]¶ Represents a Whisker Pen graphics object.
-
class
whisker.api.
Rectangle
(left: int, top: int, width: int = None, height: int = None, right: int = None, bottom: int = None)[source]¶ Represents a Whisker Rectangle graphics object.
The origin is at the top left, as per Whisker.
-
class
whisker.api.
WhiskerApi
(whisker_immsend_get_reply_fn: Callable[..., str], sysevent_prefix: str = 'sys_', **kwargs)[source]¶ 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][source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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) to100
(full volume)
Returns: success
-
audio_stop_sound
(device: str, sound: str) → bool[source]¶ 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[source]¶ Whisker command: unload all sounds from an audio device.
-
audio_unload_sound
(device: str, sound: str) → bool[source]¶ 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][source]¶ Whisker command: ask the server for an authentication challenge, using a particular software package name and client name.
-
authenticate_provide_response
(response: str) → bool[source]¶ Whisker command: respond to the server’s authentication challenge with an appropriate cryptographic answer.
-
broadcast
(*args) → bool[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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_event_callback
(event: str, callback: Callable[..., None] = None) → None[source]¶ Cancels a callback.
Parameters: - event – Whisker event name
- callback – callback function to remove from callback list
-
display_add_obj
(doc: str, obj: str, obj_type: str, *parameters) → bool[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ Whisker command: adds a Video display object. See Whisker help (http://www.whiskercontrol.com) for details.
-
display_blank
(device: str) → bool[source]¶ Whisker command: blank a display device (remove any document).
-
display_bring_to_front
(doc: str, obj: str) → bool[source]¶ 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[source]¶ 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][source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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_delete_device
(device: str) → bool[source]¶ Whisker command: deletes a display device that you’ve created with
display_create_device()
.
-
display_delete_obj
(doc: str, obj: str) → bool[source]¶ Whisker command: delete a display object from a document.
-
display_event_coords
(on: bool) → bool[source]¶ Whisker command: enable/disable
x
/y
coordinates being sent with relevant touchscreen/mouse events.
-
display_get_document_size
(doc: str) → Optional[Tuple[int, int]][source]¶ Whisker command: get the logical size of a display document.
Returns a
(width, height)
tuple in pixels, orNone
.
-
display_get_object_extent
(doc: str, obj: str) → Optional[whisker.api.Rectangle][source]¶ Whisker command: get the rectangle describing the extend of a display object (in logical coordinates).
Returns a
Rectangle
, orNone
.
-
display_get_size
(device: str) → Optional[Tuple[int, int]][source]¶ Whisker command: get the size of a display device.
Parameters: device – display device number/name Returns: (width, height)
in pixels, orNone
Return type: tuple
-
display_keyboard_events
(doc: str, key_event_type: whisker.api.KeyEventType = <KeyEventType.down: 1>) → bool[source]¶ 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[source]¶ Whisker command: set document scaling for a display device.
-
display_send_to_back
(doc: str, obj: str) → bool[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ Whisker command: show any pending changes (see
display_cache_changes()
).
-
display_show_document
(device: str, doc: str) → bool[source]¶ Whisker command: show a document on a display device.
-
flash_line_ping_pong
(line: str, on_now: bool, timing_sequence: List[int]) → None[source]¶ 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[source]¶ 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_command_boolean
(*args) → bool[source]¶ Return a boolean answer from an immediate socket command.
-
get_response_with_timestamp
(*args) → Tuple[str, Optional[int]][source]¶ Return the answer from an immediate socket command, with a timestamp.
-
get_server_time_ms
() → int[source]¶ Whisker command: get the time on the server, from its millisecond clock.
-
line_clear_event
(event: str) → bool[source]¶ Whisker command: clear (cancel) a line event or events by name.
-
line_clear_event_by_line
(line: str, event_type: whisker.api.LineEventType) → bool[source]¶ Whisker command: clear (cancel) a line event by line number/name and transition type.
-
line_clear_safety_timer
(line: str) → bool[source]¶ Whisker command: clears a safety timer for a line.
-
line_read_state
(line: str) → Optional[bool][source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ Whisker command: turns a digital output line on/off.
-
log_set_options
(events: bool = True, key_events: bool = True, client_client: bool = True, comms: bool = False, signature: bool = True) → bool[source]¶ 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?
-
permit_client_messages
(permit: bool) → bool[source]¶ Whisker command: enable/disable client messages.
-
process_backend_event
(event: str) → bool[source]¶ 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.
-
send_after_delay
(delay_ms: int, msg: str, event: str = '') → None[source]¶ 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[source]¶ Whisker command: send a message to another client.
-
set_media_directory
(directory: str) → bool[source]¶ Whisker command: set the server’s media directory.
-
timer_set_event
(event: str, duration_ms: int, reload_count: int = 0) → bool[source]¶ 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?
-
video_get_duration_ms
(doc: str, video: str) → Optional[int][source]¶ 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][source]¶ 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_seek_absolute
(doc: str, video: str, time_ms: int) → bool[source]¶ 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[source]¶ 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).
- whisker_immsend_get_reply_fn – A function that must take arguments
-
whisker.api.
assert_ducktype_colour
(colour: Any) → None[source]¶ If the parameter does not pass the test
is_ducktype_colour()
, raiseValueError
.
-
whisker.api.
assert_ducktype_pos
(pos: Any) → None[source]¶ If the parameter does not pass the test
is_ducktype_pos()
, raiseValueError
.
-
whisker.api.
is_ducktype_colour
(colour: Any) → bool[source]¶ Does the parameter look like a colour
(red, green blue)
tuple, with values for each in the range 0-255?
-
whisker.api.
is_ducktype_nonnegative_int
(x: Any) → bool[source]¶ Does the parameter look like an integer that is 0 or greater?
-
whisker.api.
is_ducktype_pos
(pos: Any) → bool[source]¶ Does the parameter look like a position
(x, y)
tuple, with values that can be made numeric?
-
whisker.api.
msg_from_args
(*args) → str[source]¶ 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.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
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)[source]¶ 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?
-
class
whisker.callback.
CallbackHandler
[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ Adds a single-shot callback for the specified 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
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
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[source]¶ 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 aNone
value would be returned, raiseValueError
)
Returns: user-supplied value
-
whisker.convenience.
connect_to_db_using_attrdict
(database_url: str, show_url: bool = False, engine_kwargs: Dict[str, Any] = None)[source]¶ Connects to a
dataset
database, and usesAttrDict
as the row type, soAttrDict
objects come back out again.
-
whisker.convenience.
insert_and_set_id
(table: dataset.table.Table, obj: Dict[str, Any], idfield: str = 'id') → Any[source]¶ Inserts an object into a
dataset.Table
and ensures that the primary key (PK) is written back to the object, in itsidfield
.(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[source]¶ 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[source]¶ 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][source]¶ 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 SQLWHERE
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
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.
-
whisker.debug_qt.
debug_object
(obj: SimpleClass) → None[source]¶ Writes a debug log message within information about the QObject.
-
whisker.debug_qt.
debug_thread
(thread: <MagicMock id='140405796558440'>) → None[source]¶ 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[source]¶ Call this to enable PySide/Qt signal debugging. This will trap all
connect()
anddisconnect()
calls, calling the user-supplied functions first and then the real things.
-
whisker.debug_qt.
enable_signal_debugging_simply
() → None[source]¶ Enables Qt signal debugging for
connect()
andemit()
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
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
[source]¶ Whisker is improperly configured; normally due to a missing library.
-
exception
whisker.exceptions.
ValidationError
(message: str)[source]¶ Exception to represent failure to validate user input.
-
exception
whisker.exceptions.
WhiskerCommandFailed
[source]¶ 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
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)[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ Override this if you need to prevent rows being deleted.
- session – SQLAlchemy
-
exception
whisker.qt.
EditCancelledException
[source]¶ Exception to represent that the user cancelled editing.
-
class
whisker.qt.
GarbageCollector
(parent: SimpleClass, debug: bool = False, interval_ms=10000)[source]¶ 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
-
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)[source]¶ 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 likeh1 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 itslessThan()
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 objectobj
, the view will callthing = getattr(obj, attr_or_func)
. Ifthing
is callable (i.e. is a member function), the view will displaystr(thing())
; otherwise, it will displaystr(thing)
. - session – SQLAlchemy
Session
- default_sort_column_name – column to sort by to begin with, or
None
to respect the original order ofdata
- 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)[source]¶ Qt override.
Parameters: - parent – parent object, specified by
QModelIndex
- args – unnecessary? Certainly unused.
- kwargs – unnecessary? Certainly unused.
Returns: number of columns
- parent – parent object, specified by
-
data
(index: <MagicMock id='140405793012872'>, role: int = <MagicMock name='mock.DisplayRole' id='140405871638344'>) → Optional[str][source]¶ 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
- index –
-
get_default_sort
() → Tuple[int, int][source]¶ 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][source]¶ 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)[source]¶ 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
- parent – parent object, specified by
-
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)[source]¶ 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'>][source]¶ Gets the model index of the selected item.
Returns: a QModelIndex
orNone
-
class
whisker.qt.
GenericListModel
(data, session: sqlalchemy.orm.session.Session, parent: SimpleClass = None, **kwargs)[source]¶ 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][source]¶ 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
- index –
-
rowCount
(parent: <MagicMock id='140405793012872'> = <MagicMock name='mock()' id='140405871621792'>, *args, **kwargs) → int[source]¶ 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
- parent – parent object, specified by
-
class
whisker.qt.
ModalEditListView
(session: sqlalchemy.orm.session.Session, modal_dialog_class: Type[whisker.qt.TransactionalEditDialogMixin], *args, readonly: bool = False, **kwargs)[source]¶ 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
- session – SQLAlchemy
-
class
whisker.qt.
RadioGroup
(value_text_tuples: Iterable[Tuple[str, Any]], default: Any = None)[source]¶ Framework for radio buttons.
Parameters: - value_text_tuples – list of
(value, text)
tuples, wherevalue
is the value stored to Python object andtext
is the text shown to the user - default – default value
Adds its button widgets to the specified Qt layout.
- value_text_tuples – list of
-
class
whisker.qt.
StatusMixin
(name: str, logger: logging.Logger, thread_info: bool = True, caller_info: bool = True, **kwargs)[source]¶ 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')[source]¶ 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 Pythonlogging
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
-
class
whisker.qt.
TransactionalEditDialogMixin
(session: sqlalchemy.orm.session.Session, obj: object, layout: <MagicMock id='140405794232360'>, readonly: bool = False, **kwargs)[source]¶ Mixin for a config-editing dialogue. Use it as:
class MyEditingDialog(TransactionalEditDialogMixin, QDialog): pass
Wraps the editing in a
SAVEPOINT
transaction. The caller must stillcommit()
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[source]¶ 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[source]¶ 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 thesubtransactions
flag; these are virtual transactions handled by SQLAlchemy.Alternatively one can use
begin_nested()
orbegin(nested=True)
, which usesSAVEPOINT
.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:
- The bugs are detailed in
sqlalchemy/dialects/sqlite/pysqlite.py
; see “Serializable isolation / Savepoints / Transactional DDL”
- The bugs are detailed in
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)[source]¶ 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][source]¶ 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
-
edit
(index: <MagicMock id='140405793012872'>, trigger: <MagicMock name='mock.EditTrigger' id='140405793084752'>, event: <MagicMock id='140405796557320'>) → bool[source]¶ 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’sState
is nowEditingState
; otherwiseFalse
- index –
-
edit_by_modelindex
(index: <MagicMock id='140405793012872'>, readonly: bool = None) → None[source]¶ Edits the specified object.
Parameters: - index –
QModelIndex
of the object to edit - readonly – read-only view? If
None
(the default), uses the setting fromself
.
- index –
-
edit_selected
(readonly: bool = None) → None[source]¶ Edits the currently selected object.
Parameters: readonly – read-only view? If None
(the default), uses the setting fromself
.
-
get_selected_modelindex
() → Optional[<MagicMock id='140405793012872'>][source]¶ Returns the
QModelIndex
of the currently selected row, orNone
.Should be overridden in derived classes.
-
get_selected_object
() → object[source]¶ Returns the SQLAlchemy ORM object that’s currently selected, or
None
.
-
get_selected_row_index
() → Optional[int][source]¶ Gets the index of the currently selected row.
Returns: index, or None
-
go_to
(row: Optional[int]) → None[source]¶ 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[source]¶ 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[source]¶ 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[source]¶ 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.
-
remove_by_index
(row_index: int, delete_from_session: bool = True) → None[source]¶ 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
)
-
whisker.qt.
exit_on_exception
(func)[source]¶ 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.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
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()
toQTcpSocket(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…
- change
- Anyway, bug was this:
- Source:
-
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)[source]¶ 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
-
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)[source]¶ 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)[source]¶ 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:
- main socket listener (
WhiskerMainSocketListener
, as the instanceself.mainsock
, in the threadself.mainsockthread
); - task (
WhiskerQtTask
, as the instanceself.task
in the threadself.taskthread
) + immediate socket blocking handler (WhiskerController
, as the instanceself.controller
in the threadself.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
- main socket listener (
-
class
whisker.qtclient.
WhiskerQtTask
(parent: SimpleClass = None, name: str = 'whisker_task', **kwargs)[source]¶ 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[source]¶ The
WhiskerOwner
makes this slot get called when theWhiskerController
is connected.You should override this.
-
set_controller
(controller: whisker.qtclient.WhiskerController) → None[source]¶ Called by
WhiskerOwner
. No need to override.
- Lives in Whisker thread (B) (see
-
whisker.qtclient.
disable_nagle
(socket: <MagicMock id='140405852664832'>) → None[source]¶ 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[source]¶ Returns a textual description of the last error on the specified Qt socket.
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
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)[source]¶ 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 shufflerandom_shuffle_indexes
: plain shuffle; seerandom_shuffle_indexes()
dwor_shuffle_indexes
: DWOR-style shuffle (will need alambda
for itsmultiplier parameter
; seedwor_shuffle_indexes()
block_shuffle_indexes_by_value
: aggregate items into blocks, defined by value, and shuffle those blocks; seeblock_shuffle_indexes_by_value()
sort_indexes
: not exactly shuffling! Seesort_indexes()
.reverse_sort_indexes
: not exactly shuffling! Seereverse_sort_indexes()
.
-
get_indexes_for_value
(x: List[Any], value: Any) → List[int][source]¶ Returns a list of indexes of
x
where its value (as defined by this layer) isvalue
.
- flat – take data as
-
whisker.random.
block_shuffle_by_attr
(x: List[Any], attrorder: List[str], start: int = None, end: int = None) → None[source]¶ 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, withinB
, thex
/y
/z
groups have been shuffled (and so on forA
andC
). Then, withinB.z
, the1
/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[source]¶ 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][source]¶ 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][source]¶ 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][source]¶ 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:
-
whisker.random.
get_dwor_list
(values: Iterable[Any], length: int, multiplier: int = 1) → List[Any][source]¶ 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][source]¶ Returns a list of indexes of
x
where its value isvalue
.
-
whisker.random.
get_unique_values
(iterable: Iterable[Any]) → List[Any][source]¶ 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 ofx
are themselves lists (perfectly common!), that givesTypeError: unhashable type: 'list'
.)
-
whisker.random.
last_index_of
(x: List[Any], value: Any) → int[source]¶ Gets the index of the last occurrence of
value
in the listx
.
-
whisker.random.
layered_shuffle
(x: List[Any], layers: List[whisker.random.ShuffleLayerMethod]) → None[source]¶ 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 theShuffleLayerMethod
(for example: “shufflex
in blocks based on the value ofx.someattr
”, or “shufflex
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))
- for each layer, it shuffles values of
-
whisker.random.
make_dwor_hat
(values: Iterable[Any], multiplier: int = 1) → List[Any][source]¶ 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][source]¶ Returns a list of indexes of
x
, randomly shuffled.
-
whisker.random.
reverse_sort_indexes
(x: List[Any]) → List[int][source]¶ Returns the indexes of
x
in an order that would reverse-sortx
by value.
-
whisker.random.
shuffle_list_chunks
(x: List[Any], chunksize: int) → None[source]¶ 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()
andcardinal_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[source]¶ Shuffles a segment of a list,
x[start:end]
, in place.Note that
start=None
means “from the beginning” andend=None
means “to the end”.
-
whisker.random.
shuffle_list_subset
(x: List[Any], indexes: List[int]) → None[source]¶ 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[source]¶ 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[source]¶ DEPRECATED:
layered_shuffle()
is more powerful.Shuffles a list
x
, in place, where list members are equal as judged by the attributeattrname
.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 withdigit == 1
has been shuffled among themselves, and similarly fordigit == 2
anddigit == 3
.
-
whisker.random.
sort_indexes
(x: List[Any]) → List[int][source]¶ Returns the indexes of
x
in an order that would sortx
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
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
[source]¶ 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[source]¶ Connect the main and immediate ports to the server.
-
connect_immediate
(server: str, portstring: Union[str, int], code: str) → bool[source]¶ Connect the immediate port to the server.
-
connect_main
(server: str, portstring: Union[str, int]) → bool[source]¶ Connect the main port to the server.
-
getlines_immsock
() → Generator[[str, None], None][source]¶ Yield a set of lines from the immediate socket.
-
getlines_mainsock
() → Generator[[str, None], None][source]¶ Yield a set of lines from the main socket.
-
send
(s: str) → None[source]¶ Send something to the server on the main socket, with a trailing newline.
-
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
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[source]¶ Works out an integer TCP/IP port number.
Parameters: x – port number or name
Returns: port number
Raises: ValueError
– bad valueTypeError
– bad type
-
whisker.socket.
socket_receive
(sock: socket.socket, bufsize: int = 4096) → str[source]¶ 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[source]¶ Sends some of the data to a network socket, encoded via ASCII, and returns the number of bytes actually sent.
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
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[source]¶ Checks the URL in
dbsettings['url']
: is it a MySQL database?
-
whisker.sqlalchemy.
database_is_postgresql
(dbsettings: Dict[str, str]) → bool[source]¶ Checks the URL in
dbsettings['url']
: is it a PostgreSQL database?
-
whisker.sqlalchemy.
database_is_sqlite
(dbsettings: Dict[str, str]) → bool[source]¶ 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[source]¶ Get an SQLAlchemy database
Engine
from a simple definition.Parameters: - settings –
a dictionary with the following keys:
url
: a stringecho
: a booleanconnect_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
- settings –
-
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][source]¶ Gets a thread-scoped SQLAlchemy
Engine
andSession
.Parameters: - settings – passed to
get_database_engine()
- readonly – make the session read-only?
- autoflush – passed to
sessionmaker()
Returns: (engine, session)
Return type: tuple
- settings – passed to
-
whisker.sqlalchemy.
get_database_session_thread_scope
(*args, **kwargs) → sqlalchemy.orm.session.Session[source]¶ 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
- args – positional arguments to
-
whisker.sqlalchemy.
get_database_session_thread_unaware
(settings: Dict[str, Any]) → sqlalchemy.orm.session.Session[source]¶ 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[source]¶ 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][source]¶ Context manager to provide an SQLAlchemy database session (which executes a
COMMIT
on success or aROLLBACK
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][source]¶ Context manager to provide a thread-safe SQLAlchemy database session (which executes a
COMMIT
on success or aROLLBACK
on failure).Parameters: - settings – passed to
get_database_session_thread_scope()
- readonly – passed to
get_database_session_thread_scope()
Yields: an SQLAlchemy
Session
- settings – passed to
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
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_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
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)[source]¶ Class deriving from
whisker.twistedclient.WhiskerTwistedTask
to demonstrate the Twisted Whisker client.
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
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)[source]¶ Whisker Twisted immediate socket handler.
Uses raw sockets.
Parameters: task – instance of WhiskerTwistedTask
-
connect
(server: str, port: int) → None[source]¶ Connects the Whisker immediate socket.
Parameters: - server – server hostname/IP address
- port – immediate port number
-
-
class
whisker.twistedclient.
WhiskerMainPortFactory
(task: whisker.twistedclient.WhiskerTwistedTask)[source]¶ A Protocol factory for the Whisker main port.
Parameters: task – instance of WhiskerTwistedTask
-
buildProtocol
(addr: str) → Optional[whisker.twistedclient.WhiskerMainPortProtocol][source]¶ Build and return the protocol.
-
-
class
whisker.twistedclient.
WhiskerMainPortProtocol
(task: whisker.twistedclient.WhiskerTwistedTask, encoding: str = 'ascii')[source]¶ Line-based Twisted protocol for the Whisker main port.
Parameters: - task – instance of
WhiskerTwistedTask
- encoding – encoding to use; normally
"ascii"
-
lineReceived
(data: bytes) → None[source]¶ Called when data is received on the main port. Sends it to
WhiskerTwistedTask.incoming_message()
viaself.task
.Parameters: data – bytes
-
rawDataReceived
(data: bytes) → None[source]¶ Raw data received. Unused; we use
lineReceived()
instead.
- task – instance of
-
class
whisker.twistedclient.
WhiskerTwistedTask
[source]¶ Base class for Whisker clients using the Twisted socket system.
- Contains
self.whisker
, an instance ofwhisker.api.WhiskerApi
. - Specimen usage: see
test_twisted.py
.
-
connect
(server: str, port: Union[str, int]) → None[source]¶ Connects to the Whisker server.
Parameters: - server – Whisker server hostname/IP address
- port – Whisker main TCP/IP port number
-
fully_connected
() → None[source]¶ 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[source]¶ 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[source]¶ Error report received from Whisker server.
Parameters: msg – message
-
incoming_event
(event: str, timestamp: int = None) → None[source]¶ General Whisker event received.
Override this function.
Parameters: - event – event
- timestamp – server timestamp (ms)
-
incoming_info
(msg: str) → None[source]¶ Information message received from Whisker server.
Parameters: msg – message
-
incoming_key_event
(key: str, depressed: bool, document: str, timestamp: int = None) → None[source]¶ 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[source]¶ Processes an incoming message from the Whisker server (via the main socket).
-
incoming_syntax_error
(msg: str) → None[source]¶ Syntax error report received from Whisker server.
Parameters: msg – message
-
incoming_warning
(msg: str) → None[source]¶ Warning received from Whisker server.
Parameters: msg – message
-
send
(*args) → None[source]¶ Builds its arguments into a string and sends that as a command to the Whisker server via the main socket.
- Contains
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
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/introTwisted
: 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