Week 6: Software bug

helpers.py

from __future__ import annotations
import string

from typing import (
    Callable,
    Collection,
)

import numpy as np
import pandas as pd
import nltk

from sklearn import (
    base,
    pipeline,
    preprocessing,
    linear_model,
)
from sklearn.pipeline import (
    Pipeline,
)
from sklearn.decomposition import (
    PCA,
)
from sklearn.preprocessing import (
    StandardScaler,
)
from sklearn.feature_extraction.text import (
    CountVectorizer,
    TfidfVectorizer,
)

import data_analysis as da


STOPWORDS = nltk.corpus.stopwords.words('english')


def load_data(
    filename: str = 'https://gist.githubusercontent.com/ShivnarenSrinivasan/58412c09d94153f74f20bf1d6973f606/raw/3a8c5a97ae81648fef60b73222dea9ecca3a23f5/issues_data.csv',
) -> pd.DataFrame:
    return pd.read_csv(filename)


def tp_title(text: str) -> str:
    def is_valid_word(word: str) -> bool:
        valid_patterns = {
            '->',
            '=>',
            '?:',
        }
        return word in valid_patterns or (
            word.isascii()
            # * plenty of useful 3 letter phrases like JIT, CSS
            # and len(word) > 3
        )

    def clean(word: str) -> str:
        strip_chars = string.punctuation + ' '
        return word.strip(strip_chars)

    stop_words_eng = set(STOPWORDS)

    return ' '.join(
        clean(word)
        for word in text.lower().split()
        if is_valid_word(word) and word not in stop_words_eng
    )


def tp_body(txt: str) -> str:
    def not_metadata(txt: str) -> bool:
        return not (txt.startswith('_') and txt.endswith('_'))

    def clean(txt: str) -> str:
        def not_url(txt: str) -> bool:
            urls = {"www", "http", ".net", ".com"}
            return not any(url in txt for url in urls)

        return ' '.join(word for word in tp_title(txt).split() if not_url(word))

    lines = txt.replace('\r', '').split('\n')

    return ' '.join(clean(line) for line in lines if not_metadata(line)).strip()


def subset_rows(df: pd.DataFrame) -> pd.DataFrame:
    return ~((df['Issue Title'].str.isspace()) & (df['Issue Body'].str.isspace()))


def get_binned_labels(ser: pd.Series, n: int = 3) -> Collection[str]:
    return ser.value_counts().index[n:]


def label_counts(data: da.Data) -> pd.Series:
    return pd.Series(data.y).value_counts()


def check_null(dataset: da.Dataset):
    train = dataset.data.train
    return pd.Series(train.y).isnull().sum(), train.X.isnull().sum()


# Data Utilities
def map_labels(labels: Collection[str], enc_label: str) -> Callable[[str], str]:
    mapping = {label: enc_label for label in labels}

    def map_label(label: str) -> str:
        return mapping.get(label, label)

    return map_label


def make_X(
    df: pd.DataFrame,
    cols: tuple[str, str] = ('Issue Title', 'Issue Body'),
    label: str = 'Issues',
) -> pd.DataFrame:
    return pd.DataFrame({label: df[cols[0]].str.cat(df[cols[1]], sep=' ')})


def make_y(df: pd.DataFrame, col: str) -> np.ndarray:
    return df.filter(like=col).to_numpy().ravel()


def gen_dataset(df: pd.DataFrame, targ: str, bin: bool = False) -> da.Dataset:
    X = make_X(df)
    y = pd.Series(make_y(df, targ))

    def binned(y: pd.Series):
        bin_labels = get_binned_labels(y)
        return y.map(map_labels(bin_labels, f'{targ}_other'))

    return da.Dataset(da.TrainTestData.from_xy(X, binned(y) if bin else y), targ)


# Training
def bag_of_words(model: base.BaseEstimator, n_pca: int = 100) -> Pipeline:
    return pipeline.make_pipeline(
        preprocessing.FunctionTransformer(lambda df: df.to_numpy().ravel()),
        CountVectorizer(),
        preprocessing.FunctionTransformer(lambda x: x.toarray()),
        StandardScaler(),
        PCA(n_pca),
        model,
    )


def _transformer(trans, n_pca: int) -> Pipeline:
    return pipeline.make_pipeline(
        preprocessing.FunctionTransformer(lambda df: df.to_numpy().ravel()),
        trans,
        preprocessing.FunctionTransformer(lambda x: x.toarray()),
        StandardScaler(),
        PCA(n_pca),
    )


def tfidf(model: base.BaseEstimator, n_pca: int = 100) -> Pipeline:
    return pipeline.make_pipeline(
        _transformer(TfidfVectorizer(), n_pca),
        model,
    )


def build_log_clf():
    return linear_model.LogisticRegressionCV(
        solver='liblinear', class_weight='balanced', n_jobs=-1
    )


def gen_report(dataset: da.Dataset, model: Pipeline) -> da.Report:
    return da.Report.from_estimator(dataset.data.test, model)

issues_data.csv