This blog post is about machine learning techniques in live coding. I particularly focused on SuperCollider (SC3) and scikit-learn library for Python3. The main procedure was to send data over Open Sound Control (OSC) protocol, using pythonosc, and to analyze the data in Python3. The data analysis results are sending back to sclang in real-time for parameter control of UGens.

Description and order of execution

The main idea is to get input of environmental sounds using a microphone (also input from speakers works fine), and to analyze the input based on Chromagram class from SC3. The next step is to send the acoustical features to python via OSC. The python script chroma-server.py is responsible to receive the data and to make the data analysis. This is done by writing the raw data in the file data.txt. The data analysis is based on IncrementalPCA algorithm. When the analysis is finished chroma-server.py appends the results in data-pca.txt, and calls the client-parasite.py script which sends the data back to sclang.

The script client-parasite.py is in the folder ./lib. The python home directory looks as follows:

$ tree -L 2
.
├── chroma-server.py
├── data-pca.txt
├── data.txt
└── lib
    ├── __init.py__
    ├── __pycache__
    └── client_parasite.py

2 directories, 5 files

*NOTES*:
• init.py is an empty file, and __pycache__ is generated by python.
• Make sure to delete data.txt and data-pca.txt every time you restart the script.

*Order of execution*:
• Run the script chroma-server.py
• Run the code chunks in SC3

IMPORTANT NOTICE: For applying Incremental PCA we have to keep records of the full data set. As such, the 'data.txt' is getting bigger and bigger. Than makes the computations CPU intensive. Pay attention to Python CPU percentage!

sclang

(
SynthDef(\chroma,{
	var in, fft, chroma, array;
	in = SoundIn.ar(0); // get sound from mic
	fft = FFT(LocalBuf(2048), in);
	chroma = Chromagram.kr(fft);
	//chroma.poll(1);
	//in
	12 do: { |i|
		SendTrig.kr(Impulse.kr(2), i, chroma[i])
	};
}).add;
)

// send to python

(
b = NetAddr.new("127.0.0.1", 5005);    // create the NetAddr
o = OSCFunc({ arg msg, time;
	b.sendMsg("/chroma", msg[2].asString, msg[3].asString);
	//[msg[2], msg[3]].postln;
},'/tr', s.addr);
)

o.free;

x = Synth(\chroma) // start sending the synth data

x.free

thisProcess.openUDPPort(7007); // open port 7007 to receive from client-parasite.py

thisProcess.openPorts; // list all open ports

// start the sound to collect some data

Ndef(\brg).fadeTime_(4)

(
Ndef('brg', { |pc1=2 pc2=2 pc3=3 pc4=4|
	FreeVerb.ar(
		Ringz.ar(
			TGrains.ar(2, LFPulse.ar(pc1/pc3), Buffer.read(s, Platform.resourceDir ++ "/sounds/a11wlk01.wav"), Sweep.ar( x = LFPulse.ar(pc2/pc4)), x, Sweep.ar(x, pc3/pc4)
			),
			1390, 0.004
		),
		0.24, 0.11, 0.12
	)
}).play
)

// get the results of the analysis and control Ndef parameters

(
t = OSCFunc( { |msg, time, addr, recvPort|
	var pca_data;
	pca_data = clump(msg.asString.findRegexp("[0-9]+\.[0-9]+") collect: { |i| i[1] }, 4);    // 4 components
	pca_data.postln;
	Ndef(\brg).set(\pc1, pca_data[1][0].interpret, \pc2, pca_data[1][1].interpret, \pc3, pca_data[1][2].interpret, \pc4, pca_data[1][3].interpret);
}, '/components');
)

t.free;

Ndef(\brg).clear(2)

python 3

chroma-server.py

"""
Live coding session with machine learning.
The script receives Chromagram data from SC3 and writes the raw data from onsets analysis to 'data.txt', and then it writes a file called 'data-pca.txt' with the iPCA results
"""
import argparse
import math
import lib.client_parasite

