HintIndexer.py

# -*- coding: utf-8 -*-
"""
HintIndexer.py
==============

Portmanteau for:

Horizontal Interval Indexer
-          ---      -------

(in honor of Humdrum)

A Python object to interpret a NoteRestIndexed DataFrame. A new DataFrame is
created to find horizontal intervals in any given horizontal, or "melodic"
line via the VIS-Framework.

Author: Reiner Kramer	
Email: reiner@music.org
Updated: 11.18.2016

@TODO: 	Rather than saving dataframes as pickled files in the data folder
		of the library directory, files should be cached via memoization
		https://wiki.python.org/moin/PythonDecoratorLibrary#Memoize

"""

import sys, os, music21, pandas, requests, vis, pyext
from vis.models.indexed_piece import Importer

try:
	print("HintIndexer.py was loaded.")
except:
	print("Loading HintIndexer.py failed.")

class Get(pyext._class):
	"""
	HintIndexer Module
	==================

	Get
	---

	Processes a DataFrame generated by the NoteRestIndexer, parses it, and 
	creates a new DataFrame that shows all the horizontal intervals in a 
	particular part or stream. The main input type (inlet 1) is a 
	note-rest-indexed pickled DataFrame.

	"""
	_inlets = 6
	_outlets = 1

	def __init__(self,
		scores_paths=0,
		scores_imported=0,
		hints=0,
		hint_settings=0,
		events=5,
		mto_frozen_dir=0,
		df_paths=[],
		direction='beginning',
		slice_start=0,
		slice_end=5):

		self.scores_paths = scores_paths
		self.scores_imported = scores_imported
		self.hints = hints
		self.hint_settings = {
			'directed': True, 
			'simple or compound': 'compound', 
			'horiz_attach_later': False, 
			'quality': False, 
			'mp': True
		}
		self.events = events
		self.mto_frozen_dir = (os.path.dirname(os.path.realpath(__file__)) 
			+ '/data/music21streams/')
		self.df_paths = df_paths
		self.direction = direction
		self.slice_start = slice_start
		self.slice_end = slice_end

	def _anything_1(self,*symbolic_scores):
		"""
		Builds horizontal interval DataFrames.
		"""

		try:
			
			# Let user know that we are looking at horizontal intervals:
			msg = ("Horizontal interval music analysis:")
			print("\n" + msg + "\n" + len(msg) * "=")
			
			# Convert score paths from symbols to strings:
			self.scores_paths = [str(x) for x in symbolic_scores]

			# Import scores from score paths:
			self.scores_imported = [Importer(x) for x in self.scores_paths]

			# Collect Music21 streams for meta data purposes:
			self.scores_mto = [x._score for x in self.scores_imported]

			# Capture meta data from Music21 streams:
			self.meta = [(x.metadata.composer + "_" + 
				x.metadata.title).replace(" ", "-") 
				for x in self.scores_mto]

			# Freeze Music21 streams for later consumption:
			self.scores_mto_frozen = [music21.converter.freeze(
				self.scores_mto[i], fmt='pickle', fp=(self.mto_frozen_dir + 
				self.meta[i] + '.pgz'))
				for i in range(len(self.scores_mto))]
			
			# Build dataframes holding horizontal intervals:
			self.hints = [x.get_data('horizontal_interval', 
				self.hint_settings) 
				for x in self.scores_imported]

			# Save NoteRestIndexed DataFrames:
			for i in range(len(self.hints)):
				
				# Build the path names, and save into a list.
				self.df_paths.append(
					os.path.dirname(os.path.realpath(__file__)) + 
					'/data/frames/hints/' + self.meta[i] + '.pkl')

				# Save the dataframes as pickle(d) files.
				self.hints[i].to_pickle(self.df_paths[i])

			# Renaming the columns to a more user friendly format:
			for x, y in zip(self.df_paths,self.hints):
				self._generate_name(x)
				
				y.columns.set_levels(['Part'], level=0, inplace=True)
				y.columns.set_names(['Score','Events'], inplace=True)
				print(y.head(self.events).to_csv(
					sep='\t',
					na_rep='^'))

		except Exception as e:

			print(e) 

	def _anything_2(self,events):
		"""
		Determines how many events are to be shown.
		"""
		if(self.hints == 0):
			self._msg_missing_scores()
		else:
			self.events = events
			# The beginning or the end of the DataFrame
			self._heads_or_tails()

	def _anything_3(self,direction):
		"""
		Determines, whether the events are shown from the beginning or 
		the end.
		"""
		if(self.hints == 0):
			self._msg_missing_scores()
		else:
			self.direction = str(direction)
			self._heads_or_tails()

	def _anything_4(self,slice_start,slice_end):
		"""
		Picks a slice from a given DataFrame.
		"""
		if(self.hints == 0):
			self._msg_missing_scores()
		else:
			for x, y in zip(self.df_paths,self.hints):
				self._generate_name(x)
				y.columns.set_levels(['Part'], level=0, inplace=True)
				y.columns.set_names(['Score','Events'], inplace=True)
				print(y.iloc[slice_start:slice_end].to_csv(
					sep='\t',
					na_rep='^'))

	def _anything_5(self,*hint_settings):
		"""
		Settings as adopted from the VIS-framework.
		"""
		self.hint_settings = {

			'simple or compound': str(hint_settings[0]),
			'quality': eval(str(hint_settings[1])),
			'directed': eval(str(hint_settings[2])),
			'mp': eval(str(hint_settings[3])),
			'horiz_attach_later': eval(str(hint_settings[4]))		

		}

		# Build dataframes holding horizontal intervals:
		self.hints = [x.get_data('horizontal_interval', 
			self.hint_settings) 
			for x in self.scores_imported]

		self._heads_or_tails()

	def bang_1(self):
		"""
		Force pass DataFrame paths to next items, e.g.: filters.
		"""
		if(self.hints == 0):
			self._outlet(1, self._msg_missing_scores())
		
		else:
			# self._outlet(1, "DataFrames exist.")
			print("The horizontal intervals DataFrames were passed on.")
			self._outlet(1, [str(x) for x in self.df_paths])

	def _generate_name(self,path):
		"""
		Private method to generate a human readable name of a composition from
		it path.
		"""
		file_name = os.path.split(path)
		file_extr = os.path.splitext(file_name[1])
		comp_name = str(file_extr[0]).replace("-"," ").replace("_",": ")

		print("\n" + comp_name)
		print(len(comp_name) * "-")

	def _heads_or_tails(self):
		"""
		Helper method to determine whether to count from the beginning
		or from the end of the DataFrame.
		"""
		for x, y in zip(self.df_paths,self.hints):
			
			self._generate_name(x)

			y.columns.set_levels(['Part'], level=0, inplace=True)
			y.columns.set_names(['Score','Events'], inplace=True)

			if(self.direction == 'end'):
				display = y.tail(self.events).to_csv(sep='\t', na_rep='^')
			else:
				display = y.head(self.events).to_csv(sep='\t', na_rep='^')
			
			print(display)

	def _msg_missing_scores(self):
		"""
		Method to indicate that no DataFrames have been loaded.
		"""
		return "Please load (a) note-rest-indexed DataFrame(s) first."