IAA-CSIC / MUTIS

mutis/flares/bayblocks.py

""" Flare analysis """

import logging

import numpy as np

Unable to import 'numpy'

import scipy as sp

Unused scipy imported as sp

Unable to import 'scipy'

import pandas as pd

Unused pandas imported as pd

Unable to import 'pandas'

import matplotlib as mplt

Unused matplotlib imported as mplt

Unable to import 'matplotlib'

import matplotlib.pyplot as plt

Unable to import 'matplotlib.pyplot'

import sys

The import sys seems to be unused.

standard import "import sys" should be placed before "import numpy as np"

import os

standard import "import os" should be placed before "import numpy as np"

The import os seems to be unused.

import glob

standard import "import glob" should be placed before "import numpy as np"

The import glob seems to be unused.

import re

The import re seems to be unused.

standard import "import re" should be placed before "import numpy as np"

from datetime import datetime

standard import "from datetime import datetime" should be placed before "import numpy as np"

Unused datetime imported from datetime

from astropy.stats import bayesian_blocks

Unable to import 'astropy.stats'

import warnings 

standard import "import warnings" should be placed before "import numpy as np"

The import warnings seems to be unused.

Trailing whitespace

import mutis.flares.flare as flare

log = logging.getLogger(__name__)

class BayesianBlocks:
    """ Return a Bayesian Block representation of the signal
    

Trailing whitespace

    Attributes
    ----------
    edges: list
            list of edges defining the blocks.
    values: list
            list of values defining the height of the blocks.
    signal: mutis.Signal
            the mutis.Signal() instance that this BayesianBlocks() represents.
    inflare: list
            list of boolean representing whether a block is in flare or not.
    """

    def __init__(self, signal, p=0.1):
        edges = bayesian_blocks(signal.times, signal.values, signal.dvalues, fitness='measures', p0=p)

This line is too long as per the coding-style (102/100).

Trailing whitespace

        values = list()
        for i in range( len(edges)-1 ):

No space allowed before bracket

No space allowed after bracket

            msk = (edges[i] < signal.times) & (signal.times < edges[i+1])
            if np.sum(msk) > 0:
                value = np.average( signal.values[msk], weights=1/signal.dvalues[msk] )

No space allowed after bracket

No space allowed before bracket

                values.append( value )

No space allowed after bracket

No space allowed before bracket

            else:
                values.append( 0 )

No space allowed after bracket

No space allowed before bracket

Trailing whitespace

        self.edges = np.array(edges)
        self.values = np.array(values)
        self.signal = signal
        

Trailing whitespace

        self.inflare = None
    

Trailing whitespace

    def plot(self, style='color.strict', ax=None, **kwargs):

Argument name "ax" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        """ Plot the bayesian block representation
        

Trailing whitespace

        E.g:
        >>> signal = mutis.Signal(data['jyear'], data['Flux'], data['Flux_err'])
        >>> BayesianBlocks(signal).plot(color='k')

        """
        

Trailing whitespace

        ax = plt.gca() if ax is None else ax
        

Trailing whitespace

        #self.signal.plot()
        

Trailing whitespace

        x = self.edges

Variable name "x" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        y = self.values[np.r_[0:len(self.values),-1]]

Exactly one space required after comma

Variable name "y" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Trailing whitespace

        if self.inflare is None or style == 'none':
            

Trailing whitespace

            ax.step(x, y, 'k', where='post', **kwargs)
            

Trailing whitespace

        else:
            

Trailing whitespace

            if style == 'area':
                

Trailing whitespace

                ax.step(x, y, 'k', where='post')
                for i in range(len(self.inflare)):
                    if self.inflare[i]:
                        ax.axvspan(x[i], x[i+1], facecolor='r', edgecolor=None, alpha=0.2, **kwargs)
                        

Trailing whitespace

            elif style == 'color.simple':
                

Trailing whitespace

                for i in range(len(self.edges)-2):
                    ax.step(self.edges[[i,i+1]], self.values[[i,i+1]], 'r' if self.inflare[i] else 'k', where='post', **kwargs)

This line is too long as per the coding-style (127/100).

Exactly one space required after comma

                ax.step(self.edges[[-2,-1]], self.values[[-1,-1]], 'r' if self.inflare[-1] else 'k', where='post', **kwargs)

This line is too long as per the coding-style (124/100).

Exactly one space required after comma

Trailing whitespace

            elif style == 'color.loose':
                

Trailing whitespace

                for i in range(len(self.edges)-2):
                    ax.step(self.edges[[i,i+1]], self.values[[i,i]], 'r' if self.inflare[i] else 'k', where='post', **kwargs)

This line is too long as per the coding-style (125/100).

Exactly one space required after comma

                    ax.step(self.edges[[i+1,i+1]], self.values[[i,i+1]], 'r' if self.inflare[i] or self.inflare[i+1] else 'k', where='post', **kwargs)

Exactly one space required after comma

This line is too long as per the coding-style (150/100).

                ax.step(self.edges[[-2,-1]], self.values[[-1,-1]], 'r' if self.inflare[-1] else 'k', where='post', **kwargs)

Exactly one space required after comma

This line is too long as per the coding-style (124/100).

Trailing whitespace

            elif style == 'color.strict':
                

Trailing whitespace

                for i in range(len(self.edges)-2):
                    ax.step(self.edges[[i,i+1]], self.values[[i,i]], 'r' if self.inflare[i] else 'k', where='post', **kwargs)

Exactly one space required after comma

