Source code for story_graph.models


import ast
import logging
import uuid

from asgiref.sync import async_to_sync, sync_to_async
from django.core.exceptions import ValidationError
from django.db import models, transaction
from django.db.models import Q, signals
from django.dispatch import receiver
from django.utils.translation import gettext as _
from django_stubs_ext.db.models import TypedModelMeta

from gencaster.distributor import GenCasterChannel

log = logging.getLogger(__name__)

[docs]class Graph(models.Model): """A collection of :class:`~Node` and :class:`~Edge`. This can be considered a score as well as a program as it has an entry point as a :class:`~Node` and can jump to any other :class:`~Node`, also allowing for recursive loops/cycles. Each node can be considered a little program on its own which can consist of multiple :class:`~ScriptCell` which can be coded in a variety of languages which can control the frontend and the audio (by e.g. speaking on the stream) or setting a background music. The story graph is a core concept and can be edited with a native editor. """
[docs] class StreamAssignmentPolicy(models.TextChoices): """ Each graph can handle different "connection" mechanisms when a listener accesses a graph. The implementation of each policy is defined in :class:`~story_graph.engine.Engine`. .. list-table:: :header-rows: 1 * - StreamAssignmentPolicy - Comment * - `ONE_GRAPH_ONE_STREAM` - All users share the same stream. When the first user visits a graph, a new stream will be set up. Any following user visiting the same story graph stream will be "redirected" to the same stream as long as there is still *any* user listening to the graph. All users still execute the story graph from the beginning. * - `ONE_USER_ONE_STREAM` - Upon connection, each user will obtain a new and exclusive stream and the graph will be executed upon the stream. * - `DEACTIVATE` - Non functional, for administrative work. """ ONE_GRAPH_ONE_STREAM = "one_graph_one_stream", _( "Each graph has only one stream" ) ONE_USER_ONE_STREAM = "one_user_one_stream", _("Each user gets its own stream") DEACTIVATE = "deactivate", _("No stream assignment")
[docs] class GraphDetailTemplate(models.TextChoices): DEFAULT = "default", _("Default template")
uuid = models.UUIDField( primary_key=True, editable=False, default=uuid.uuid4, unique=True, ) name = models.CharField( max_length=255, verbose_name=_("Name"), help_text=_("Name of the graph"), unique=True, ) display_name = models.CharField( max_length=512, verbose_name=_("Display name"), help_text=_("Will be used as a display name in the frontend"), ) slug_name = models.SlugField( verbose_name=_("Slug name"), max_length=256, unique=True, help_text=_("Will be used as a URL"), ) stream_assignment_policy = models.CharField( max_length=255, help_text=_("Manages the stream assignment for this graph"), choices=StreamAssignmentPolicy.choices, default=StreamAssignmentPolicy.ONE_USER_ONE_STREAM, ) public_visible = models.BooleanField( verbose_name=_("Public visible?"), help_text=_( "If the graph is not public it will not be listed in the frontend, yet it is still accessible via URL" ), default=True, null=False, blank=False, ) template_name = models.CharField( max_length=255, verbose_name=_("Frontend template"), help_text=_( "Allows to switch to a different template in the frontend with different connection flows or UI" ), choices=GraphDetailTemplate.choices, default=GraphDetailTemplate.DEFAULT, blank=False, null=False, ) start_text = models.TextField( verbose_name=_("Start text (markdown)"), help_text=_( "Text about the graph which will be displayed at the start of a stream - only if this is set" ), default="", blank=True, null=False, ) about_text = models.TextField( verbose_name=_("About text (markdown)"), help_text=_( "Text about the graph which can be accessed during a stream - only if this is set" ), default="", blank=True, null=False, ) end_text = models.TextField( verbose_name=_("End text (markdown)"), help_text=_("Text which will be displayed at the end of a stream"), default="", blank=True, null=False, )
[docs] async def aget_entry_node(self) -> "Node": """ See :func:`Graph.create_entry_node`. """ return await Node.objects.aget(is_entry_node=True, graph=self)
[docs] async def acreate_entry_node(self) -> "Node": """ Every graph needs a deterministic, unique entry node which is used to start the iteration over the graph. The creator of the graph is responsible for calling this method as we can not implicit call it because there are a multitude of ways of creating a Graph (async (asave), sync (save) or in a bulk where we do not have a handle at all). """ return await Node.objects.acreate( is_entry_node=True, graph=self, name="Start", )
class Meta: verbose_name = "Graph" verbose_name_plural = "Graphs" ordering = ["name"] def __str__(self) -> str: return
def update_graph_db_to_ws(graph_uuid: uuid.UUID): # sorry for this atrocity - there seems to be race conditions with signals # which makes updates out-dated, see # transaction.on_commit( lambda: async_to_sync(GenCasterChannel.send_graph_update)(graph_uuid) ) @receiver(signals.post_save, sender=Graph, dispatch_uid="update_graph_ws") def update_graph_ws(sender, instance: Graph, **kwargs) -> None: update_graph_db_to_ws(instance.uuid)
[docs]class Node(models.Model): """ A node. """ uuid = models.UUIDField( primary_key=True, editable=False, default=uuid.uuid4, unique=True, ) name = models.CharField( max_length=255, verbose_name=_("Name"), help_text=_("Name of the node"), default="", ) color = models.CharField( verbose_name=_("HEX color of the node in graph canvas"), max_length=16, default="#fff", ) position_x = models.FloatField( help_text=_("x-Position in graph canvas"), default=0.0, ) position_y = models.FloatField( help_text=_("y-Position in graph canvas"), default=0.0, ) graph = models.ForeignKey( Graph, related_name="nodes", on_delete=models.CASCADE, blank=False, null=False, ) is_entry_node = models.BooleanField( verbose_name="Is Entry node?", help_text=_( "Acts as a singular entrypoint for our graph." "Only one such node can exist per graph." ), default=False, ) is_blocking_node = models.BooleanField( verbose_name="Is blocking node?", help_text=_( "If we encounter this node during graph execution we will halt execution indefinitely on this node. This is useful if we have setup a state and do not want to change it anymore." ), default=False, ) async def aget_default_out_door(self) -> "NodeDoor": return await sync_to_async(self.get_default_out_door)() def get_default_out_door(self) -> "NodeDoor": default_out_door = NodeDoor.objects.filter( node=self, door_type=NodeDoor.DoorType.OUTPUT, is_default=True, ).first() if default_out_door is None: raise NodeDoorMissing(f"Default out door for node {self} is missing") return default_out_door async def aget_default_in_door(self) -> "NodeDoor": return await sync_to_async(self.get_default_in_door)() def get_default_in_door(self) -> "NodeDoor": default_in_door = NodeDoor.objects.filter( node=self, door_type=NodeDoor.DoorType.INPUT, is_default=True, ).first() if default_in_door is None: raise NodeDoorMissing(f"Default in door for node {self} is missing") return default_in_door
[docs] def save(self, *args, **kwargs): create_default_doors = self._state.adding super().save(*args, **kwargs) if create_default_doors: for t in NodeDoor.DoorType: NodeDoor( node=self, name="default", door_type=t, is_default=True, code="True", ).save()
class Meta: ordering = ["graph"] constraints = [ models.UniqueConstraint( fields=["graph"], condition=Q(is_entry_node=True), name="unique_entry_point", ) ] def __str__(self) -> str: return
def update_node_db_to_ws(node_uuid: uuid.UUID): # sorry for this atrocity - there seems to be race conditions with signals # which makes updates out-dated, see # transaction.on_commit( lambda: async_to_sync(GenCasterChannel.send_node_update)(node_uuid) ) @receiver(signals.post_delete, sender=Node, dispatch_uid="delete_node_ws") @receiver(signals.post_save, sender=Node, dispatch_uid="update_node_ws") def update_node_ws(sender, instance: Node, **kwargs) -> None: update_node_db_to_ws(instance.uuid) update_graph_db_to_ws(instance.graph.uuid)
[docs]class NodeDoorMissing(Exception): """Exception that can be thrown if a node door is missing. Normally each node should have a default in- and out :class:`~NodeDoor` via a signal, but as this is not forced via the database it is necessary to check for it. In case this check fails, this exception can be raised. """
[docs]class NodeDoor(models.Model): """A :class:`~Node` can be entered and exited via multiple paths, where each of these exits and entrances is called a *door*. A connection between nodes can only be made via their doors. There are two types of doors: .. list-table:: Door types :header-rows: 1 * - Kind - Description * - **INPUT** - Allows to enter a node. Currently each Node only has one entry point but for future development and a nicer database operations it is also represented. * - **OUTPUT** - Allows to exit a node. After all script cells of a node has been executed, the condition of each door will be evaluated (like in a switch case). Once a condition has been met, the door will be stepped through. This allows to have a visual representation of logic branches. It is only possible to connect an **OUTPUT** to an **INPUT** door via an :class:`~Edge`. """
[docs] class DoorType(models.TextChoices): INPUT = "input", _("Input") OUTPUT = "output", _("output")
uuid = models.UUIDField( primary_key=True, editable=False, default=uuid.uuid4, unique=True, ) door_type = models.CharField( max_length=20, blank=False, null=False, choices=DoorType.choices, default=DoorType.OUTPUT, ) node = models.ForeignKey( to=Node, on_delete=models.CASCADE, related_name="node_doors", ) name = models.CharField( max_length=512, blank=False, null=False, ) order = models.IntegerField( default=0, ) is_default = models.BooleanField( default=False, ) code = models.TextField( null=False, blank=False, default="", ) # only here for type-hints on reverse-relations out_edges: models.QuerySet["Edge"] in_edges: models.QuerySet["Edge"] class Meta(TypedModelMeta): ordering = [ "node", "is_default", "order", "name", ] verbose_name = _("Node door") verbose_name_plural = _("Node doors") constraints = [ models.UniqueConstraint( fields=["node", "door_type"], # see condition=Q(is_default=True), name="unique_default_per_type_and_node", ), ]
[docs] def save(self, *args, **kwargs): try: ast.parse(self.code) except SyntaxError as e: log.debug(f"Syntax error on node door {self}: {e}") raise e except Exception as e: log.error(f"Unexpected error on saving {self}: {e}") raise e # ignore args/kwargs b/c of "Cannot force both insert and updating in model saving" problem return super().save()
def __str__(self) -> str: return f"{self.node}: {self.door_type}_{}"
@receiver(signals.post_save, sender=NodeDoor, dispatch_uid="update_node_door_ws") def update_node_door_ws(sender, instance: NodeDoor, **kwargs) -> None: update_graph_db_to_ws(instance.node.graph.uuid) update_node_db_to_ws(instance.node.uuid) @receiver(signals.post_delete, sender=NodeDoor, dispatch_uid="delete_node_door_ws") def delete_node_door_ws(*args, **kwargs): return update_node_door_ws(*args, **kwargs)
[docs]class Edge(models.Model): """Connects two :class:`~Node` with each other by using their respective :class:`~NodeDoor`. .. important:: It is important to note that an edge flows from ``out_node_door`` to ``in_node_door`` as we follow the notion from the perspective of a :class:`story_graph.models.Node` rather than from the edge. .. graphviz:: digraph Connection { rank = same; subgraph cluster_node_a { rank = same; label = "NODE_A"; NODE_A [shape=Msquare, label="NODE_A\\n\\nscript_cell_1\\nscript_cell_2"]; subgraph cluster_in_nodes_a { label = "IN_NODES"; in_node_door_a [label="in_node_door"]; } subgraph cluster_out_nodes_a { label = "OUT_NODES"; out_node_door_a_1 [label="out_node_door 1"]; out_node_door_a_2 [label="out_node_door 2"]; } in_node_door_a -> NODE_A [label="DB\\nreference"]; {out_node_door_a_1, out_node_door_a_2} -> NODE_A; in_node_door_a -> NODE_A [style=dashed, color=red, fontcolor=red, label="Engine\\nProgression"]; NODE_A -> out_node_door_a_1 [style=dashed, color=red]; } edge_ [shape=Msquare, label="EDGE"]; edge_ -> out_node_door_a_1 [label="out_node_door"]; edge_ -> in_node_door_b [label="in_node_door"]; out_node_door_a_1 -> edge_ [style=dashed, color=red]; edge_ -> in_node_door_b [style=dashed, color=red]; subgraph cluster_node_b { rank = same; label = "NODE_B"; NODE_B [shape=Msquare]; subgraph cluster_in_nodes_b { label = "IN_NODES"; in_node_door_b [label="in_node_door"]; } subgraph cluster_out_nodes_b { label = "OUT_NODES"; out_node_door_b_1 [label="out_node_door 1"]; out_node_door_b_2 [label="out_node_door 2"]; } in_node_door_b -> NODE_B; {out_node_door_b_1, out_node_door_b_2} -> NODE_B; in_node_door_b -> NODE_B [style=dashed, color=red]; NODE_B -> out_node_door_b_1 [style=dashed, color=red]; } } """ uuid = models.UUIDField( primary_key=True, editable=False, default=uuid.uuid4, unique=True, ) in_node_door = models.ForeignKey( NodeDoor, related_name="in_edges", on_delete=models.CASCADE, # this should not be none but as we # added it during the lifecycle of # gencaster it is optional as otherwise # all prior data has to be deleted null=True, ) out_node_door = models.ForeignKey( NodeDoor, related_name="out_edges", on_delete=models.CASCADE, # see in_node_door null=True, )
[docs] def save(self, *args, **kwargs): """Checks if ``in_node_door`` and ``out_node_door`` have their respective types in order to avoid any *wrong* directions within our graph. """ if self.in_node_door: if self.in_node_door.door_type != NodeDoor.DoorType.INPUT: raise ValidationError(_("in_node_door needs to be an input door")) if self.out_node_door: if self.out_node_door.door_type != NodeDoor.DoorType.OUTPUT: raise ValidationError(_("out_node_door needs to be an output door")) super().save(*args, **kwargs)
class Meta(TypedModelMeta): constraints = [ models.UniqueConstraint( fields=["in_node_door", "out_node_door"], name="unique_edge", ), ] def __str__(self) -> str: return f"{self.in_node_door} -> {self.out_node_door}"
@receiver(signals.pre_delete, sender=Edge, dispatch_uid="delete_edge_ws") @receiver(signals.post_save, sender=Edge, dispatch_uid="update_edge_ws") def update_edge_ws(sender, instance: Edge, **kwargs) -> None: if instance.out_node_door: update_graph_db_to_ws(instance.out_node_door.node.graph.uuid) update_node_db_to_ws(instance.out_node_door.node.uuid) if instance.in_node_door: update_node_db_to_ws(instance.in_node_door.node.uuid)
[docs]class AudioCell(models.Model): """Stores information for playback of static audio files."""
[docs] class PlaybackChoices(models.TextChoices): """Different kinds of playback. .. list-table:: Playback types :header-rows: 1 * - Name - Description * - ``SYNC`` - Plays back an audio file and waits for the playback to finish before continuing the execution of the script cells. * - ``ASYNC`` - Plays back an audio file and immediately continues the execution of script cells. This is fitting for e.g. background music. """ SYNC_PLAYBACK = ["sync_playback", _("Sync playback")] ASYNC_PLAYBACK = ["async_playback", _("Async playback")]
uuid = models.UUIDField( primary_key=True, editable=False, default=uuid.uuid4, unique=True, ) playback = models.CharField( max_length=512, choices=PlaybackChoices.choices, default=PlaybackChoices.SYNC_PLAYBACK, null=False, blank=False, ) audio_file = models.ForeignKey( "stream.AudioFile", on_delete=models.CASCADE, related_name="audio_cells", null=False, blank=False, ) volume = models.FloatField( default=0.2, ) def __str__(self) -> str: return f"{self.audio_file} ({self.playback})"
[docs]class CellType(models.TextChoices): """A :class:`~story_graph.models.ScriptCell` can contain different types of code, each with unique functionality. Both, the database and :class:`~story_graph.engine.Engine`, implement some specific details according to these types. .. list-table:: Cell types :header-rows: 1 * - Name - Description - Database - Engine * - Markdown - Allows to write arbitrary text which will get rendered as an audio file via a text to speech service, see :class:`~stream.models.TextToSpeech` for conversion and :class:`~story_graph.markdown_parser.GencasterRenderer` for the extended Markdown syntax. - - :class:`~stream.models.TextToSpeech` - - :func:`~story_graph.engine.Engine.execute_markdown_code` - :class:`~story_graph.markdown_parser.GencasterRenderer` * - Python - Allows to execute python code via :func:`exec` which allows to trigger e.g. Dialogs in the frontend (see :class:`~stream.frontend_types.Dialog`) or calculate or fetch any kind of data and store its value as a :class:`~stream.models.StreamVariable`. - - - :func:`~story_graph.engine.Engine.execute_python_cell` * - SuperCollider - Executes *sclang* code on the associated server. This can be used to control the sonic content on the server. - - :class:`~stream.models.StreamInstruction` - - :func:`~story_graph.engine.Engine.execute_sc_code` - :ref:`OSC Server` * - Comment - Does not get executed, but allows to put comments into the graph. - - * - Audio - Allows to playback static audio files. The instruction will be translated into *sclang* code and will be executed as such on the associated stream. - - :class:`~story_graph.models.AudioCell` - :class:`~stream.models.AudioFile` - - :func:`~story_graph.engine.Engine.execute_audio_cell` """ MARKDOWN = ["markdown", _("Markdown")] PYTHON = ["python", _("Python")] SUPERCOLLIDER = ["supercollider", _("SuperCollider")] COMMENT = ["comment", _("Comment")] AUDIO = ["audio", _("Audio")]
[docs]class ScriptCell(models.Model): """Stores a script which can be executed with our :class:`~story_graph.engine.Engine` on a :class:`~stream.models.Stream`. """ uuid = models.UUIDField( primary_key=True, editable=False, default=uuid.uuid4, unique=True, ) node = models.ForeignKey( Node, related_name="script_cells", on_delete=models.CASCADE, ) cell_type = models.CharField( max_length=128, choices=CellType.choices, default=CellType.COMMENT, verbose_name=_("Cell type"), null=False, blank=False, ) cell_code = models.TextField( verbose_name=_("Cell code"), ) cell_order = models.IntegerField( default=0, ) audio_cell = models.OneToOneField( AudioCell, on_delete=models.CASCADE, related_name="script_cell", null=True, blank=True, ) class Meta: # ordering by uuid provides a deterministic order # in case cell_order is not unique ordering = ["node", "cell_order", "uuid"] constraints = [ models.CheckConstraint( check=Q(cell_type=CellType.AUDIO, audio_cell__isnull=False) | ~Q(cell_type=CellType.AUDIO), name="audio_type_needs_audio_cell_information", ) ] def __str__(self) -> str: return f"{self.node}-{self.cell_order} ({self.cell_type})"
@receiver(signals.post_delete, sender=ScriptCell, dispatch_uid="delete_script_cell") @receiver(signals.post_save, sender=ScriptCell, dispatch_uid="update_script_cell_ws") def update_script_cell_ws(sender, instance: ScriptCell, **kwargs) -> None: update_node_db_to_ws(instance.node.uuid)