<commit_before>"""Managing clustering, fnotching and cut application here."""
from __future__ import division, print_function

import importlib
import logging
from pathlib import Path

import matplotlib
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from numpy.linalg import norm
from scipy.stats import circmean
from sklearn.preprocessing import minmax_scale, normalize, robust_scale, scale

from . import io, markings
from .dbscan import DBScanner, HDBScanner

importlib.reload(logging)
logpath = Path.home() / 'p4reduction.log'
logging.basicConfig(filename=str(logpath), filemode='w', level=logging.DEBUG,
                    format='%(asctime)s %(levelname)s: %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S')

matplotlib.style.use('bmh')


class ClusteringManager(object):

    """Control class to manage the clustering pipeline.

    Parameters
    ----------
    dbname : str or pathlib.Path, optional
        Path to the database used by DBManager. If not provided, DBManager
        will find the most recent one and use that.
    fnotch_distance : int
        Parameter to control the distance below which fans and blotches are
        determined to be 2 markings of the same thing, creating a FNOTCH
        chimera. The ratio between blotch and fan marks determines a fnotch
        value that can be used as a discriminator for the final object catalog.
    eps : int
        Parameter to control the exclusion distance for the DBSCAN
    output_dir : str or pathlib.Path
        Path to folder where to store output. Default: io.data_root / 'output'
    output_format : {'hdf', 'csv', 'both'}
        Format to save the output in. Default: 'hdf'
    cut : float
        Value to apply for fnotch cutting.
    min_samples_factor : float
        Value to multiply the number of unique classifications per image_id with
        to determine the `min_samples` value for DBSCAN to use. Default: 0.1
    do_dynamic_min_samples : bool, default is False
        Switch to decide if `min_samples` is being dynamically calculated.

    Attributes
    ----------
    confusion : list
        List of confusion data.
    fnotched_fans : list
        List of Fan objects after fnotches have been removed.
    fnotched_blotches : list
        List of Blotch objects after fnotches have been removed.
    fnotches : list
        List of Fnotch objects, as determined by `do_the_fnotch`.
    clustered_fans : list
        List of clustered Fan objects after clustering and averaging.
    clustered_blotches : list
        List of clustered Blotch objects after clustering and averaging.
    current_coords : list
        List of coordinate columns currently used for clustering
    current_markings : pandas.DataFrame
        Dataframe with the currently to-be-clustered marking data
    reduced_data : dictionary
        Stores reduced average cluster markings
    n_clustered_fans
    n_clustered_blotches
    output_dir_clustered : pathlib.Path
        Path to full clustering results, without removal of fnotched clusters.
    cut_dir : pathlib.Path
        Path to final fan and blotch clusters, after applying `cut`.
    """
    scalers = {'norm':normalize,
               'robust': robust_scale,
               'minmax': minmax_scale,
               'scale': scale}
    def __init__(self, dbname=None, fnotch_distance=10, eps=10,
                 output_dir=None, output_format='csv', cut=0.5,
                 min_samples_factor=0.15,
                 include_angle=True, id_=None, pm=None,
                 include_distance=False, include_radius=False,
                 do_dynamic_min_samples=False,
                 quiet=True, normalize=False,
                 scaler='robust',
                 use_DBSCAN=True,
                 hdbscan_min_samples=1):
        self.db = io.DBManager(dbname)
        self.dbname = self.db.dbname
        self.fnotch_distance = fnotch_distance
        self.output_dir = output_dir
        self.output_format = output_format
        self.eps = eps
        self.cut = cut
        self.include_angle = include_angle
        self.include_distance = include_distance
        self.include_radius = include_radius
        self.confusion = []
        self.min_samples_factor = min_samples_factor
        self.do_dynamic_min_samples = do_dynamic_min_samples
        self.quiet = quiet
        self.normalize = normalize
        self.scaler = scaler
        self.use_DBSCAN = use_DBSCAN
        self.hdbscan_min_samples = hdbscan_min_samples

        # to be defined at runtime:
        self.current_coords = None
        self.current_markings = None
        self.reduced_data = None
        self.fnotches = None
        self.fnotched_blotches = None
        self.fnotched_fans = None
        self.p4id = None
        self.newfans = None
        self.newblotches = None


        if pm is not None:
            self.pm = pm
        else:
            self.pm = io.PathManager(output_dir, id_=id_,
                                     suffix='.'+self.output_format)

    @property
    def n_clustered_fans(self):
        """int : Number of clustered fans."""
        return len(self.clustered_data['fan'])

    @property
    def n_clustered_blotches(self):
        """int : Number of clustered blotches."""
        return len(self.clustered_data['blotch'])

    def pre_processing(self, data, kind):
        """Preprocess before clustering.

        Depending on the flags used when constructing this manager,
        different columns end up being clustered on.

        Parameters
        ----------
        data : pd.DataFrame
            Dataframe with data to cluster on
        kind : str
            String indicating if to cluster on blotch or fan data
        """

        # add unit circle coordinates for angles
        angles = data['angle']
        data['xang'] = np.cos(np.deg2rad(angles))
        data['yang'] = np.sin(np.deg2rad(angles))

        # filter for the marking for `kind`
        marking_data = data[data.marking == kind]
        if len(marking_data) == 0:
            return None

        coords = ['x', 'y']

        if kind == 'fan':
            if self.include_distance:
                coords.append('distance')
            if self.include_angle:
                coords += ['xang', 'yang']
<commit_msg>use NotEnoughMarkingData Exception<commit_after>"""Managing clustering, fnotching and cut application here."""
from __future__ import division, print_function

import importlib
import logging
from pathlib import Path

import matplotlib
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from numpy.linalg import norm
from scipy.stats import circmean
from sklearn.preprocessing import minmax_scale, normalize, robust_scale, scale