import numpy as np
#np.set_printoptions(threshold=np.nan)
from sklearn.decomposition import IncrementalPCA
from scipy import stats

from pythonosc import dispatcher
from pythonosc import osc_server

def fw_pca(explained_variance, singular_values):
  """
  Write the results of the IPCA in data-pca.txt file
  """
  f = open('data-pca.txt', 'a')
  data_array = explained_variance + '\n' +  singular_values + '\n'
  f.write(data_array)
  f.close()

def pca(fname):
  """
  Perform IPCA on the features extracted from Chromagram in SC3
  """
  d = {}
  for x in range(0,12):
    d["val{0}".format(x)] = 0
  data_array = []
  with open(fname) as f:
    content = f.readlines()
  content = [x.strip() for x in content]
  #print('CONTENT: ', content)
  #print('CONTENT_SIZE: ', len(content))
  for elem in content:
    for i in range(0,12):
      if elem[0] == str(i) and elem[1] == ':':
        d['val'+str(i)] = float(elem.lstrip(str(i)+':').strip())
      elif elem[1] == '0':
        d['val10'] = float(elem.lstrip('10:').strip())
      elif elem[1] == '1':
        d['val11'] = float(elem.lstrip('11:').strip())
    data_array.append([d['val0'], d['val1'], d['val2'], d['val3'], d['val4'], d['val5'], d['val6'], d['val7'], d['val8'], d['val9'], d['val10'], d['val11']])
#print('DATA_ARRAY_SIZE: ', len(data_array))
  try:
    X = np.array(data_array)
    print('X = ', X)
    Y = stats.zscore(X)
    mypca = IncrementalPCA(n_components=4, batch_size=None) # configure batch size
    mypca.fit(Y)
    print(mypca.explained_variance_ratio_)
    print(mypca.singular_values_)
    lib.client_parasite.read_pca_data(str(mypca.explained_variance_ratio_),str(mypca.singular_values_))
  except:
    pass
  return fw_pca(str(mypca.explained_variance_ratio_), str(mypca.singular_values_))


def write_chroma(unused_addr, args, data):
  """
  Write incoming acoustical features of Chromagram to data.txt file
  """
  if args[0] is not None:
    chromadata = "{}: {}\n".format(args, data)
    f = open('data.txt', 'a')
    f.write(chromadata)  # python will convert \n to os.linesep
    f.close()
    with open('data.txt') as fn: # readlines in every entry
      for i, l in enumerate(fn):
        pass
      lines = i + 1
      if lines % 360 == 0: # n_features=12 * 5
        pca('data.txt')
      return print(lines)

if __name__ == "__main__":
  parser = argparse.ArgumentParser()
  parser.add_argument("--ip",
                      default="127.0.0.1", help="The ip to listen on")
  parser.add_argument("--port",
                      type=int, default=5005, help="The port to listen on")
  args = parser.parse_args()

  dispatcher = dispatcher.Dispatcher()
  dispatcher.map("/chroma", print)
  dispatcher.map("/chroma", write_chroma)

  server = osc_server.ThreadingOSCUDPServer(
    (args.ip, args.port), dispatcher)
  print("Serving on {}".format(server.server_address))
  server.serve_forever()

client-parasite.py

"""
The script reads the data-pca.txt and replies to SC3
"""
import argparse
import numpy as np

from pythonosc import osc_message_builder
from pythonosc import udp_client

def read_pca_data(first, second):
  parser = argparse.ArgumentParser()
  parser.add_argument("--ip", default="127.0.0.1",
                      help="The ip of the OSC server")
  parser.add_argument("--port", type=int, default=7007,
                       help="The port the OSC server is listening on")
  args = parser.parse_args()
  client = udp_client.SimpleUDPClient(args.ip, args.port)
  return client.send_message("/components", str([first, second]))

I would like to thank Ioannis Zannos for his suggestion to focus on OSC communication between sclang and python.