This line is too long as per the coding-style (125/100).

                    ax.step(self.edges[[i+1,i+1]], self.values[[i,i+1]], 'r' if self.inflare[i] and self.inflare[i+1] else 'k', where='post', **kwargs)

Exactly one space required after comma

This line is too long as per the coding-style (151/100).

                ax.step(self.edges[[-2,-1]], self.values[[-1,-1]], 'r' if self.inflare[-1] else 'k', where='post', **kwargs)

Exactly one space required after comma

This line is too long as per the coding-style (124/100).

Trailing whitespace

            else:
                raise Exception("Unknown style. Available styles are 'none', 'area', 'color.simple', 'color.strict', 'color.loose'.")

This line is too long as per the coding-style (133/100).

Trailing whitespace

Trailing whitespace

    def get_flares(self, thresh=1):
        """ Get a list of flares following the algorithm proposed in 

Trailing whitespace

[Meyer, Scargle, Blandford (2019)]
(https://iopscience.iop.org/article/10.3847/1538-4357/ab1651/pdf):
        

Trailing whitespace

```There is no generally accepted consensus on the best way to
determine which data points belong to a flaring state and which
characterize the quiescent level. Nalewajko (2013)suggested
the simple definition that a flare is a continuous time interval
associated with a flux peak in which the flux is larger than half
the peak flux value. This definition is intuitive, however, and it
is unclear how to treat overlapping flares and identify flux
peaks in an objective way. Here we use a simple two-step
procedure tailored to the block representation: (1)identify a
block that is higher than both the previous and subsequent
blocks as a peak, and (2)proceed downward from the peak in
both directions as long as the blocks are successively lower.```
        """
        di_max = 5
        

Trailing whitespace

        # get list of local maxima
        # bad beheaviour at 0 and -1:
        #imaxL = sp.signal.argrelextrema(self.values, np.greater)[0]
        imaxL = list()

Variable name "imaxL" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        for i in range(0,len(self.values)):

Exactly one space required after comma

            if i == 0 and thresh*self.values[0] > self.values[1]:
                imaxL.append(i)
            elif i == len(self.values)-1 and self.values[i-1] < thresh*self.values[i]:
                imaxL.append(i)
            elif self.values[i-1] < thresh*self.values[i] > self.values[i+1]:
                imaxL.append(i)
        imaxL = np.array(imaxL)

Variable name "imaxL" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Trailing whitespace

Trailing whitespace

        inflareL = np.full(self.values.shape, False)

Variable name "inflareL" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Trailing whitespace

        # get list of local maxima over a threshold (flare peaks)
        for imax in imaxL:
            inflare = True
            

Trailing whitespace

            if imax == 0:
                pass
            elif not (self.values[imax-1] < thresh*self.values[imax]):

There is an unnecessary parenthesis after not.

                inflare = False
            

Trailing whitespace

Trailing whitespace

            if imax == len(self.values)-1:
                pass
            elif  not (thresh*self.values[imax] > self.values[imax+1]):

There is an unnecessary parenthesis after not.

                inflare = False
                

Trailing whitespace

            inflareL[imax] = inflare
           

Trailing whitespace

        # extend the flare to adyacent blocks
        for imax in np.argwhere(inflareL):
            

Trailing whitespace

            if imax == 0:
                pass
            else:
                di = 1

Variable name "di" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

                stop = False
                while not stop and di < di_max:
                    if imax-di-1 < 0:
                        break
                    

Trailing whitespace

                    if self.values[imax-di-1] < thresh*self.values[imax-di]:
                        inflareL[imax-di] = True
                    else:
                        stop = True
                    

Trailing whitespace

Trailing whitespace

                    di = di + 1

Variable name "di" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Trailing whitespace

            if imax == len(self.values)-1:
                pass
            else:
                di = 1

Variable name "di" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

                stop = False
                while not stop and di < di_max:
                    if imax+di+1 > len(self.values)-1:
                        break
                    

Trailing whitespace

                    if thresh*self.values[imax+di] > self.values[imax+di+1]:
                        inflareL[imax+di] = True
                    else:
                        stop = True
                    

Trailing whitespace

Trailing whitespace

                    di = di + 1

Variable name "di" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

Trailing whitespace

Trailing whitespace

        self.inflare = np.array(inflareL)
        

Trailing whitespace

    def get_flare_list(self):
        """ Join all flares into a list of mutis.flares.Flare() """

        groups = np.split(np.arange(len(self.inflare)), 

Trailing whitespace

                          np.where((np.abs(np.diff(np.asarray(self.inflare, dtype=int))) == 1))[0]+1)

This line is too long as per the coding-style (101/100).

Trailing whitespace

        # (groups contains the groups of indices of self.inflare/self.values 

Trailing whitespace

        # that correspond to flares or no flares)
        

Trailing whitespace

        #print(self.inflare)
        #print([list(grp) for grp in groups])
        

Trailing whitespace

        flareL = list()

Variable name "flareL" doesn't conform to snake_case naming style ('([^\\W\\dA-Z][^\\WA-Z]2,|_[^\\WA-Z]*|__[^\\WA-Z\\d_][^\\WA-Z]+__)$' pattern)

        for i, grp in enumerate(groups):

The variable i seems to be unused.

            if np.all(self.inflare[grp]):
                #print(grp)
                tstart = self.edges[grp[0]]
                tstop = self.edges[grp[-1]+1] # flare ends when no flare begins
                

Trailing whitespace

                flareL.append(flare.Flare(tstart,tstop))

Exactly one space required after comma

Trailing whitespace

        #print(flareL)
        

Trailing whitespace

        return flareL

Final newline missing