bAnalysisResults

Define and store SanPy spike-analysis results and their runtime schema.

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 in _CORE_ANALYSIS_RESULT_NAMES:
            v = defaultDict[k]
            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()

Methods:

__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 in _CORE_ANALYSIS_RESULT_NAMES:
        v = defaultDict[k]
        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]

Methods:

__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:

getDefaultDict()

Create an empty analysis-result definition.

Returns: A new dictionary containing every field required by an analysis-result definition.

Source code in sanpy/bAnalysisResults.py

def getDefaultDict() -> dict[str, Any]:
    """Create an empty analysis-result definition.

    Returns:
        A new dictionary containing every field required by an analysis-result
        definition.
    """
    defaultDict = {
        "category": AnalysisResultCategory.CUSTOM,
        "axis_label": "",
        "show_in_plot_menu": False,
        "is_categorical": False,
        "type": "",  # like: int, float, boolean, list
        "default": "",  # default value, can be 0, None, NaN, ...
        "units": "",  # real world units like point, mV, dvdt
        "depends on detection": "",  # organize documentation and refer to bDetect keys
        "error": "",  # if this analysis results can trigger an error
        "description": "",  # long description for documentation
    }
    return defaultDict.copy()

get_plot_result_definitions()

Return result definitions shown in X/Y statistic selectors.

Registry declaration order defines presentation order. Each definition is copied so interface code cannot mutate the authoritative registry.

Returns: Plottable definitions keyed by their internal analysis-result names.

Source code in sanpy/bAnalysisResults.py

def get_plot_result_definitions() -> dict[str, dict[str, Any]]:
    """Return result definitions shown in X/Y statistic selectors.

    Registry declaration order defines presentation order. Each definition is
    copied so interface code cannot mutate the authoritative registry.

    Returns:
        Plottable definitions keyed by their internal analysis-result names.
    """
    return {
        name: definition.copy()
        for name, definition in analysisResultDict.items()
        if definition["show_in_plot_menu"]
    }

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")

        from sanpy.sanpy_version import getSanPyProvenance
        sanpyProvenance = getSanPyProvenance()

        print(f"Generated {myDate} with SanPy version {sanpyProvenance.version}")
        print(str)

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

register_analysis_result(name, *, category=AnalysisResultCategory.CUSTOM, value_type='unknown', default=None, units='', axis_label='', show_in_plot_menu, is_categorical, description='', depends_on_detection='', error='')

Register a user-defined result in the authoritative runtime schema.

Identical repeated registration is accepted because SanPy may discover a user-analysis plugin more than once. Conflicting or invalid definitions fail immediately because they are programmer errors.

Args: name: Internal analysis-result name used as the dataframe column. category: Presentation category for the result. value_type: Runtime value type name. default: Default value for newly created results. units: Physical or logical units. axis_label: Human-readable plot-axis label. show_in_plot_menu: Whether X/Y statistic selectors show the result. is_categorical: Whether plots interpret values as discrete groups. description: Human-readable explanation of the result. depends_on_detection: Detection parameter dependencies, if any. error: Error condition documented for the result, if any.

Returns: True when the definition was added or already registered identically.

Raises: TypeError: If category is not an AnalysisResultCategory. ValueError: If name already has a different definition.

Source code in sanpy/bAnalysisResults.py

def register_analysis_result(
    name: str,
    *,
    category: AnalysisResultCategory = AnalysisResultCategory.CUSTOM,
    value_type: str = "unknown",
    default: Any = None,
    units: str = "",
    axis_label: str = "",
    show_in_plot_menu: bool,
    is_categorical: bool,
    description: str = "",
    depends_on_detection: str = "",
    error: str = "",
) -> bool:
    """Register a user-defined result in the authoritative runtime schema.

    Identical repeated registration is accepted because SanPy may discover a
    user-analysis plugin more than once. Conflicting or invalid definitions
    fail immediately because they are programmer errors.

    Args:
        name: Internal analysis-result name used as the dataframe column.
        category: Presentation category for the result.
        value_type: Runtime value type name.
        default: Default value for newly created results.
        units: Physical or logical units.
        axis_label: Human-readable plot-axis label.
        show_in_plot_menu: Whether X/Y statistic selectors show the result.
        is_categorical: Whether plots interpret values as discrete groups.
        description: Human-readable explanation of the result.
        depends_on_detection: Detection parameter dependencies, if any.
        error: Error condition documented for the result, if any.

    Returns:
        ``True`` when the definition was added or already registered
        identically.

    Raises:
        TypeError: If ``category`` is not an ``AnalysisResultCategory``.
        ValueError: If ``name`` already has a different definition.
    """
    if not isinstance(category, AnalysisResultCategory):
        raise TypeError(
            f'Analysis-result category for "{name}" must be AnalysisResultCategory'
        )
    definition = getDefaultDict()
    definition.update(
        {
            "category": category,
            "type": value_type,
            "default": default,
            "units": units,
            "axis_label": axis_label or name,
            "show_in_plot_menu": show_in_plot_menu,
            "is_categorical": is_categorical,
            "depends on detection": depends_on_detection,
            "error": error,
            "description": description,
        }
    )
    existing = analysisResultDict.get(name)
    if existing is None:
        analysisResultDict[name] = definition
        return True
    if existing == definition:
        return True
    raise ValueError(f'Conflicting analysis-result definition for "{name}"')