from . import io, markings
from .dbscan import DBScanner, HDBScanner

importlib.reload(logging)
logpath = Path.home() / 'p4reduction.log'
logging.basicConfig(filename=str(logpath), filemode='w', level=logging.DEBUG,
                    format='%(asctime)s %(levelname)s: %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S')

matplotlib.style.use('bmh')


class NotEnoughMarkingData(Exception):
    def __init__(self):
        Exception.__init__(self, "Not enough data to cluster (< 3 items).")


class ClusteringManager(object):

    """Control class to manage the clustering pipeline.

    Parameters
    ----------
    dbname : str or pathlib.Path, optional
        Path to the database used by DBManager. If not provided, DBManager
        will find the most recent one and use that.
    fnotch_distance : int
        Parameter to control the distance below which fans and blotches are
        determined to be 2 markings of the same thing, creating a FNOTCH
        chimera. The ratio between blotch and fan marks determines a fnotch
        value that can be used as a discriminator for the final object catalog.
    eps : int
        Parameter to control the exclusion distance for the DBSCAN
    output_dir : str or pathlib.Path
        Path to folder where to store output. Default: io.data_root / 'output'
    output_format : {'hdf', 'csv', 'both'}
        Format to save the output in. Default: 'hdf'
    cut : float
        Value to apply for fnotch cutting.
    min_samples_factor : float
        Value to multiply the number of unique classifications per image_id with
        to determine the `min_samples` value for DBSCAN to use. Default: 0.1
    do_dynamic_min_samples : bool, default is False
        Switch to decide if `min_samples` is being dynamically calculated.

    Attributes
    ----------
    confusion : list
        List of confusion data.
    fnotched_fans : list
        List of Fan objects after fnotches have been removed.
    fnotched_blotches : list
        List of Blotch objects after fnotches have been removed.
    fnotches : list
        List of Fnotch objects, as determined by `do_the_fnotch`.
    clustered_fans : list
        List of clustered Fan objects after clustering and averaging.
    clustered_blotches : list
        List of clustered Blotch objects after clustering and averaging.
    current_coords : list
        List of coordinate columns currently used for clustering
    current_markings : pandas.DataFrame
        Dataframe with the currently to-be-clustered marking data
    reduced_data : dictionary
        Stores reduced average cluster markings
    n_clustered_fans
    n_clustered_blotches
    output_dir_clustered : pathlib.Path
        Path to full clustering results, without removal of fnotched clusters.
    cut_dir : pathlib.Path
        Path to final fan and blotch clusters, after applying `cut`.
    """
    scalers = {'norm':normalize,
               'robust': robust_scale,
               'minmax': minmax_scale,
               'scale': scale}
    def __init__(self, dbname=None, fnotch_distance=10, eps=10,
                 output_dir=None, output_format='csv', cut=0.5,
                 min_samples_factor=0.15,
                 include_angle=True, id_=None, pm=None,
                 include_distance=False, include_radius=False,
                 do_dynamic_min_samples=False,
                 quiet=True, normalize=False,
                 scaler='robust',
                 use_DBSCAN=True,
                 hdbscan_min_samples=1):
        self.db = io.DBManager(dbname)
        self.dbname = self.db.dbname
        self.fnotch_distance = fnotch_distance
        self.output_dir = output_dir
        self.output_format = output_format
        self.eps = eps
        self.cut = cut
        self.include_angle = include_angle
        self.include_distance = include_distance
        self.include_radius = include_radius
        self.confusion = []
        self.min_samples_factor = min_samples_factor
        self.do_dynamic_min_samples = do_dynamic_min_samples
        self.quiet = quiet
        self.normalize = normalize
        self.scaler = scaler
        self.use_DBSCAN = use_DBSCAN
        self.hdbscan_min_samples = hdbscan_min_samples

        # to be defined at runtime:
        self.current_coords = None
        self.current_markings = None
        self.reduced_data = None
        self.fnotches = None
        self.fnotched_blotches = None
        self.fnotched_fans = None
        self.p4id = None
        self.newfans = None
        self.newblotches = None


        if pm is not None:
            self.pm = pm
        else:
            self.pm = io.PathManager(output_dir, id_=id_,
                                     suffix='.'+self.output_format)

    @property
    def n_clustered_fans(self):
        """int : Number of clustered fans."""
        return len(self.clustered_data['fan'])

    @property
    def n_clustered_blotches(self):
        """int : Number of clustered blotches."""
        return len(self.clustered_data['blotch'])

    def pre_processing(self, data, kind):
        """Preprocess before clustering.

        Depending on the flags used when constructing this manager,
        different columns end up being clustered on.

        Parameters
        ----------
        data : pd.DataFrame
            Dataframe with data to cluster on
        kind : str
            String indicating if to cluster on blotch or fan data
        """

        # add unit circle coordinates for angles
        angles = data['angle']
        data['xang'] = np.cos(np.deg2rad(angles))
        data['yang'] = np.sin(np.deg2rad(angles))
        if len(marking_data) < 3:
            raise NotEnoughMarkingData
            # return None

        # filter for the marking for `kind`
        marking_data = data[data.marking == kind]
        if len(marking_data) == 0:
            return None

        coords = ['x', 'y']

        if kind == 'fan':
            if self.include_distance:
                coords.append('distance')
            if self.include_angle:
                coords += ['xang', 'yang']

{'fraction_non_alphanumeric': 0.0748775367389783, 'fraction_numerical': 0.0029546691548091126, 'mean_word_length': 2.3140943055913423, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '197207', 'n_tokens_mistral': 3676, 'n_tokens_neox': 3519, 'n_words': 1127}