# ------------------------------------------------------------------------------
# Name: features/base.py
# Purpose: Feature extractors base classes.
#
# Authors: Christopher Ariza
# Michael Scott Asato Cuthbert
#
# Copyright: Copyright © 2011-2023 Michael Scott Asato Cuthbert
# License: BSD, see license.txt
# ------------------------------------------------------------------------------
from __future__ import annotations
from collections import Counter
from collections.abc import Collection, Iterable, KeysView, Sequence
import os
import pathlib
import pickle
import typing as t
import unittest
from music21 import chord
from music21 import common
from music21.common.parallel import safeToParallize
from music21.common.types import StreamType
from music21 import converter
from music21 import corpus
from music21 import environment
from music21 import exceptions21
from music21 import note
from music21 import pitch
from music21 import stream
from music21 import text
from music21.metadata.bundles import MetadataBundle, MetadataEntry
if t.TYPE_CHECKING:
from music21.features import outputFormats
from music21.stream.base import SecondsMapEntry
environLocal = environment.Environment('features.base')
# A Stream or a path/reference that can be parsed into one.
type StreamOrPath = stream.Stream|MetadataEntry|str|pathlib.Path
# A single datum that a DataSet/DataInstance can ingest.
type DataSource = StreamOrPath|DataInstance
# A concrete class label or id value: a plain scalar.
type ClassValue = str|float|int
# A class value or id: either a fixed value or a pickleable function of the
# parsed Stream that produces one (evaluated lazily after parsing).
type ValueOrFunction = ClassValue|t.Callable[[stream.Stream], ClassValue]
# ------------------------------------------------------------------------------
class FeatureException(exceptions21.Music21Exception):
pass
[docs]
class Feature:
'''
An object representation of a feature, capable of presentation in a variety of formats,
and returned from FeatureExtractor objects.
Feature objects are simple. It is FeatureExtractors that store all metadata and processing
routines for creating Feature objects. Normally you wouldn't create one of these yourself.
* Changed in v11: the `.vector` starts out as an empty list rather than None.
>>> myFeature = features.Feature()
>>> myFeature.dimensions = 3
>>> myFeature.name = 'Random arguments'
>>> myFeature.isSequential = True
This is a continuous Feature, so we will set discrete to false.
>>> myFeature.discrete = False
The .vector is the most important part of the feature, and it starts out empty.
>>> myFeature.vector
[]
Calling .prepareVectors() gives it a list of zeros of the length of the dimensions.
>>> myFeature.prepareVectors()
>>> myFeature.vector
[0, 0, 0]
Now we can set the vector parts:
>>> myFeature.vector[0] = 4
>>> myFeature.vector[1] = 2
>>> myFeature.vector[2] = 1
It's okay just to assign a new list to .vector itself.
There is a "normalize()" method which normalizes the values
of a histogram to sum to 1.
>>> myFeature.normalize()
>>> myFeature.vector
[0.571..., 0.285..., 0.142...]
And that's it! FeatureExtractors are much more interesting.
'''
def __init__(self) -> None:
# these values will be filled by the extractor
self.dimensions: int = 1 # number of dimensions
# data storage; possibly use numpy array. Populated by prepareVectors().
self.vector: list[int|float] = []
# consider not storing these values, as they may not be necessary
self.name: str = '' # string name representation
self.description: str = '' # string description
self.isSequential: bool = True # True or False
self.discrete: bool = True # is discrete or continuous
def _getVectors(self) -> list[int|float]:
'''
Prepare a vector of appropriate size and return it.
'''
return [0] * self.dimensions
[docs]
def prepareVectors(self) -> None:
'''
Prepare the vector stored in this feature.
'''
self.vector = self._getVectors()
[docs]
def normalize(self) -> None:
'''
Normalize the vector so that the sum of its elements is 1.
'''
s = sum(self.vector)
try:
scalar = 1.0 / s # get floating point scalar for speed
except ZeroDivisionError:
raise FeatureException('cannot normalize zero vector')
temp = self._getVectors()
for i, v in enumerate(self.vector):
temp[i] = v * scalar
self.vector = temp
# ------------------------------------------------------------------------------
# ------------------------------------------------------------------------------
# ------------------------------------------------------------------------------
[docs]
class DataInstance:
'''
A data instance for analysis. This object prepares a Stream
(by stripping ties, etc.) and stores
multiple commonly-used stream representations once, providing rapid processing.
'''
# pylint: disable=redefined-builtin
# noinspection PyShadowingBuiltins
def __init__(self, streamOrPath: StreamOrPath|None = None,
id: ValueOrFunction|None = None) -> None:
self.stream: stream.Stream|None
if isinstance(streamOrPath, stream.Stream):
self.stream = streamOrPath
self.streamPath = None
else:
self.stream = None
self.streamPath = streamOrPath
# store an id for the source stream: file path url, corpus url
# or metadata title
self._id: ValueOrFunction
if id is not None:
self._id = id
elif ((s := self.stream) is not None
and hasattr(s, 'metadata')
and (md := s.metadata) is not None
and md.title is not None
):
self._id = md.title
elif self.stream is not None and hasattr(self.stream, 'sourcePath'):
self._id = self.stream.sourcePath
elif self.streamPath is not None:
if hasattr(self.streamPath, 'sourcePath'):
self._id = str(self.streamPath.sourcePath)
else:
self._id = str(self.streamPath)
else:
self._id = ''
# the attribute name in the data set for this label
self.classLabel: str = ''
# store the class value for this data instance
self._classValue: ValueOrFunction|None = None
self.partsCount = 0
self.forms: StreamForms|None = None
# store a list of voices, extracted from each part,
self.formsByVoice: list[StreamForms] = []
# if parts exist, store a forms for each
self.formsByPart: list[StreamForms] = []
self.featureExtractorClassesForParallelRunning: list[type[FeatureExtractor]] = []
if self.stream is not None:
self.setupPostStreamParse()
[docs]
def setupPostStreamParse(self) -> None:
'''
Set up the StreamForms objects and other things that
need to be done after a Stream is passed in but before
feature extracting is run.
Run automatically at instantiation if a Stream is passed in.
'''
# perform basic operations that are performed on all
# streams
if self.stream is None: # pragma: no cover
return
# store a dictionary of StreamForms
self.forms = StreamForms(self.stream)
# if parts exist, store a forms for each
self.formsByPart = []
if isinstance(self.stream, stream.Score):
self.partsCount = len(self.stream.parts)
for p in self.stream.parts:
# note that this will join ties and expand rests again
self.formsByPart.append(StreamForms(p))
else:
self.partsCount = 0
for v in self.stream[stream.Voice]:
self.formsByPart.append(StreamForms(v))
[docs]
def setClassLabel(self, classLabel: str, classValue: ValueOrFunction|None = None) -> None:
'''
Set the class label, as well as the class value if known.
The class label is the attribute name used to define the class of this data instance.
>>> #_DOCS_SHOW s = corpus.parse('bwv66.6')
>>> s = stream.Stream() #_DOCS_HIDE
>>> di = features.DataInstance(s)
>>> di.setClassLabel('Composer', 'Bach')
'''
self.classLabel = classLabel
self._classValue = classValue
[docs]
def getClassValue(self) -> ClassValue:
classValue = self._classValue
if classValue is None:
return ''
if callable(classValue):
if self.stream is None:
return ''
classValue = classValue(self.stream)
self._classValue = classValue
return classValue
[docs]
def getId(self) -> str:
idValue = self._id
if callable(idValue):
if self.stream is None:
return ''
idValue = idValue(self.stream)
self._id = idValue
# make sure there are no spaces; ids that are not strings are an error
if isinstance(idValue, str):
return idValue.replace(' ', '_')
raise AttributeError(str(idValue))
[docs]
def parseStream(self) -> None:
'''
If a path to a Stream has been passed in at creation,
then this will parse it (whether it's a corpus string,
a converter string (url or filepath), a pathlib.Path,
or a metadata.bundles.MetadataEntry).
'''
if self.stream is not None:
return
if isinstance(self.streamPath, str):
# could be corpus or file path
if os.path.exists(self.streamPath) or self.streamPath.startswith('http'):
s = converter.parse(self.streamPath)
else: # assume corpus
s = corpus.parse(self.streamPath)
elif isinstance(self.streamPath, pathlib.Path):
# could be corpus or file path
if self.streamPath.exists():
s = converter.parse(self.streamPath)
else: # assume corpus
s = corpus.parse(self.streamPath)
elif isinstance(self.streamPath, MetadataEntry):
s = self.streamPath.parse()
else:
raise ValueError(f'Invalid streamPath type: {type(self.streamPath)}')
self.stream = s
self.setupPostStreamParse()
[docs]
def __getitem__(self, key: str):
# the return is deliberately left untyped: a "form" can be a Stream,
# a Counter, a list, a float, etc., depending on the key, and any
# concrete annotation would be a lie that breaks the many call sites.
'''
Get a form of this Stream, using a cached version if available.
>>> di = features.DataInstance('bach/bwv66.6')
>>> len(di['flat'])
197
>>> len(di['flat.pitches'])
163
>>> len(di['flat.notes'])
163
>>> len(di['getElementsByClass(Measure)'])
40
>>> len(di['flat.getElementsByClass(TimeSignature)'])
4
'''
self.parseStream()
if key in ['parts']:
# return a list of Forms for each part
return self.formsByPart
elif key in ['voices']:
# return a list of Forms for voices
return self.formsByVoice
# try to create by calling the attribute
# will raise an attribute error if there is a problem
if self.forms is None: # pragma: no cover
raise FeatureException('cannot get a form from an unparsed DataInstance')
return self.forms[key]
# ------------------------------------------------------------------------------
class DataSetException(exceptions21.Music21Exception):
pass
[docs]
class DataSet:
'''
A set of features, as well as a collection of data to operate on.
Comprises multiple DataInstance objects, a FeatureSet, and an OutputFormat.
>>> ds = features.DataSet(classLabel='Composer')
>>> f = [features.jSymbolic.PitchClassDistributionFeature,
... features.jSymbolic.ChangesOfMeterFeature,
... features.jSymbolic.InitialTimeSignatureFeature]
>>> ds.addFeatureExtractors(f)
>>> ds.addData('bwv66.6', classValue='Bach')
>>> ds.addData('bach/bwv324.xml', classValue='Bach')
>>> ds.process()
>>> ds.getFeaturesAsList()[0]
['bwv66.6', 0.196..., 0.0736..., 0.006..., 0.098..., 0.0368..., 0.177..., 0.0,
0.085..., 0.134..., 0.018..., 0.171..., 0.0, 0, 4, 4, 'Bach']
>>> ds.getFeaturesAsList()[1]
['bach/bwv324.xml', 0.25, 0.0288..., 0.125, 0.0, 0.144..., 0.125, 0.0, 0.163..., 0.0, 0.134...,
0.0288..., 0.0, 0, 4, 4, 'Bach']
>>> ds = ds.getString()
By default, all exceptions are caught and printed if debug mode is on.
Set ds.failFast = True to not catch them.
Set ds.quiet = False to print them regardless of debug mode.
'''
def __init__(self, classLabel: str|None = None,
featureExtractors: Collection[type[FeatureExtractor]] = ()) -> None:
# assume a two dimensional array
self.dataInstances: list[DataInstance] = []
# order of feature extractors is the order used in the presentations
self._featureExtractors: list[type[FeatureExtractor]] = []
self._instantiatedFeatureExtractors: list[FeatureExtractor] = []
# the label of the class
self._classLabel = classLabel
# store a multidimensional storage of all features
self.features: list[list[Feature]] = []
self.failFast = False
self.quiet = True
self.runParallel = True
# set extractors
self.addFeatureExtractors(featureExtractors)
[docs]
def getClassLabel(self) -> str|None:
return self._classLabel
[docs]
def getAttributeLabels(self, includeClassLabel: bool = True,
includeId: bool = True) -> list[str]:
'''
Return a list of all attribute labels. Optionally add a class
label field and/or an id field.
>>> f = [features.jSymbolic.PitchClassDistributionFeature,
... features.jSymbolic.ChangesOfMeterFeature]
>>> ds = features.DataSet(classLabel='Composer', featureExtractors=f)
>>> ds.getAttributeLabels(includeId=False)
['Pitch_Class_Distribution_0',
'Pitch_Class_Distribution_1',
...
...
'Pitch_Class_Distribution_11',
'Changes_of_Meter',
'Composer']
'''
post = []
# place ids first
if includeId:
post.append('Identifier')
for fe in self._instantiatedFeatureExtractors:
post += fe.getAttributeLabels()
if self._classLabel is not None and includeClassLabel:
post.append(self._classLabel.replace(' ', '_'))
return post
[docs]
def getDiscreteLabels(self, includeClassLabel: bool = True,
includeId: bool = True) -> list[bool|None]:
'''
Return column labels for discrete status.
>>> f = [features.jSymbolic.PitchClassDistributionFeature,
... features.jSymbolic.ChangesOfMeterFeature]
>>> ds = features.DataSet(classLabel='Composer', featureExtractors=f)
>>> ds.getDiscreteLabels()
[None, False, False, False, False, False, False, False, False, False,
False, False, False, True, True]
'''
post: list[bool|None] = []
if includeId:
post.append(None) # just a spacer
for fe in self._instantiatedFeatureExtractors:
# need as many statements of discrete as there are dimensions
post += [fe.discrete] * fe.dimensions
# class label is assumed always discrete
if self._classLabel is not None and includeClassLabel:
post.append(True)
return post
[docs]
def getClassPositionLabels(self, includeId: bool = True) -> list[bool|None]:
'''
Return column labels for the presence of a class definition.
>>> f = [features.jSymbolic.PitchClassDistributionFeature,
... features.jSymbolic.ChangesOfMeterFeature]
>>> ds = features.DataSet(classLabel='Composer', featureExtractors=f)
>>> ds.getClassPositionLabels()
[None, False, False, False, False, False, False, False, False,
False, False, False, False, False, True]
'''
post: list[bool|None] = []
if includeId:
post.append(None) # just a spacer
for fe in self._instantiatedFeatureExtractors:
# need as many statements of discrete as there are dimensions
post += [False] * fe.dimensions
# class label is assumed always discrete
if self._classLabel is not None:
post.append(True)
return post
[docs]
def addMultipleData(
self,
dataList: Sequence[DataSource]|MetadataBundle,
classValues: Sequence[ValueOrFunction]|t.Callable[[stream.Stream], ClassValue],
ids: Sequence[ValueOrFunction|None]|t.Callable[[stream.Stream], str]|None = None,
) -> None:
'''
Add multiple data points at the same time.
Requires a sequence (including MetadataBundle) for dataList holding
types that can be passed to addData, an equally sized sequence of
classValues, and an equally sized sequence of ids (or None).
classValues can also be a pickleable function that will be called on
each instance after parsing, as can ids.
'''
if (not callable(classValues)
and len(dataList) != len(classValues)):
raise DataSetException(
'If classValues is not a function, it must have the same length as dataList')
if (ids is not None
and not callable(ids)
and len(dataList) != len(ids)):
raise DataSetException(
'If ids is not a function or None, it must have the same length as dataList')
classValueList: Sequence[ValueOrFunction|None]
if callable(classValues):
try:
pickle.dumps(classValues)
except pickle.PicklingError:
raise DataSetException('classValues if a function must be pickleable. '
+ 'Lambda and some other functions are not.')
classValueList = [classValues] * len(dataList)
else:
classValueList = classValues
idList: Sequence[ValueOrFunction|None]
if callable(ids):
try:
pickle.dumps(ids)
except pickle.PicklingError:
raise DataSetException('ids if a function must be pickleable. '
+ 'Lambda and some other functions are not.')
idList = [ids] * len(dataList)
elif ids is None:
idList = [None] * len(dataList)
else:
idList = ids
for i in range(len(dataList)):
self.addData(dataList[i], classValueList[i], idList[i])
# pylint: disable=redefined-builtin
# noinspection PyShadowingBuiltins
[docs]
def addData(self, dataOrStreamOrPath: DataSource,
classValue: ValueOrFunction|None = None,
id: ValueOrFunction|None = None) -> None:
'''
Add a Stream, DataInstance, MetadataEntry, or path (Posix or str)
to a corpus or local file to this data set.
The class value passed here is assumed to be the same as
the classLabel assigned at startup.
'''
if self._classLabel is None:
raise DataSetException(
'cannot add data unless a class label for this DataSet has been set.')
if isinstance(dataOrStreamOrPath, DataInstance):
di = dataOrStreamOrPath
else:
di = DataInstance(dataOrStreamOrPath, id=id)
di.setClassLabel(self._classLabel, classValue)
self.dataInstances.append(di)
[docs]
def process(self) -> None:
'''
Process all Data with all FeatureExtractors.
Processed data is stored internally as numerous Feature objects.
'''
if self.runParallel and safeToParallize():
self._processParallel()
else:
self._processNonParallel()
def _processParallel(self) -> None:
'''
Run a set of processes in parallel.
'''
for di in self.dataInstances:
di.featureExtractorClassesForParallelRunning = self._featureExtractors
shouldUpdate = not self.quiet
# print('about to run parallel')
outputData = common.runParallel([(di, self.failFast) for di in self.dataInstances],
_dataSetParallelSubprocess,
updateFunction=shouldUpdate,
updateMultiply=1,
unpackIterable=True
)
featureData, errors, classValues, ids = zip(*outputData)
errors = common.flattenList(errors)
for e in errors:
if self.quiet is True:
environLocal.printDebug(e)
else:
environLocal.warn(e)
self.features = list(featureData)
for i, di in enumerate(self.dataInstances):
if callable(di._classValue):
di._classValue = classValues[i]
if callable(di._id):
di._id = ids[i]
def _processNonParallel(self) -> None:
'''
The traditional way: run non-parallel.
'''
# clear features
self.features = []
for data in self.dataInstances:
row = []
for fe in self._instantiatedFeatureExtractors:
fe.setData(data)
# in some cases there might be problem; to not fail
try:
fReturned = fe.extract()
except Exception as e: # pylint: disable=broad-exception-caught
# for now take any error
fList = ['failed feature extractor:', fe, str(e)]
if self.quiet is True:
environLocal.printDebug(fList)
else:
environLocal.warn(fList)
if self.failFast is True:
raise e
# provide a blank feature extractor
fReturned = fe.getBlankFeature()
row.append(fReturned) # get feature and store
# rows will align with data the order of DataInstances
self.features.append(row)
[docs]
def getFeaturesAsList(self, includeClassLabel: bool = True, includeId: bool = True,
concatenateLists: bool = True) -> list:
'''
Get processed data as a list of lists, merging any sub-lists
in multidimensional features.
'''
post: list = []
for i, row in enumerate(self.features):
v: list = []
di = self.dataInstances[i]
if includeId:
v.append(di.getId())
for f in row:
if concatenateLists:
v += f.vector
else:
v.append(f.vector)
if includeClassLabel:
v.append(di.getClassValue())
post.append(v)
if not includeClassLabel and not includeId:
return post[0]
else:
return post
[docs]
def getUniqueClassValues(self) -> list[ClassValue]:
'''
Return a list of unique class values.
'''
post: list[ClassValue] = []
for di in self.dataInstances:
v = di.getClassValue()
if v not in post:
post.append(v)
return post
def _getOutputFormat(self, featureFormat: str) -> outputFormats.OutputFormat|None:
from music21.features import outputFormats
outputFormat: outputFormats.OutputFormat
if featureFormat.lower() in ['tab', 'orange', 'taborange']:
outputFormat = outputFormats.OutputTabOrange(dataSet=self)
elif featureFormat.lower() in ['csv', 'comma']:
outputFormat = outputFormats.OutputCSV(dataSet=self)
elif featureFormat.lower() in ['arff', 'attribute']:
outputFormat = outputFormats.OutputARFF(dataSet=self)
else:
return None
return outputFormat
def _getOutputFormatFromFilePath(self, fp: str|pathlib.Path) -> outputFormats.OutputFormat|None:
'''
Get an output format from a file path if possible, otherwise return None.
>>> ds = features.DataSet()
>>> ds._getOutputFormatFromFilePath('test.tab')
<music21.features.outputFormats.OutputTabOrange object at ...>
>>> ds._getOutputFormatFromFilePath('test.csv')
<music21.features.outputFormats.OutputCSV object at ...>
>>> ds._getOutputFormatFromFilePath('junk') is None
True
'''
# get format from fp if possible
fp = str(fp)
of = None
if '.' in fp:
of = self._getOutputFormat(fp.rsplit('.', maxsplit=1)[-1])
return of
[docs]
def getString(self, outputFmt: str = 'tab') -> str:
'''
Get a string representation of the data set in a specific format.
'''
# pass reference to self to output
outputFormat = self._getOutputFormat(outputFmt)
if outputFormat is None: # pragma: no cover
raise DataSetException(f'no output format could be defined from {outputFmt}')
return outputFormat.getString()
# pylint: disable=redefined-builtin
[docs]
def write(self, fp: str|pathlib.Path|None = None, format: str|None = None,
includeClassLabel: bool = True) -> str|pathlib.Path:
'''
Set the output format object.
'''
if format is None and fp is not None:
outputFormat = self._getOutputFormatFromFilePath(fp)
elif format is None:
raise ValueError('Specify a format or a file path')
else:
outputFormat = self._getOutputFormat(format)
if outputFormat is None:
raise DataSetException('no output format could be defined from file path '
f'{fp} or format {format}')
return outputFormat.write(fp=fp, includeClassLabel=includeClassLabel)
def _dataSetParallelSubprocess(
dataInstance: DataInstance, failFast: bool
) -> tuple[list[Feature], list[str], ClassValue, str]:
row: list[Feature] = []
errors: list[str] = []
# howBigWeCopied = len(pickle.dumps(dataInstance))
# print('Starting ', dataInstance, ' Size: ', howBigWeCopied)
for feClass in dataInstance.featureExtractorClassesForParallelRunning:
fe = feClass()
fe.setData(dataInstance)
# in some cases there might be problem; to not fail
try:
fReturned = fe.extract()
except Exception as e: # pylint: disable=broad-exception-caught
# for now take any error
errors.append('failed feature extractor:' + str(fe) + ': ' + str(e))
if failFast:
raise e
# provide a blank feature extractor
fReturned = fe.getBlankFeature()
row.append(fReturned) # get feature and store
# rows will align with data the order of DataInstances
return row, errors, dataInstance.getClassValue(), dataInstance.getId()
[docs]
def allFeaturesAsList(streamInput: DataSource) -> list:
# noinspection PyShadowingNames
'''
Returns a list containing ALL currently implemented feature extractors.
`streamInput` can be a Stream, DataInstance, or path to a corpus or local
file to this data set.
>>> s = converter.parse('tinynotation: 4/4 c4 d e2')
>>> f = features.allFeaturesAsList(s)
>>> f[2:5]
[[2], [2], [1.0]]
>>> len(f) > 85
True
'''
from music21.features import jSymbolic, native
ds = DataSet(classLabel='')
f = list(jSymbolic.featureExtractors) + list(native.featureExtractors)
ds.addFeatureExtractors(f)
ds.addData(streamInput)
ds.process()
allData = ds.getFeaturesAsList(includeClassLabel=False,
includeId=False,
concatenateLists=False)
return allData
# ------------------------------------------------------------------------------
[docs]
def vectorById(streamObj: stream.Stream|DataInstance, vectorId: str|Iterable[str],
library: str|Iterable[str] = ('jSymbolic', 'native')) -> list[int|float]|None:
'''
Utility function to get a vector from an extractor.
>>> s = stream.Stream()
>>> s.append(note.Note('A4'))
>>> features.vectorById(s, 'p20')
[1.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
'''
extractorClass = extractorById(vectorId, library)
if extractorClass is None:
return None # could raise exception
fe = extractorClass(streamObj) # call class with stream
return fe.extract().vector
[docs]
def getIndex(featureString: str, extractorType: str|None = None) -> tuple[int, str]|None:
'''
Returns the list index of the given feature extractor and the feature extractor
category (jsymbolic or native). If the feature extractor string is not in either
the jsymbolic or native feature extractors, returns None.
Optionally include the extractorType ('jsymbolic' or 'native') if known
and searching will be made more efficient.
>>> features.getIndex('Range')
(61, 'jsymbolic')
>>> features.getIndex('Ends With Landini Melodic Contour')
(18, 'native')
>>> features.getIndex('aBrandNewFeature!') is None
True
>>> features.getIndex('Fifths Pitch Histogram', 'jsymbolic')
(70, 'jsymbolic')
>>> features.getIndex('Tonal Certainty', 'native')
(1, 'native')
'''
from music21.features import jSymbolic, native
if extractorType is None or extractorType == 'jsymbolic':
indexCount = 0
for feature in jSymbolic.featureExtractors:
if feature().name == featureString:
return (indexCount, 'jsymbolic')
indexCount += 1
if extractorType is None or extractorType == 'native':
indexCount = 0
for feature in native.featureExtractors:
if feature().name == featureString:
return (indexCount, 'native')
indexCount += 1
return None
# ------------------------------------------------------------------------------
class Test(unittest.TestCase):
def testStreamFormsA(self):
from music21 import features
self.maxDiff = None
s = corpus.parse('corelli/opus3no1/1grave')
# s.chordify().show()
di = features.DataInstance(s)
self.assertEqual(len(di['flat']), 292)
self.assertEqual(len(di['flat.notes']), 238)
# di['chordify'].show('t')
self.assertEqual(len(di['chordify']), 27)
chordifiedChords = di['chordify.flat.getElementsByClass(Chord)']
self.assertEqual(len(chordifiedChords), 145)
histo = di['chordify.flat.getElementsByClass(Chord).setClassHistogram']
# print(histo)
self.assertEqual(histo,
{'3-11': 30, '2-4': 26, '1-1': 25, '2-3': 16, '3-9': 12, '2-2': 6,
'3-7': 6, '2-5': 6, '3-4': 5, '3-6': 5, '3-10': 4,
'3-8': 2, '3-2': 2})
self.assertEqual(di['chordify.flat.getElementsByClass(Chord).typesHistogram'],
{'isMinorTriad': 6, 'isAugmentedTriad': 0,
'isTriad': 34, 'isSeventh': 0, 'isDiminishedTriad': 4,
'isDiminishedSeventh': 0, 'isIncompleteMajorTriad': 26,
'isHalfDiminishedSeventh': 0, 'isMajorTriad': 24,
'isDominantSeventh': 0, 'isIncompleteMinorTriad': 16})
self.assertEqual(di['flat.notes.quarterLengthHistogram'],
{0.5: 116, 1.0: 39, 1.5: 27, 2.0: 31, 3.0: 2, 4.0: 3,
0.75: 4, 0.25: 16})
# can access parts by index
self.assertEqual(len(di['parts']), 3)
# stored in parts are StreamForms instances, caching their results
self.assertEqual(len(di['parts'][0]['flat.notes']), 71)
self.assertEqual(len(di['parts'][1]['flat.notes']), 66)
# getting a measure by part
self.assertEqual(len(di['parts'][0]['getElementsByClass(Measure)']), 19)
self.assertEqual(len(di['parts'][1]['getElementsByClass(Measure)']), 19)
self.assertEqual(di['parts'][0]['pitches.pitchClassHistogram'],
[9, 1, 11, 0, 9, 13, 0, 11, 0, 12, 5, 0])
# the sum of the two arrays is the pitch class histogram of the complete
# work
self.assertEqual(di['pitches.pitchClassHistogram'],
[47, 2, 25, 0, 25, 42, 0, 33, 0, 38, 22, 4])
def testStreamFormsB(self):
from music21 import features
s = stream.Stream()
for p in ['c4', 'c4', 'd-4', 'd#4', 'f#4', 'a#4', 'd#5', 'a5', 'a5']:
s.append(note.Note(p))
di = features.DataInstance(s)
self.assertEqual(di['pitches.midiIntervalHistogram'],
[2, 1, 1, 1, 1, 1, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0,
0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0])
def testStreamFormsC(self):
from pprint import pformat
from music21 import features
s = stream.Stream()
for p in ['c4', 'c4', 'd-4', 'd#4', 'f#4', 'a#4', 'd#5', 'a5']:
s.append(note.Note(p))
di = features.DataInstance(s)
self.assertEqual(pformat(di['flat.secondsMap']), '''[{'durationSeconds': 0.5,
'element': <music21.note.Note C>,
'endTimeSeconds': 0.5,
'offsetSeconds': 0.0,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note C>,
'endTimeSeconds': 1.0,
'offsetSeconds': 0.5,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note D->,
'endTimeSeconds': 1.5,
'offsetSeconds': 1.0,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note D#>,
'endTimeSeconds': 2.0,
'offsetSeconds': 1.5,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note F#>,
'endTimeSeconds': 2.5,
'offsetSeconds': 2.0,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note A#>,
'endTimeSeconds': 3.0,
'offsetSeconds': 2.5,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note D#>,
'endTimeSeconds': 3.5,
'offsetSeconds': 3.0,
'voiceIndex': None},
{'durationSeconds': 0.5,
'element': <music21.note.Note A>,
'endTimeSeconds': 4.0,
'offsetSeconds': 3.5,
'voiceIndex': None}]''', pformat(di['secondsMap']))
def testDataSetOutput(self):
from music21 import features
from music21.features import outputFormats
# test just a few features
featureExtractors = features.extractorsById(['ql1', 'ql2', 'ql4'], 'native')
# need to define what the class label will be
ds = features.DataSet(classLabel='Composer')
ds.runParallel = False
ds.addFeatureExtractors(featureExtractors)
# add works, defining the class value
ds.addData('bwv66.6', classValue='Bach')
ds.addData('corelli/opus3no1/1grave', classValue='Corelli')
ds.process()
# manually create an output format and get output
of = outputFormats.OutputCSV(ds)
post = of.getString(lineBreak='//')
self.assertEqual(
post,
'Identifier,Unique_Note_Quarter_Lengths,'
'Most_Common_Note_Quarter_Length,Range_of_Note_Quarter_Lengths,'
'Composer//bwv66.6,3,1.0,1.5,Bach//corelli/opus3no1/1grave,8,0.5,3.75,Corelli')
# without id
post = of.getString(lineBreak='//', includeId=False)
self.assertEqual(
post,
'Unique_Note_Quarter_Lengths,Most_Common_Note_Quarter_Length,'
'Range_of_Note_Quarter_Lengths,Composer//3,1.0,1.5,Bach//8,0.5,3.75,Corelli')
fp1 = ds.write(format='tab')
fp2 = ds.write(format='csv')
# Also test providing fp
fp3 = environLocal.getTempFile(suffix='.arff')
ds.write(fp=fp3, format='arff')
for fp in (fp1, fp2, fp3):
os.remove(fp)
def testFeatureFail(self):
from music21 import features
from music21 import base
featureExtractors = ['p10', 'p11', 'p12', 'p13']
featureExtractors = features.extractorsById(featureExtractors,
'jSymbolic')
ds = features.DataSet(classLabel='Composer')
ds.addFeatureExtractors(featureExtractors)
# create problematic streams
s = stream.Stream()
# s.append(None) # will create a wrapper -- NOT ANYMORE
s.append(base.ElementWrapper(None))
ds.addData(s, classValue='Monteverdi')
ds.addData(s, classValue='Handel')
# process with all feature extractors, store all features
ds.failFast = True
# Tests that some exception is raised, not necessarily that only one is
with self.assertRaises(features.FeatureException):
ds.process()
def testEmptyStreamCustomErrors(self):
from music21 import analysis
from music21 import features
from music21.features import jSymbolic, native
ds = DataSet(classLabel='')
f = list(jSymbolic.featureExtractors) + list(native.featureExtractors)
bareStream = stream.Stream()
bareScore = stream.Score()
singlePart = stream.Part()
singleMeasure = stream.Measure()
singlePart.append(singleMeasure)
bareScore.insert(singlePart)
ds.addData(bareStream)
ds.addData(bareScore)
ds.addFeatureExtractors(f)
for data in ds.dataInstances:
for fe in ds._instantiatedFeatureExtractors:
fe.setData(data)
try:
fe.extract()
# is every error wrapped?
except (features.FeatureException,
analysis.discrete.DiscreteAnalysisException):
pass
# --------------------------------------------------------------------------
# silent tests
# def testGetAllExtractorsMethods(self):
# '''
# ahh..this test takes a really long time.
# '''
# from music21 import stream, features, pitch
# s = corpus.parse('bwv66.6').measures(1, 5)
# self.assertEqual( len(features.alljSymbolicFeatures(s)), 70)
# self.assertEqual(len (features.allNativeFeatures(s)),21)
# self.assertEqual(str(features.alljSymbolicVectors(s)[1:5]),
# '[[2.6630434782608696], [2], [2], [0.391304347826087]]')
# self.assertEqual(str(features.allNativeVectors(s)[0:4]),
# '[[1], [1.0328322202181006], [2], [1.0]]')
def x_testComposerClassificationJSymbolic(self): # pragma: no cover
'''
Demonstrating writing out data files for feature extraction. Here,
features are used from the jSymbolic library.
'''
from music21 import features
featureExtractors = ['r31', 'r32', 'r33', 'r34', 'r35', 'p1', 'p2', 'p3',
'p4', 'p5', 'p6', 'p7', 'p8', 'p9', 'p10', 'p11', 'p12',
'p13', 'p14', 'p15', 'p16', 'p19', 'p20', 'p21']
# will return a list
featureExtractors = features.extractorsById(featureExtractors,
'jSymbolic')
# worksBach = corpus.getBachChorales()[100:143] # a middle range
worksMonteverdi = corpus.search('monteverdi').search('.xml')[:43]
worksBach = corpus.search('bach').search(numberOfParts=4)[:5]
# need to define what the class label will be
ds = features.DataSet(classLabel='Composer')
ds.addFeatureExtractors(featureExtractors)
# add works, defining the class value
# for w in worksBach:
# ds.addData(w, classValue='Bach')
for w in worksMonteverdi:
ds.addData(w, classValue='Monteverdi')
for w in worksBach:
ds.addData(w, classValue='Bach')
# process with all feature extractors, store all features
ds.process()
ds.write(format='tab')
ds.write(format='csv')
ds.write(format='arff')
def x_testRegionClassificationJSymbolicA(self): # pragma: no cover
'''
Demonstrating writing out data files for feature extraction. Here,
features are used from the jSymbolic library.
'''
from music21 import features
featureExtractors = features.extractorsById(['r31', 'r32', 'r33', 'r34', 'r35',
'p1', 'p2', 'p3', 'p4', 'p5', 'p6',
'p7', 'p8', 'p9', 'p10', 'p11', 'p12',
'p13', 'p14', 'p15', 'p16', 'p19',
'p20', 'p21'], 'jSymbolic')
oChina1 = corpus.parse('essenFolksong/han1')
oChina2 = corpus.parse('essenFolksong/han2')
oMitteleuropa1 = corpus.parse('essenFolksong/boehme10')
oMitteleuropa2 = corpus.parse('essenFolksong/boehme20')
ds = features.DataSet(classLabel='Region')
ds.addFeatureExtractors(featureExtractors)
# add works, defining the class value
for o, name in [(oChina1, 'han1'),
(oChina2, 'han2')]:
for w in o.scores:
songId = f'essenFolksong/{name}-{w.metadata.number}'
ds.addData(w, classValue='China', id=songId)
for o, name in [(oMitteleuropa1, 'boehme10'),
(oMitteleuropa2, 'boehme20')]:
for w in o.scores:
songId = f'essenFolksong/{name}-{w.metadata.number}'
ds.addData(w, classValue='Mitteleuropa', id=songId)
# process with all feature extractors, store all features
ds.process()
ds.getString(outputFmt='tab')
ds.getString(outputFmt='csv')
ds.getString(outputFmt='arff')
def x_testRegionClassificationJSymbolicB(self): # pragma: no cover
'''
Demonstrating writing out data files for feature extraction.
Here, features are used from the jSymbolic library.
'''
from music21 import features
# features common to both collections
featureExtractors = features.extractorsById(
['r31', 'r32', 'r33', 'r34', 'r35', 'p1', 'p2', 'p3', 'p4',
'p5', 'p6', 'p7', 'p8', 'p9', 'p10', 'p11', 'p12', 'p13',
'p14', 'p15', 'p16', 'p19', 'p20', 'p21'], 'jSymbolic')
# first bundle
ds = features.DataSet(classLabel='Region')
ds.addFeatureExtractors(featureExtractors)
oChina1 = corpus.parse('essenFolksong/han1')
oMitteleuropa1 = corpus.parse('essenFolksong/boehme10')
# add works, defining the class value
for o, name in [(oChina1, 'han1')]:
for w in o.scores:
songId = f'essenFolksong/{name}-{w.metadata.number}'
ds.addData(w, classValue='China', id=songId)
for o, name in [(oMitteleuropa1, 'boehme10')]:
for w in o.scores:
songId = f'essenFolksong/{name}-{w.metadata.number}'
ds.addData(w, classValue='Mitteleuropa', id=songId)
# process with all feature extractors, store all features
ds.process()
ds.write('/_scratch/chinaMitteleuropaSplit-a.tab')
ds.write('/_scratch/chinaMitteleuropaSplit-a.csv')
ds.write('/_scratch/chinaMitteleuropaSplit-a.arff')
# create second data set from alternate collections
ds = features.DataSet(classLabel='Region')
ds.addFeatureExtractors(featureExtractors)
oChina2 = corpus.parse('essenFolksong/han2')
oMitteleuropa2 = corpus.parse('essenFolksong/boehme20')
# add works, defining the class value
for o, name in [(oChina2, 'han2')]:
for w in o.scores:
songId = f'essenFolksong/{name}-{w.metadata.number}'
ds.addData(w, classValue='China', id=songId)
for o, name in [(oMitteleuropa2, 'boehme20')]:
for w in o.scores:
songId = f'essenFolksong/{name}-{w.metadata.number}'
ds.addData(w, classValue='Mitteleuropa', id=songId)
# process with all feature extractors, store all features
ds.process()
ds.write('/_scratch/chinaMitteleuropaSplit-b.tab')
ds.write('/_scratch/chinaMitteleuropaSplit-b.csv')
ds.write('/_scratch/chinaMitteleuropaSplit-b.arff')
# all these are written using orange-Py2 code; need better.
# def xtestOrangeBayesA(self): # pragma: no cover
# '''
# Using an already created test file with a BayesLearner.
# '''
# import orange # pylint: disable=import-error
# data = orange.ExampleTable(
# '~/music21Ext/mlDataSets/bachMonteverdi-a/bachMonteverdi-a.tab')
# classifier = orange.BayesLearner(data)
# for i in range(len(data)):
# c = classifier(data[i])
# print('original', data[i].getclass(), 'BayesLearner:', c)
# def xtestClassifiersA(self): # pragma: no cover
# '''
# Using an already created test file with a BayesLearner.
# '''
# import orange, orngTree # pylint: disable=import-error
# data1 = orange.ExampleTable(
# '~/music21Ext/mlDataSets/chinaMitteleuropa-b/chinaMitteleuropa-b1.tab')
#
# data2 = orange.ExampleTable(
# '~/music21Ext/mlDataSets/chinaMitteleuropa-b/chinaMitteleuropa-b2.tab')
#
# majority = orange.MajorityLearner
# bayes = orange.BayesLearner
# tree = orngTree.TreeLearner
# knn = orange.kNNLearner
#
# for classifierType in [majority, bayes, tree, knn]:
# print('')
# for classifierData, classifierStr, matchData, matchStr in [
# (data1, 'data1', data1, 'data1'),
# (data1, 'data1', data2, 'data2'),
# (data2, 'data2', data2, 'data2'),
# (data2, 'data2', data1, 'data1'),
# ]:
#
# # train with data1
# classifier = classifierType(classifierData)
# mismatch = 0
# for i in range(len(matchData)):
# c = classifier(matchData[i])
# if c != matchData[i].getclass():
# mismatch += 1
#
# print(
# f'{classifierStr} {classifierType}: misclassified '
# f'{mismatch}/{len(matchData)} of '
# f'{matchStr}'
# )
#
# # if classifierType == orngTree.TreeLearner:
# # orngTree.printTxt(classifier)
# def xtestClassifiersB(self): # pragma: no cover
# '''
# Using an already created test file with a BayesLearner.
# '''
# import orange, orngTree # pylint: disable=import-error
# data1 = orange.ExampleTable(
# '~/music21Ext/mlDataSets/chinaMitteleuropa-b/chinaMitteleuropa-b1.tab')
#
# data2 = orange.ExampleTable(
# '~/music21Ext/mlDataSets/chinaMitteleuropa-b/chinaMitteleuropa-b2.tab',
# use=data1.domain)
#
# data1.extend(data2)
# data = data1
#
# majority = orange.MajorityLearner
# bayes = orange.BayesLearner
# tree = orngTree.TreeLearner
# knn = orange.kNNLearner
#
# folds = 10
# for classifierType in [majority, bayes, tree, knn]:
# print('')
#
# cvIndices = orange.MakeRandomIndicesCV(data, folds)
# for fold in range(folds):
# train = data.select(cvIndices, fold, negate=1)
# test = data.select(cvIndices, fold)
#
# for classifierData, classifierStr, matchData, matchStr in [
# (train, 'train', test, 'test'),
# ]:
#
# # train with data1
# classifier = classifierType(classifierData)
# mismatch = 0
# for i in range(len(matchData)):
# c = classifier(matchData[i])
# if c != matchData[i].getclass():
# mismatch += 1
#
# print(
# f'{classifierStr} {classifierType}: misclassified '
# f'{mismatch}/{len(matchData)} of '
# f'{matchStr}'
# )
# def xtestOrangeClassifiers(self): # pragma: no cover
# '''
# This test shows how to compare four classifiers; replace the file path
# with a path to the .tab data file.
# '''
# import orange, orngTree # pylint: disable=import-error
# data = orange.ExampleTable(
# '~/music21Ext/mlDataSets/bachMonteverdi-a/bachMonteverdi-a.tab')
#
# # setting up the classifiers
# majority = orange.MajorityLearner(data)
# bayes = orange.BayesLearner(data)
# tree = orngTree.TreeLearner(data, sameMajorityPruning=1, mForPruning=2)
# knn = orange.kNNLearner(data, k=21)
#
# majority.name='Majority'
# bayes.name='Naive Bayes'
# tree.name='Tree'
# knn.name='kNN'
# classifiers = [majority, bayes, tree, knn]
#
# # print the head
# print('Possible classes:', data.domain.classVar.values)
# print('Original Class', end=' ')
# for l in classifiers:
# print(f'{l.name:<13}', end=' ')
# print()
#
# for example in data:
# print(f'({example.getclass():<10}) ', end=' ')
# for c in classifiers:
# p = c([example, orange.GetProbabilities])
# print(f'{p[0]:5.3f} ', end=' ')
# print('')
# def xtestOrangeClassifierTreeLearner(self): # pragma: no cover
# import orange, orngTree # pylint: disable=import-error
# data = orange.ExampleTable(
# '~/music21Ext/mlDataSets/bachMonteverdi-a/bachMonteverdi-a.tab')
#
# tree = orngTree.TreeLearner(data, sameMajorityPruning=1, mForPruning=2)
# # tree = orngTree.TreeLearner(data)
# for i in range(len(data)):
# p = tree(data[i], orange.GetProbabilities)
# print(
# f'{i + 1}: {p[1]:5.3f} (originally {data[i].getclass()})'
# )
#
# orngTree.printTxt(tree)
def testParallelRun(self):
from music21 import features
# test just a few features
featureExtractors = features.extractorsById(['ql1', 'ql2', 'ql4'], 'native')
# need to define what the class label will be
ds = features.DataSet(classLabel='Composer')
ds.addFeatureExtractors(featureExtractors)
# add works, defining the class value
ds.addData('bwv66.6', classValue='Bach')
ds.addData('corelli/opus3no1/1grave', classValue='Corelli')
ds.runParallel = True
ds.quiet = True
ds.process()
self.assertEqual(len(ds.features), 2)
self.assertEqual(len(ds.features[0]), 3)
fe00 = ds.features[0][0]
self.assertEqual(fe00.vector, [3])
# # pylint: disable=redefined-outer-name
# def x_fix_parallel_first_testMultipleSearches(self):
# from music21.features import outputFormats
# from music21 import features
#
# # Need explicit import for pickling within the testSingleCoreAll context
# from music21.features.base import _pickleFunctionNumPitches
# import textwrap
#
# self.maxDiff = None
#
# fewBach = corpus.search('bach/bwv6')
#
# self.assertEqual(len(fewBach), 13)
# ds = features.DataSet(classLabel='NumPitches')
# ds.addMultipleData(fewBach, classValues=_pickleFunctionNumPitches)
# featureExtractors = features.extractorsById(['ql1', 'ql4'], 'native')
# ds.addFeatureExtractors(featureExtractors)
# ds.runParallel = True
# ds.process()
# # manually create an output format and get output
# of = outputFormats.OutputCSV(ds)
# post = of.getString(lineBreak='\n')
# self.assertEqual(post.strip(), textwrap.dedent('''
# Identifier,Unique_Note_Quarter_Lengths,Range_of_Note_Quarter_Lengths,NumPitches
# bach/bwv6.6.mxl,4,1.75,164
# bach/bwv60.5.mxl,6,2.75,282
# bach/bwv62.6.mxl,5,1.75,182
# bach/bwv64.2.mxl,4,1.5,179
# bach/bwv64.4.mxl,5,2.5,249
# bach/bwv64.8.mxl,5,3.5,188
# bach/bwv65.2.mxl,4,3.0,148
# bach/bwv65.7.mxl,7,2.75,253
# bach/bwv66.6.mxl,3,1.5,165
# bach/bwv67.4.xml,3,1.5,173
# bach/bwv67.7.mxl,4,2.5,132
# bach/bwv69.6-a.mxl,4,1.5,170
# bach/bwv69.6.xml,8,4.25,623
# ''').strip())
def _pickleFunctionNumPitches(bachStream) -> int:
'''
A function for documentation testing of a pickleable function.
'''
return len(bachStream.pitches)
# ------------------------------------------------------------------------------
# define presented order in documentation
_DOC_ORDER = [DataSet, Feature, FeatureExtractor]
if __name__ == '__main__':
import music21
music21.mainTest(Test) # , runTest='testStreamFormsA')