1

I have a search algorithm that is used in conjunction with a machine learning algorithm. I separated the two because I wanted to be able to swap out search algorithms on the fly. However, the coupling is currently too tight. It looks something like:

class ML_Algorithm:

    def __init__(self,observers):
        import someMlLibrary
        self.observers = observers 
        self.classifier = someMlLibrary.model()
        self.searchWindow = SearchingWindow(self.__Classify)


    def Train(self, trainingData):
        preprocessedOutput = self.__Preprocess(trainingData)
        transformedOutput = self.__Transform(preprocessedOutput)
        self.classifier.fit(transformedOutput)


    def __call__(self, data):
        # The main entry point for all the processing
        self.searchWindow.AddData(data)
        for output in searchWindow:
           self.UpdateObservers(output)

    def __Classify(self, dataSubset):
        preprocessedOutput = self.__Preprocess(dataSubset)
        transformedOutput = self.__Transform(preprocessedOutput)
        return self.classifier.predict(transformedOutput)

    def __Prepocess(self,data):
        import someLibrary
        return someLibrary.doSomething(data)

    def __Transform(self,data):
        import someOtherLibrary
        return someOtherLibrary.doSomething(data)


    def UpdateObservers(self, output):
        for observer in self.observers:
            observer.OnOutput(output)

class Observer:

    def OnOutput(self, output):
        pass

class SearchingWindow:
    def __init__(self, fn, negativeResult=0):
        from collections import deque 
        self.fn = fn
        self.buffer = deque(maxlen=1000)
        self.negativeResult = negativeResult

    def AddData(self, data):
        self.buffer.extend(data)

    def __iter__(self):
         start = 0
         step = 1
         stop = 3*step
         while notDone:
            start, stop, step = self.GetNextStartStopStep(start, stop, step)
            for i in range(start,stop,step):
                dataSubset = self.buffer[start:stop]
                result1, result2, ..., resultN = self.fn(dataSubset)
                if result1 != self.negativeResult:
                    yield dataSubset

I realized the tight coupling when I wanted to access result1, ..., resultN to do some diagnostics through the ML_Algorithm object but couldn't.

Any help is much and truly appreciated.

4
+50

What you want is often referred to as the strategy pattern. You want to have a strategy for searching, that is easy to swap out. The easiest way to implement this is to accept the search strategy in the constructor of your main class.

class MyFavoriteAlgorithm:
    def __init__(self, observers, search_algorithm):
        .
        .
        self.search_algorithm = search_algorithm

As long as all the search strategies interface the same way with your code, this will allow you to swap out different search algorithms whenever you want. You can even leave a default search algorithm if there's one you want to use more often.

0

I would change the approach used, by doing the following:

  • Make the ML algorithm class simpler: put only train, classify, preprocess and transform methods (remove the observers and concrete reference to search window);
  • Put the observers stuff into the SearchingWindow;
  • Inject a ML classifier into the Searching window, it needs to implement the same interface (Train/Classify methods)
  • Place the entry point for processing into the SearchingWindow (or somewhere else, just please remove it from ML_Algorithm!)

Here's some code to illustrate my suggestion:

#Represents a classification method, therefore it only
#contains methods for training, preprocessing, feature transformation
#and classification
class ML_Algorithm:
    def __init__(self,observers):
        import someMlLibrary
        self.classifier = someMlLibrary.model()

    def Train(self, trainingData):
        preprocessedOutput = self.__Preprocess(trainingData)
        transformedOutput = self.__Transform(preprocessedOutput)
        self.classifier.fit(transformedOutput)

    def Classify(self, dataSubset): #public
        preprocessedOutput = self.__Preprocess(dataSubset)
        transformedOutput = self.__Transform(preprocessedOutput)
        return self.classifier.predict(transformedOutput)

    def __Prepocess(self,data):
        import someLibrary
        return someLibrary.doSomething(data)

    def __Transform(self,data):
        import someOtherLibrary
        return someOtherLibrary.doSomething(data)

#Application observers
class Observer:
    def OnOutput(self, output):
        pass

#Search window, which uses some classifier during the operation
class SearchingWindow:
    def __init__(self, clf, negativeResult=0):
        from collections import deque 
        self.observers = observers 
        self.classifier = clf  #now, the classifier is injected
        self.buffer = deque(maxlen=1000)
        self.negativeResult = negativeResult

    def AddData(self, data):
        self.buffer.extend(data)

    def UpdateObservers(self, output):
        for observer in self.observers:
            observer.OnOutput(output)


    def __call__(self, data):
        # The main entry point for all the processing
        self.searchWindow.AddData(data)
        self.classifier.Train(data)  #The main processing requires training
        for output in self:
           self.UpdateObservers(output) 

    def __iter__(self):
         start = 0
         step = 1
         stop = 3*step
         while notDone:
            start, stop, step = self.GetNextStartStopStep(start, stop, step)
            for i in range(start,stop,step):
                dataSubset = self.buffer[start:stop]
                result1, result2, ..., resultN = self.classifier.Classify(dataSubset)
                if result1 != self.negativeResult:
                    yield dataSubset

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.