bAnalysisResults

Classes

NumpyEncoder

Bases: JSONEncoder

Special json encoder for numpy types

Source code in sanpy/bAnalysisResults.py

class NumpyEncoder(json.JSONEncoder):
    """Special json encoder for numpy types"""

    def default(self, obj):
        if isinstance(obj, np.integer):
            return int(obj)
        elif isinstance(obj, np.floating):
            return float(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        return json.JSONEncoder.default(self, obj)

analysisResult

Source code in sanpy/bAnalysisResults.py

class analysisResult:
    def __init__(self, theDict=None):
        """Create an anlysis item (for one spike)

        Args:
            theDict: Pre-existing dict when we load form h5 file
        """
        # this is the raw definition of analysis results (See above)
        # pull from this to create key/value self.rDict
        # self._dDict = analysisResultDict
        defaultDict = analysisResultDict
        # this is simple key/value pairs as we will in detection
        self._rDict = {}
        for k, v in defaultDict.items():
            default = v["default"]
            self._rDict[k] = default

        if theDict is not None:
            for k, v in theDict.items():
                self[k] = v  # calls __setitem__()

    # this was interfering with converting to DataFrame ???
    """
    def __str__(self):
        printList = []
        for k,v in self._rDict.items():
            if isinstance(v, dict):
                for k2,v2 in v.items():
                    printList.append(f'  {k2} : {v2} {type(v2)}')
            else:
                printList.append(f'{k} : {v} {type(v)}')
        return '\n'.join(printList)
    """

    def print(self):
        printList = []
        for k, v in self._rDict.items():
            if isinstance(v, list):
                for item in v:
                    for k2, v2 in item.items():
                        printList.append(f"  {k2} : {v2} {type(v2)}")
            else:
                printList.append(f"{k} : {v} {type(v)}")
        return "\n".join(printList)

    def addNewKey(self, theKey, theDefault=None):
        """
        Add a new key to this spike.

        Returns: (bool) True if new key added, false if key already exists.
        """
        if theDefault is None:
            # theType = 'float'
            theDefault = float("nan")

        # check if key exists
        keyExists = theKey in self._rDict.keys()
        addedKey = False
        if keyExists:
            # key exists, don't modify
            # logger.warning(f'The key "{theKey}" already exists and has value "{self._rDict[theKey]}"')
            pass
        else:
            self._rDict[theKey] = theDefault
            addedKey = True

        #
        return addedKey

    def asDict(self):
        """
        Returns underlying dictionary
        """
        return self._rDict

    def __getitem__(self, key):
        # to mimic a dictionary
        ret = None
        try:
            # return self._dDict[key]['currentValue']
            ret = self._rDict[key]
        except KeyError as e:
            logger.error(f'Error getting key "{key}"')
            logger.error(f'possible keys are: {self._rDict.keys()}')
            raise
        #
        return ret

    def __setitem__(self, key, value):
        # to mimic a dictionary
        try:
            # self._dDict[key]['currentValue'] = value
            self._rDict[key] = value
        except KeyError as e:
            logger.error(f"{e}")

    def items(self):
        # to mimic a dictionary
        return self._rDict.items()

    def keys(self):
        # to mimic a dictionary
        return self._rDict.keys()

Functions

__init__(theDict=None)

Create an anlysis item (for one spike)

Args: theDict: Pre-existing dict when we load form h5 file

Source code in sanpy/bAnalysisResults.py

def __init__(self, theDict=None):
    """Create an anlysis item (for one spike)

    Args:
        theDict: Pre-existing dict when we load form h5 file
    """
    # this is the raw definition of analysis results (See above)
    # pull from this to create key/value self.rDict
    # self._dDict = analysisResultDict
    defaultDict = analysisResultDict
    # this is simple key/value pairs as we will in detection
    self._rDict = {}
    for k, v in defaultDict.items():
        default = v["default"]
        self._rDict[k] = default

    if theDict is not None:
        for k, v in theDict.items():
            self[k] = v  # calls __setitem__()

addNewKey(theKey, theDefault=None)

Add a new key to this spike.

Returns: (bool) True if new key added, false if key already exists.

Source code in sanpy/bAnalysisResults.py

def addNewKey(self, theKey, theDefault=None):
    """
    Add a new key to this spike.

    Returns: (bool) True if new key added, false if key already exists.
    """
    if theDefault is None:
        # theType = 'float'
        theDefault = float("nan")

    # check if key exists
    keyExists = theKey in self._rDict.keys()
    addedKey = False
    if keyExists:
        # key exists, don't modify
        # logger.warning(f'The key "{theKey}" already exists and has value "{self._rDict[theKey]}"')
        pass
    else:
        self._rDict[theKey] = theDefault
        addedKey = True

    #
    return addedKey

asDict()

Returns underlying dictionary

Source code in sanpy/bAnalysisResults.py

def asDict(self):
    """
    Returns underlying dictionary
    """
    return self._rDict

analysisResultList

Class encapsulating a list of analysis results.

Each row is an analysisResultDict for one spike.

These are keys in bAnalysis_ spike dict and columns in output reports

Source code in sanpy/bAnalysisResults.py

class analysisResultList:
    """Class encapsulating a list of analysis results.

    Each row is an analysisResultDict for one spike.

    These are keys in bAnalysis_ spike dict and columns in output reports
    """

    def __init__(self):
        # one copy for entire list

        # TODO: put xxx in a function getAnalysisResltDict()
        self._dDict = analysisResultDict

        # list of analysisResultDict
        self._myList = []

        self._iterIdx = -1

    def setFromListDict(self, listOfDict: List[dict]):
        """Set analysis results from a list of dict.

        Used when loading sanpy.bAnalysis from h5 file.

        When we create self (during spike detect) we have a list of class analysisResult.
        When we save/load we have a list of dict.

        This is assuming we re-create self every time we do spike detection
        """

        # do not do this, we are an analysisResultList as a list of analysisResult
        # self._myList = listOfDict

        self._myList = []
        for oneDict in listOfDict:
            oneAnalysisResult = analysisResult(theDict=oneDict)
            self._myList.append(oneAnalysisResult)

    def analysisDate(self):
        if len(self) > 0:
            return self._myList[0]["analysisDate"]
        else:
            return None

    def analysisTime(self):
        if len(self) > 0:
            return self._myList[0]["analysisTime"]
        else:
            return None

    def _old_save(self, saveBase):
        savePath = saveBase + "-analysis.json"

        analysisList = self.asList()

        # print(analysisList[0].print())
        print(self._myList[0])

        with open(savePath, "w") as f:
            json.dump(analysisList, f, cls=NumpyEncoder, indent=4)

    def _old_load(self, loadBase):
        loadPath = loadBase + "-analysis.json"

        if not os.path.isfile(loadPath):
            logger.error(f"Did not find file: {loadPath}")
            return

        with open(loadPath, "r") as f:
            self._myList = json.load(f)

    def appendDefault(self):
        """Append a spike to analysis.

        Used in bAnalysis spike detection.
        """
        oneResult = analysisResult()
        self._myList.append(oneResult)

    def appendAnalysis(self, analysisResultList):
        for analysisResult in analysisResultList:
            # analysisResult is for one spike
            self._myList.append(analysisResult)

    def addAnalysisResult(self, theKey, theDefault=None):
        # go through list and add to each [i] dict
        for spike in self:
            spike.addNewKey(theKey, theDefault=theDefault)

    def asList(self):
        """
        Return underlying list.
        """
        # return [spike.asDict() for spike in self._myList]
        return [x.asDict() for x in self._myList]

    def asDataFrame(self):
        """
        Note: underlying _myList is a list of analysisResult
        """
        return pd.DataFrame(self.asList())

    def __getitem__(self, key):
        """
        Allow [] indexing with self[int].
        """
        try:
            # return self._dDict[key]['currentValue']
            return self._myList[key]
        except IndexError as e:
            logger.error(f"{e}")
            # logger.error(f'possible keys are: {self._myList.keys()}')

    def __len__(self):
        """Allow len() with len(this)"""
        return len(self._myList)

    def __iter__(self):
        """Allow iteration with "for item in self"
        """
        _iterIdx = -1
        return self

    def __next__(self):
        """Allow iteration with "for item in self"
        """
        self._iterIdx += 1
        if self._iterIdx >= len(self._myList):
            self._iterIdx = -1  # reset to initial value
            raise StopIteration
        else:
            return self._myList[self._iterIdx]

Functions

__getitem__(key)

Allow [] indexing with self[int].

Source code in sanpy/bAnalysisResults.py

def __getitem__(self, key):
    """
    Allow [] indexing with self[int].
    """
    try:
        # return self._dDict[key]['currentValue']
        return self._myList[key]
    except IndexError as e:
        logger.error(f"{e}")

__iter__()

Allow iteration with "for item in self"

Source code in sanpy/bAnalysisResults.py

def __iter__(self):
    """Allow iteration with "for item in self"
    """
    _iterIdx = -1
    return self

__len__()

Allow len() with len(this)

Source code in sanpy/bAnalysisResults.py

def __len__(self):
    """Allow len() with len(this)"""
    return len(self._myList)

__next__()

Allow iteration with "for item in self"

Source code in sanpy/bAnalysisResults.py

def __next__(self):
    """Allow iteration with "for item in self"
    """
    self._iterIdx += 1
    if self._iterIdx >= len(self._myList):
        self._iterIdx = -1  # reset to initial value
        raise StopIteration
    else:
        return self._myList[self._iterIdx]

appendDefault()

Append a spike to analysis.

Used in bAnalysis spike detection.

Source code in sanpy/bAnalysisResults.py

def appendDefault(self):
    """Append a spike to analysis.

    Used in bAnalysis spike detection.
    """
    oneResult = analysisResult()
    self._myList.append(oneResult)

asDataFrame()

Note: underlying _myList is a list of analysisResult

Source code in sanpy/bAnalysisResults.py

def asDataFrame(self):
    """
    Note: underlying _myList is a list of analysisResult
    """
    return pd.DataFrame(self.asList())

asList()

Return underlying list.

Source code in sanpy/bAnalysisResults.py

def asList(self):
    """
    Return underlying list.
    """
    # return [spike.asDict() for spike in self._myList]
    return [x.asDict() for x in self._myList]

setFromListDict(listOfDict)

Set analysis results from a list of dict.

Used when loading sanpy.bAnalysis from h5 file.

When we create self (during spike detect) we have a list of class analysisResult. When we save/load we have a list of dict.

This is assuming we re-create self every time we do spike detection

Source code in sanpy/bAnalysisResults.py

def setFromListDict(self, listOfDict: List[dict]):
    """Set analysis results from a list of dict.

    Used when loading sanpy.bAnalysis from h5 file.

    When we create self (during spike detect) we have a list of class analysisResult.
    When we save/load we have a list of dict.

    This is assuming we re-create self every time we do spike detection
    """

    # do not do this, we are an analysisResultList as a list of analysisResult
    # self._myList = listOfDict

    self._myList = []
    for oneDict in listOfDict:
        oneAnalysisResult = analysisResult(theDict=oneDict)
        self._myList.append(oneAnalysisResult)

Functions

printDocs()

Print out human readable detection parameters and convert to markdown table.

Requires: pip install tabulate

See: bDetection.printDocs()

Source code in sanpy/bAnalysisResults.py

def printDocs():
    """Print out human readable detection parameters and convert to markdown table.

    Requires:
        pip install tabulate

    See: bDetection.printDocs()
    """
    import pandas as pd
    from datetime import datetime

    logger.info("Ensure there are no errors")

    dictList = []
    for k, v in analysisResultDict.items():
        # iterating on getDefaultDict() to ensure all code above has valid k/v pairs
        # lineStr = k + '\t'
        oneDict = {
            "Name": k,
        }
        for k2 in getDefaultDict().keys():
            # print(f'  {k}: {k2}: {v[k2]}')
            # lineStr += f'{v[k2]}' + '\t'
            oneDict[k2] = v[k2]
        #
        # print(lineStr)

        dictList.append(oneDict)

        # check that k is in headerDefaultDict
        for k3 in v:
            if not k3 in getDefaultDict().keys():
                logger.error(f'Found extra key "{k}" in "analysisResultDict"')

    #
    df = pd.DataFrame(dictList)

    if 1:
        # to markdown for mkdocs md file
        # str = df.to_markdown()
        str = df.to_html()
        myDate = datetime.today().strftime("%Y-%m-%d")
        print(f"Generated {myDate} with sanpy.analysisVersion {sanpy.analysisVersion}")
        print(str)

    if 0:
        path = "/Users/cudmore/Desktop/sanpy-analysis-results.csv"
        print("saving:", path)
        df.to_csv(path, index=False)