<jupyter_start><jupyter_text>Task -Project MalJPEG  Imports<jupyter_code># Imports
import numpy as np # Support for large arrays and matrices, along with high-level mathematical functions.
import seaborn as sns # Graphing/Plotting module.
import pandas as pd # CSV handling with operations on tabular data.
import lightgbm as lgb

from ast import literal_eval
from xgboost import XGBClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.tree import DecisionTreeClassifier
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report,confusion_matrix, accuracy_score, roc_auc_score
from sklearn.preprocessing import LabelBinarizer, StandardScaler, LabelEncoder

from collections import Counter
def checkPredictions(lgb_p, dtc_p, rfc_p, xgb_p, labels):
    for idx, _ in enumerate(lgb_p):
        if lgb_p[idx]!= labels[idx] or dtc_p[idx]!= labels[idx] or rfc_p[idx]!= labels[idx] or xgb_p[idx]!= labels[idx]:
            print(lgb_p[idx], dtc_p[idx], rfc_p[idx], xgb_p[idx], labels[idx])

# Post ML
def count_voles(votes):
    # count the votes for type
    vote_count=Counter(votes)
 
    # find the maximum number of votes
    max_votes = max(vote_count.values())
 
    # search for type having maximum votes and store in a list
    lst=[i for i in vote_count.keys() if vote_count[i]==max_votes]
 
    # sort the list
    return max_votes, sorted(lst)[0]

def final_predictions(dtc_p, rfc_p, xgb_p):
    final_result = []
    for idx, _ in enumerate(dtc_p):
        if dtc_p[idx] != rfc_p[idx] or dtc_p[idx] != xgb_p[idx] or rfc_p[idx] != xgb_p[idx]:
            max_votes, chosen = count_voles([dtc_p[idx], rfc_p[idx], xgb_p[idx]])
            if max_votes == 1:
                final_result.append(xgb_p[idx])
            else:
                final_result.append(chosen)
        else:
            final_result.append(dtc_p[idx])
    return final_result<jupyter_output><empty_output><jupyter_text>Read Data and Preprocess it to fit into DeepMAL model<jupyter_code># Read Data
dataset_type_1 = 'markers_image_my.csv' # other options: 'markers_image.csv' or 'metadata.csv'

filepath = f'./datasets/{dataset_type_1}'
df_1 = pd.read_csv(filepath)

dataset_type_2 = 'metadata.csv' # other options: 'markers_image.csv' or 'metadata.csv'

filepath = f'./datasets/{dataset_type_2}'
df_2 = pd.read_csv(filepath)
label_encoder = LabelEncoder()
df_2['Make'] = label_encoder.fit_transform(df_2['Make'])
df = df_1.merge(df_2, on='filename', how='left')<jupyter_output><empty_output><jupyter_text>Preprocess the data<jupyter_code>label_type = 'label_x' # options: 'label'

X = np.stack([
    # MalJPEG
    df['Marker_EOI_content_after_num'], df['File_markers_num'], df['File_size'],
    df['Marker_APP1_size_max'], df['Marker_APP12_size_max'], df['Marker_COM_size_max'], 
    df['Marker_DHT_num'], df['Marker_DHT_size_max'], df['Marker_DQT_num'], 
    df['Marker_DQT_size_max'],
    # My features
    df['Marker_APP_other_size_max'], df['Marker_APP_other_num'], 
    df['Marker_SOF_size_max'], df['Marker_DRI_num'], df['Marker_RST_num'],
    df['Marker_SOS_len'], df['Marker_EOI'],
    # My features from metadata
    df['ExifInteroperabilityOffset'], df['SceneCaptureType'],  df['MeteringMode'], 
    df['LightSource'], df['Flash'], df['ColorSpace'],  
    df['Make'], df['ExposureProgram'], df['ResolutionUnit'], df['Exception_num'],
]).T

y = np.stack(df[label_type])

scaler_mix = StandardScaler()
scaler_mix.fit(X)
X = scaler_mix.transform(X)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.30, random_state=1666, stratify=y)<jupyter_output><empty_output><jupyter_text>Train/Fit<jupyter_code># LGBM classifier
lgb_clf = lgb.LGBMClassifier(n_estimators=500, n_jobs=4, random_state=2021,
                             boosting_type='gbdt', class_weight=None, colsample_bytree=1.0,
                             importance_type='split', learning_rate=0.1, max_depth=-1,
                             min_child_samples=20, min_child_weight=0.001, min_split_gain=0.0,
                             num_leaves=31, objective=None, reg_alpha=0.0, reg_lambda=0.0,
                             subsample=1.0, subsample_for_bin=200000, subsample_freq=0)
lgb_clf.fit(X_train, y_train)
# Decision Tree classifier
dtc_clf = DecisionTreeClassifier(max_depth=30, random_state=1914)
dtc_clf.fit(X_train, y_train)
# Random Forest classifier
rfc_clf = RandomForestClassifier(n_estimators=500, max_depth=30, n_jobs=4, random_state=2021)
rfc_clf.fit(X_train, y_train)
# XGB classifier
xgb_clf = XGBClassifier(n_estimators=500, max_depth=30, n_jobs=4, eval_metric='mlogloss', 
                        random_state=1941, use_label_encoder=False)
