Source code for VMO_Score.improviser

"""Module implementing methods for the off-line improvisation with an existing
Petri net"""

from random import randint
import utils
import copy
from collections import defaultdict
import snakes.plugins

snakes.plugins.load('tpn', 'snakes.nets', 'nets')


[docs]class Improviser(object): """Class representing the off-line improviser""" # time unit for the improvisation time_unit = 1.0 def __init__(self, pn, path_conf): """Initialization of the Improver class. Args: pn (PetriNet): Petri net to be walked path_conf (str): Path to the configuration file """ self._pn = copy.deepcopy(pn) configuration = utils.load_configuration(path_conf) self._actions = self._get_actions(configuration) self._pn.update_from_config(path_conf) self._add_environment_pn() self._current_time = 0.0 self.restart() @property def pn(self): """PetriNet: the Petri net for improvisation""" return self._pn @property def current_time(self): """float: the amount of time that has elapsed""" return self._current_time @staticmethod def _get_actions(conf): """Return the user actions from the configuration file Args: conf (dict): Dictionary representing the configuration file Returns: dict: Dictionary whose keys are dates and values are the actions performed at the corresponding date. """ actions = conf['actions'] new = defaultdict(list) if actions is not None: # if there are no actions for a in actions: new[a['time']].append(a) return new def _add_environment_pn(self): """Add the environment to the Petri Net Add a new place in the Petri net that denotes the interaction with the environment. This new place has arcs to transitions whose guards depends on values send by the environment. """ pn = self.pn.pn pn.add_place(nets.Place('env', [])) for t in pn.transition(): if t.guard != nets.Expression('True'): # It's different to True pn.add_input('env', t.name, nets.Tuple((nets.Variable('a'), nets.Variable('v')))) if t.max_time is not None: t1_name = t.name + "_default" pn.add_transition(nets.Transition(t1_name, min_time=t.max_time, max_time=t.max_time)) for (p, l) in t.input(): if p.name != 'env': pn.add_input(p.name, t1_name, nets.Value(nets.dot)) for (p, l) in t.output(): pn.add_output(p.name, t1_name, nets.Value(nets.dot))
[docs] def add_action(self, addr, value): """Add a user action to the Petri Net. Add a new token to the place that represents the actions sent by the environment, Args: addr (str): String representing the OSC address of the message value (str): String representing the OSC value of the message """ if self.pn.pn.has_place('env'): place_env = self.pn.pn.place('env') place_env.add([(addr, value)])
[docs] def delete_action(self, addr, value): """Remove a user action from the Petri Net Remove a token from the place that represents the actions sent by the environment. Args: addr (str): String representing the OSC address of the message value (str): String representing the OSC value of the message """ if self.pn.pn.has_place('env'): place_env = self.pn.pn.place('env') place_env.remove([(addr, value)])
[docs] def get_enabled_transitions(self): """Enabled transitions Return a list of enabled transitions with the current marking """ return [t.name for t in self.pn.pn.transition() if len(t.modes()) > 0 and t.enabled(t.modes()[0])]
[docs] def restart(self): """Restart to the initial marking of the Petri Net""" self.pn.pn.reset() self._current_time = 0.0
[docs] def next_time_unit(self): """Increment the global clock and the transition time by one time-unit""" step = self.pn.pn.step() self._current_time += self.time_unit for trans in self.pn.pn.transition(): if (step is None) or (step > 0.0): if trans.time is not None: trans.time += self.time_unit else: if trans.time == trans.max_time: trans.time = None
[docs] def fire_transition(self, t_name): """Fire a transition Take randomly a mode of an enabled transition and fire it Args: t_name (str): Name of the transition to be fired """ t = self.pn.pn.transition(t_name) modes = t.modes() n_modes = len(modes) if n_modes > 0: m = modes[randint(0, n_modes - 1)] print "Mode selected was ->", m t.fire(m)
[docs] def make_step(self): """Make a step in a Petri Net Take a logical step in the Petri Net. That is, an enabled transition is taken randomly and then it is fired. """ enabled_t = self.get_enabled_transitions() n_t = len(enabled_t) if n_t > 0: t = enabled_t[randint(0, n_t - 1)] print "Firing Transition ->", t self.fire_transition(t)
[docs] def is_final_marking(self): """Return if the current state of the Petri Net is a final state Returns: bool: The Petri Net is in a final state """ return self.pn.final_place in self.pn.pn.get_marking().keys()