#!/usr/bin/env python3
# encoding: UTF-8
# This file is part of turberfield.
#
# Turberfield is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Turberfield is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with turberfield. If not, see <http://www.gnu.org/licenses/>.
import asyncio
from collections.abc import Callable
from collections.abc import MutableSequence
import logging
import sys
import textwrap
import time
import wave
import pkg_resources
import simpleaudio
from turberfield.dialogue.model import Model
from turberfield.dialogue.model import SceneScript
from turberfield.dialogue.schema import SchemaBase
from turberfield.utils.assembly import Assembly
from turberfield.utils.db import Connection
from turberfield.utils.db import Creation
[docs]class TerminalHandler:
"""
The default handler for events from scene script files.
It generates output for a console terminal.
The class is written to be a callable, stateful object.
Its `__call__` method delegates to handlers specific to each type of event.
You can subclass it and override those methods to suit your own application.
:param terminal: A stream object.
:param str dbPath: An optional URL to the internal database.
:param float pause: The time in seconds to pause on a line of dialogue.
:param float dwell: The time in seconds to dwell on a word of dialogue.
:param log: An optional log object.
"""
pause = 1.5
dwell = 0.2
[docs] @staticmethod
def handle_audio(obj, wait=False):
"""Handle an audio event.
This function plays an audio file.
Currently only `.wav` format is supported.
:param obj: An :py:class:`~turberfield.dialogue.model.Model.Audio`
object.
:param bool wait: Force a blocking wait until playback is complete.
:return: The supplied object.
"""
fp = pkg_resources.resource_filename(obj.package, obj.resource)
data = wave.open(fp, "rb")
nChannels = data.getnchannels()
bytesPerSample = data.getsampwidth()
sampleRate = data.getframerate()
nFrames = data.getnframes()
framesPerMilliSecond = nChannels * sampleRate // 1000
offset = framesPerMilliSecond * obj.offset
duration = nFrames - offset
duration = min(
duration,
framesPerMilliSecond * obj.duration if obj.duration is not None else duration
)
data.readframes(offset)
frames = data.readframes(duration)
for i in range(obj.loop):
waveObj = simpleaudio.WaveObject(frames, nChannels, bytesPerSample, sampleRate)
playObj = waveObj.play()
if obj.loop > 1 or wait:
playObj.wait_done()
return obj
[docs] def handle_interlude(
self, obj, folder, index, ensemble, branches,
loop=None, **kwargs
):
"""Handle an interlude event.
Interlude functions permit branching. They return a folder which the
application can choose to adopt as the next supplier of dialogue.
This handler calls the interlude with the supplied arguments and
returns the result.
:param obj: A callable object.
:param folder: A
:py:class:`~turberfield.dialogue.model.SceneScript.Folder` object.
:param int index: Indicates which scene script in the folder
is being processed.
:param ensemble: A sequence of Python objects.
:param branches: A sequence of
:py:class:`~turberfield.dialogue.model.SceneScript.Folder` objects.
from which to pick a branch in the action.
:return: A :py:class:`~turberfield.dialogue.model.SceneScript.Folder`
object.
"""
if obj is None:
return folder
else:
return obj(folder, index, ensemble, branches, loop=loop, **kwargs)
[docs] def handle_line(self, obj):
"""Handle a line event.
This function displays a line of dialogue. It generates a blocking wait
for a period of time calculated from the length of the line.
:param obj: A :py:class:`~turberfield.dialogue.model.Model.Line` object.
:return: The supplied object.
"""
if obj.persona is None:
return obj
name = getattr(obj.persona, "_name", "")
print(
textwrap.indent(
"{t.normal}{name}".format(name=name, t=self.terminal),
" " * 2
),
end="\n",
file=self.terminal.stream
)
print(
textwrap.indent(
"{t.normal}{obj.text}".format(
obj=obj, t=self.terminal
),
" " * 10
),
end="\n" * 2,
file=self.terminal.stream
)
interval = self.pause + self.dwell * obj.text.count(" ")
time.sleep(interval)
return obj
[docs] def handle_memory(self, obj):
"""Handle a memory event.
This function accesses the internal database. It writes a record
containing state information and an optional note.
:param obj: A :py:class:`~turberfield.dialogue.model.Model.Memory`
object.
:return: The supplied object.
"""
if obj.subject is not None:
with self.con as db:
rv = SchemaBase.note(
db,
obj.subject,
obj.state,
obj.object,
text=obj.text,
html=obj.html,
)
return obj
[docs] def handle_property(self, obj):
"""Handle a property event.
This function will set an attribute on an object if the event requires
it.
:param obj: A :py:class:`~turberfield.dialogue.model.Model.Property`
object.
:return: The supplied object.
"""
if obj.object is not None:
try:
setattr(obj.object, obj.attr, obj.val)
except AttributeError as e:
self.log.error(". ".join(getattr(e, "args", e) or e))
print(
"{t.dim}{obj.object._name}.{obj.attr} = {obj.val!s}{t.normal}".format(
obj=obj, t=self.terminal
),
end="\n" * 2,
file=self.terminal.stream
)
return obj
[docs] def handle_scene(self, obj):
"""Handle a scene event.
This function applies a blocking wait at the start of a scene.
:param obj: A :py:class:`~turberfield.dialogue.model.Model.Shot`
object.
:return: The supplied object.
"""
print(
"{t.dim}{scene}{t.normal}".format(
scene=obj.scene.capitalize(), t=self.terminal
),
end="\n" * 3,
file=self.terminal.stream
)
time.sleep(self.pause)
return obj
[docs] def handle_scenescript(self, obj):
"""Handle a scene script event.
:param obj: A :py:class:`~turberfield.dialogue.model.SceneScript.Folder`
object.
:return: The supplied object.
"""
self.log.debug(obj.fP)
return obj
[docs] def handle_shot(self, obj):
"""Handle a shot event.
:param obj: A :py:class:`~turberfield.dialogue.model.Model.Shot` object.
:return: The supplied object.
"""
print(
"{t.dim}{shot}{t.normal}".format(
shot=obj.name.capitalize(), t=self.terminal
),
end="\n" * 3,
file=self.terminal.stream
)
return obj
def handle_creation(self):
with self.con as db:
rv = Creation(
*SchemaBase.tables.values()
).run(db)
db.commit()
self.log.info("Created {0} tables in {1}.".format(len(rv), self.dbPath))
return rv
def handle_references(self, obj):
with self.con as db:
rv = SchemaBase.populate(db, obj)
self.log.info("Populated {0} rows.".format(rv))
return rv
def __init__(
self, terminal, dbPath=None,
pause=pause, dwell=dwell, log=None
):
self.terminal = terminal
self.dbPath = dbPath
self.pause = pause
self.dwell = dwell
self.log = log or logging.getLogger("turberfield.dialogue.handle")
self.shot = None
self.con = Connection(**Connection.options(paths=[dbPath] if dbPath else []))
self.handle_creation()
def __call__(self, obj, *args, loop, **kwargs):
if isinstance(obj, Model.Line):
try:
yield self.handle_line(obj)
except AttributeError:
pass
elif isinstance(obj, Model.Audio):
yield self.handle_audio(obj)
elif isinstance(obj, Model.Memory):
yield self.handle_memory(obj)
elif isinstance(obj, Model.Property):
yield self.handle_property(obj)
elif isinstance(obj, Model.Shot):
if self.shot is None or obj.scene != self.shot.scene:
yield self.handle_scene(obj)
if self.shot is None or obj.name != self.shot.name:
yield self.handle_shot(obj)
else:
yield obj
self.shot = obj
elif isinstance(obj, SceneScript):
yield self.handle_scenescript(obj)
elif asyncio.iscoroutinefunction(obj):
raise NotImplementedError
elif isinstance(obj, MutableSequence):
yield self.handle_references(obj)
elif (obj is None or isinstance(obj, Callable)) and len(args) == 4:
yield self.handle_interlude(obj, *args, loop=loop, **kwargs)
else:
yield obj
class CGIHandler(TerminalHandler):
def handle_audio(self, obj):
path = pkg_resources.resource_filename(obj.package, obj.resource)
pos = path.find("lib", len(sys.prefix))
if pos != -1:
print(
"event: audio",
"data: ../{0}\n".format(path[pos:]),
sep="\n",
end="\n",
file=self.terminal.stream
)
self.terminal.stream.flush()
return obj
def handle_line(self, obj):
if obj.persona is None:
return obj
print(
"event: line",
"data: {0}\n".format(Assembly.dumps(obj)),
sep="\n",
end="\n",
file=self.terminal.stream
)
self.terminal.stream.flush()
interval = self.pause + self.dwell * obj.text.count(" ")
time.sleep(interval)
return obj
def handle_property(self, obj):
log = logging.getLogger("turberfield")
log.info(obj)
if obj.object is not None:
try:
setattr(obj.object, obj.attr, obj.val)
except AttributeError as e:
self.log.error(". ".join(getattr(e, "args", e) or e))
print(
"event: property",
"data: {0}\n".format(Assembly.dumps(obj)),
sep="\n",
end="\n",
file=self.terminal.stream
)
self.terminal.stream.flush()
time.sleep(self.pause)
return obj
def handle_scene(self, obj):
time.sleep(self.pause)
return obj
def handle_scenescript(self, obj):
return obj
def handle_shot(self, obj):
return obj