xgb_clf.fit(X_train, y_train)<jupyter_output><empty_output><jupyter_text>Test/Predict<jupyter_code>lgb_predictions = lgb_clf.predict(X_test)
dtc_predictions = dtc_clf.predict(X_test)
rfc_predictions = rfc_clf.predict(X_test)
xgb_predictions = xgb_clf.predict(X_test)
    
true_labels = y_test
cf_matrix = confusion_matrix(true_labels, lgb_predictions)

accuracy = accuracy_score(true_labels, lgb_predictions)
print("accuracy score: {0:.2f}%".format(accuracy*100))
print("TPR: {0:.3f}".format(cf_matrix[1][1]/(cf_matrix[0][1]+cf_matrix[1][1])))
print("FPR: {0:.3f}".format(cf_matrix[0][1]/(cf_matrix[0][0]+cf_matrix[1][0])))
print("AUC&ROC", roc_auc_score(true_labels, lgb_predictions))
print(classification_report(true_labels, lgb_predictions))

heatmap = sns.heatmap(cf_matrix, annot=True, cmap='Blues', fmt='g', 
                      xticklabels=np.unique(true_labels), 
                      yticklabels=np.unique(true_labels))
cf_matrix = confusion_matrix(true_labels, dtc_predictions)

accuracy = accuracy_score(true_labels, dtc_predictions)
print("accuracy score: {0:.2f}%".format(accuracy*100))
print("TPR: {0:.3f}".format(cf_matrix[1][1]/(cf_matrix[0][1]+cf_matrix[1][1])))
print("FPR: {0:.3f}".format(cf_matrix[0][1]/(cf_matrix[0][0]+cf_matrix[1][0])))
print("AUC&ROC", roc_auc_score(true_labels, dtc_predictions))
print(classification_report(true_labels, dtc_predictions))

heatmap = sns.heatmap(cf_matrix, annot=True, cmap='Blues', fmt='g', 
                      xticklabels=np.unique(true_labels), 
                      yticklabels=np.unique(true_labels))
cf_matrix = confusion_matrix(true_labels, rfc_predictions)

accuracy = accuracy_score(true_labels, rfc_predictions)
print("accuracy score: {0:.2f}%".format(accuracy*100))
print("TPR: {0:.3f}".format(cf_matrix[1][1]/(cf_matrix[0][1]+cf_matrix[1][1])))
print("FPR: {0:.3f}".format(cf_matrix[0][1]/(cf_matrix[0][0]+cf_matrix[1][0])))
print("AUC&ROC", roc_auc_score(true_labels, rfc_predictions))
print(classification_report(true_labels, rfc_predictions))

heatmap = sns.heatmap(cf_matrix, annot=True, cmap='Blues', fmt='g', 
                      xticklabels=np.unique(true_labels), 
                      yticklabels=np.unique(true_labels))
cf_matrix = confusion_matrix(true_labels, xgb_predictions)

accuracy = accuracy_score(true_labels, xgb_predictions)
print("accuracy score: {0:.2f}%".format(accuracy*100))
print("TPR: {0:.3f}".format(cf_matrix[1][1]/(cf_matrix[0][1]+cf_matrix[1][1])))
print("FPR: {0:.3f}".format(cf_matrix[0][1]/(cf_matrix[0][0]+cf_matrix[1][0])))
print("AUC&ROC", roc_auc_score(true_labels, xgb_predictions))
print(classification_report(true_labels, xgb_predictions))

heatmap = sns.heatmap(cf_matrix, annot=True, cmap='Blues', fmt='g', 
                      xticklabels=np.unique(true_labels), 
                      yticklabels=np.unique(true_labels))
print(lgb_clf.feature_importances_)
print(dtc_clf.feature_importances_)
print(rfc_clf.feature_importances_)
print(xgb_clf.feature_importances_)
predictions = final_predictions(xgb_predictions, lgb_predictions, rfc_predictions)
cf_matrix = confusion_matrix(true_labels, predictions)

accuracy = accuracy_score(true_labels, predictions)
print("accuracy score: {0:.2f}%".format(accuracy*100))
print("TPR: {0:.3f}".format(cf_matrix[1][1]/(cf_matrix[0][1]+cf_matrix[1][1])))
print("FPR: {0:.3f}".format(cf_matrix[0][1]/(cf_matrix[0][0]+cf_matrix[1][0])))
print("AUC&ROC", roc_auc_score(true_labels, predictions))
print(classification_report(true_labels, predictions))

heatmap = sns.heatmap(cf_matrix, annot=True, cmap='Blues', fmt='g', 
                      xticklabels=np.unique(true_labels), 
                      yticklabels=np.unique(true_labels))
checkPredictions(lgb_predictions, dtc_predictions, rfc_predictions, xgb_predictions, true_labels)<jupyter_output>1 0 0 0 0
1 1 1 1 0
1 1 1 1 0
1 1 1 1 0
0 0 0 0 1
0 0 0 0 1
0 1 1 1 0
0 0 0 0 1
0 1 0 0 0
0 0 0 0 1
StarcoderdataJupyterStructuredCleanDedup