sc3.base.stream module

Thread.sc & Stream.sc part 1

class TimeThread(func)

Bases: object

class State(value)

Bases: enum.Enum

An enumeration.

Init = 1

Running = 2

Suspended = 3

Paused = 4

Done = 5

property is_playing

property thread_player

property rand_seed

Random seed of the routine.

By default, routine’s random generators are inherited from the parent routine and only change when seeded.

property rand_state

Return the state of the random.Random internal generator.

exception RoutineException

Bases: Exception

exception StopStream

Bases: StopIteration

exception PausedStream

Bases: sc3.base.stream.StopStream

exception YieldAndReset(yield_value=None)

Bases: Exception

exception AlwaysYield(terminal_value=None)

Bases: Exception

class Stream

Bases: sc3.base.absobject.AbstractObject, abc.ABC

Lazy sequence of values.

Streams are iterator-generators compatible objects that implement a specific interface to interact with clocks and patterns.

abstract next(inval=None)

reset()

all(inval=None)

Same as list(stream) but with inval argument.

class UnopStream(selector, a)

Bases: sc3.base.stream.Stream

next(inval=None)

reset()

class BinopStream(selector, a, b)

Bases: sc3.base.stream.Stream

next(inval=None)

reset()

class NaropStream(selector, a, *args)

Bases: sc3.base.stream.Stream

next(inval=None)

reset()

class FunctionStream(next_func, reset_func=None, data=None)

Bases: sc3.base.stream.Stream

Create a stream from function evaluations.

next(inval=None)

Return the next value from the stream.

Its behaviour is the same as in Routine.next.

reset()

Reset the stream by executing the reset_func if provided.

If no reset_func was provided this method does nothing.

class Routine(func)

Bases: sc3.base.stream.TimeThread, sc3.base.stream.Stream

Routines are iterator-generator compatible objects that implement the interfaces needed to interact with clocks and patterns and can keep track of individual random states. They could be understood as timelines for sequencing code excecution.

Routines can be played, paused, resumed, stoped and reset. When played, they should yield time values, as int or float, that represent a precise timing offset between code excecution that is internally converted to OSC timetags for server commands. If running in a clock, the return value of the yield statement, as well as the initial value passed to the routine’s function, is a tuple (self, clock) that can be accessed to change states without keeping an external reference by closure.

Routines can be nested and played within others, the start time of the nested routines will be at the same logical time of the parent thus keeping synchronization and time relations even if played in a different clock than the parent’s. By default, nested routines inherit the clock and random number generator of the parent.

Note

To use different random states and seeds the functions provided by the builtins module must be used.

classmethod run(func, clock=None, quant=None)

Create and play a routine from a common function.

This method is a convenience constructor.

play(clock=None, quant=None)

Schedule the Routine in a clock to play it.

Parameters

• clock (Clock) – An optional clock to schedule the routine. By default, routines inherit the clock from the current time thread when played. The default clock of the main time thread is SystemClock.

• quant (Quant) – A Quant object or a any value that can be cast into one with Quant.as_quant constructor. This parameter only works for TempoClock and is ignored by other clocks.

next(inval=None)

Return the next value from the routine.

This method accepts an optional input value and act the same way as the generator’s send method. Routines behave like iterators and generators at the same time, once instantiated, this method evaluates its internal function and inval can be passed as the initial argument to that function. If the function is a generator function, an iterator will be created internally the first time this method is called and return the value of the first yield statement.

reset()

Reset the routine to its initial state.

pause()

Pause the routine and remove it from the clock if it is playing.

Raises

RoutineException – If this method is called from within the routine’s function itself.

resume(clock=None, quant=None)

Resume the routine, this method does nothing if wasn’t paused before.

Parameters

• clock (Clock) – An optionally different clock to re-schedule the routine. By default, the clock previously passed to the play function will be used.

• quant (Quant) – A Quant object or a any value that can be cast into one with Quant.as_quant constructor. This parameter only works for TempoClock and is ignored by other clocks.

stop()

Stop the routine and remove it from the clock if it is playing.

Raises

RoutineException – If this method is called from within the routine’s function itself.

class routine(func)

Bases: object

Decorator to convert generataor functions into Routines.

This decorator class is redundant with the Routine class, it returns an instance of that class, but its use is recommended when used as decorator.

static run(clock=None, quant=None)

Convenience decorator method equivalent to Routine.run(func, clock, quant).

class Condition(test=False)

Bases: object

Stop the execution of a routine playing on a clock until a condition is meet.

Clocks’ threads can’t be blocked with common locks. This class acts like a blocking condition by removing the routine from the clock’s scheduler and putting it in a waiting queue. See wait and hang methods for examples usage.

Parameters

test (bool | callable) – Initial test condition it can be a callable that return a boolean.

property test

Get the truth value of the test.

wait()

Return a generator that will remove the routine from the clock by returning a string and add it to a waiting queue when the condition is set to True and signaled. If the test condition is already True it rechedules the routine immediately.

cond = Condition()
@routine
def r():
    yield from cond.wait()  # Queued.
    print('resumed')
r.play()

cond.test = True  # Condition must be True to resume.
cond.signal()  # Signal the routine.

Raises

Exeption – If the generator is yield outside a routine.

signal()

Check the test and reschedule the routine if True.

unhang()

Unhang a previously hung routine.

class FlowVar

Bases: object

Defer the execution of a routine playing in a clock until a value is set.

This class is similar of awaitables for routines. Internaly it uses the Condition class defined in this library. See value property for example usage.

property value

Return a generator that adds the routine to a waiting queue and yield a string from the internal Condition. The routine will be recheduled to return its value when when it is set.

f_var = FlowVar()
@routine
def r():
    v = yield from f_var.value  # Queued.
    print('say', v)
r.play()

f_var.value = 'hello'  # Resume the routine.

Raises

Exeption – If the generator is yield outside a routine.

class ValueStream(value)

Bases: sc3.base.stream.Stream

Create a stream from any object.

next(inval=None)

class DictionaryStream(value)

Bases: sc3.base.stream.ValueStream

Create a stream from dict objects.

Dictionaries have a special meaning because they are the base clase for events and can be used as specifications deferring the event object creation.

next(indict=None)

stream(obj)

Convert any object into a Stream.

embed(obj, inval=None)

Convert any object into a Stream and return its embeddable form passing inval to the next calls.