<jupyter_start><jupyter_text>The distribution is not equal thus we need to create some new samples<jupyter_code>print(num_of_samples)
max_samples = 3000

for num_samples in num_of_samples:
  print('Samples to create:', max_samples - num_samples)
  # for i in range(max_samples - num_samples):
    # create sample<jupyter_output>Samples to create: 2820
Samples to create: 1020
Samples to create: 990
Samples to create: 1740
Samples to create: 1230
Samples to create: 1350
Samples to create: 2640
Samples to create: 1710
Samples to create: 1740
Samples to create: 1680
Samples to create: 1200
Samples to create: 1830
Samples to create: 1110
Samples to create: 1080
Samples to create: 2310
Samples to create: 2460
Samples to create: 2640
Samples to create: 2010
Samples to create: 1920
Samples to create: 2820
Samples to create: 2700
Samples to create: 2730
Samples to create: 2670
Samples to create: 2550
Samples to create: 2760
Samples to create: 1650
Samples to create: 2460
Samples to create: 2790
Samples to create: 2520
Samples to create: 2760
Samples to create: 2610
Samples to create: 2310
Samples to create: 2790
Samples to create: 2401
Samples to create: 2640
Samples to create: 1920
Samples to create: 2670
Samples to create: 2820
Samples to create: 1140
Samples to create: 2730
Samples to create: 2700
Samples to create[...]<jupyter_text>Data augmentation -  Rotating<jupyter_code>def rotate_image(image, max_angle=15):
  """Rotates an image by a given angle
  
  Parameters
  ----------
  image -- numpy array representing an image

  max_angle -- max angle to rotate
    default = 15

  Returns
  -------
  Rotated image by a given angle
  """

  return rotate(image, np.random.uniform(-max_angle, max_angle), mode='edge')
plt.imshow(rotate_image(X_train_data[0]))<jupyter_output><empty_output><jupyter_text>-  Translation (moving image)<jupyter_code>def translate_image(image, max_trans=10, height=32, width=32):
  """Translates an image

  Parameters
  ----------
  image -- numpy array representing an image

  max_trans -- max pixels to translate
    default = 10

  height -- height of the image
    default = 32

  width -- width of the image
    default = 32
  
  Returns
  -------
  Translated image
  """

  translate_x = max_trans*np.random.uniform() - max_trans/2
  translate_y = max_trans*np.random.uniform() - max_trans/2
  translation_mat = np.float32([[1,0,translate_x],[0,1,translate_y]])
  translation = cv2.warpAffine(image, translation_mat, (height, width))
  return translation
plt.imshow(translate_image(X_train_data[0]))<jupyter_output><empty_output><jupyter_text>-  Projection transform (stretching)<jupyter_code>def projection_transform(image, max_warp=0.6, height=32, width=32):
    """Applies a projection transform (stretches an image)

    Parameters
    ----------
    image -- numpy array representing an image

    max_warp -- max warp ratio to stretch
      default = 0.6

    height -- height of the image
      default = 32

    width -- width of the image
      default = 32

    Returns
    -------
    Stretched image
    """

    #Warp Location
    d = height * 0.3 * np.random.uniform(0,max_warp)
    
    #Warp co-ordinates
    tl_top = np.random.uniform(-d, d)     # Top left corner, top margin
    tl_left = np.random.uniform(-d, d)    # Top left corner, left margin
    bl_bottom = np.random.uniform(-d, d)  # Bottom left corner, bottom margin
    bl_left = np.random.uniform(-d, d)    # Bottom left corner, left margin
    tr_top = np.random.uniform(-d, d)     # Top right corner, top margin
    tr_right = np.random.uniform(-d, d)   # Top right corner, right margin
    br_bottom = np.random.uniform(-d, d)  # Bottom right corner, bottom margin
    br_right = np.random.uniform(-d, d)   # Bottom right corner, right margin
        
    ##Apply Projection
    transform = ProjectiveTransform()
    transform.estimate(np.array((
                (tl_left, tl_top),
                (bl_left, height - bl_bottom),
                (height - br_right, height - br_bottom),
                (height - tr_right, tr_top)
            )), np.array((
                (0, 0),
                (0, height),
                (height, height),
                (height, 0)
            )))
    output_image = warp(image, transform, output_shape=(height, width), order = 1, mode = 'edge')
    return output_image
plt.imshow(projection_transform(X_train_data[0]))<jupyter_output><empty_output><jupyter_text>Transformation in a pipeline<jupyter_code>def transform_image_pipeline(image, max_angle=15, max_trans=5, max_warp=0.8):
  """Applies all previous transformations to an image
  
  Parameters
  ----------
  image -- numpy array representing an image

  max_angle -- max angle to rotate
    default = 15

  max_trans -- max pixels to translate
    default = 10
  
  max_warp -- max warp ratio to stretch
    default = 0.6

  height -- height of the image
    default = 32

  width -- width of the image
    default = 32
  
  Returns
  -------
  Image after all transformations
  """

  height, width, channels = image.shape

  # Rotating image
  rotated_image = rotate_image(image, max_angle)

  # Translating image
  translated_image = translate_image(rotated_image, max_trans, height, width)

  # Project image
  projected_image = projection_transform(translated_image, max_warp, height, width)

  return projected_image


def data_augment(image):
  """Applies rotation, warp and biteral filtering to an image

  Parameters
  ----------
  image -- numpy array representing an image

  Returns
  -------
  Image after all transformations
  """

  rows= image.shape[0]
  cols = image.shape[1]
  
  # Rotating
  M_rot = cv2.getRotationMatrix2D((cols/2,rows/2),10,1)
  
  # Translating
  M_trans = np.float32([[1,0,3],[0,1,6]])

  img = cv2.warpAffine(image,M_rot,(cols,rows))
  img = cv2.warpAffine(img,M_trans,(cols,rows))
  
  # Bilateral filtering
  img = cv2.bilateralFilter(img, 9, 75, 75)

  return img
plt.imshow(data_augment(X_train_data[0]))<jupyter_output><empty_output><jupyter_text>Creating new samples<jupyter_code>def create_new_samples(X_train_data, y_train, n_classes):
  """Creates new data samples

  Parameters
  ----------
  X_train_data -- image dataset

  y_train -- dataset labels

  n_classes -- total number of classes

  Returns
  -------
  Newly created dataset
  """

  X_train_aug = X_train_data.copy()
  y_train_aug = y_train.copy()

  X_aug = []
  Y_aug = []

  for i in range(0, n_classes):
    class_records = np.where(y_train==i)[0].size
    max_records = 3000  # max(num_samples)
    if class_records != max_records:
        ovr_sample = max_records - class_records
        samples = X_train_data[np.where(y_train==i)[0]]
        X_aug_temp = []
        Y_aug_temp = [i] * ovr_sample
        print('Class:', i)
        
        for x in range(ovr_sample):
            img = samples[x % class_records]
            trans_img = data_augment(img)
            X_aug_temp.append(trans_img)
            
        X_train_aug = np.concatenate((X_train_aug, X_aug_temp), axis=0)
        y_train_aug = np.concatenate((y_train_aug, Y_aug_temp)) 
        
        Y_aug = Y_aug + Y_aug_temp
        X_aug = X_aug + X_aug_temp
  return X_train_aug, y_train_aug
# # RUN IF NECESSARY

# X_train_aug, y_train_aug = create_new_samples(X_train_data, y_train_data, num_classes)

# X_test_aug, y_test_aug = create_new_samples(X_test_data, y_test_data, num_classes)
# X_val_aug, y_val_aug = create_new_samples(X_val_data, y_val_data, num_classes)
# a = 3
# for j in range(10):
#   fig = plt.figure(figsize = (10, 1))
#   fig.subplots_adjust(left = 0, right = 1, bottom = 0, top = 1, hspace = 0.05, wspace = 0.05)
#   j=1
#   a+=20
#   for i in range(10):
#       axis = fig.add_subplot(1, 10, i + 1, xticks=[], yticks=[])
#       axis.imshow(X_train_aug[i+120000 + j + a])
#       j+=20<jupyter_output><empty_output><jupyter_text>Saving data<jupyter_code># # RUN IF NECESSARY

# with open('aug_data/augmented_X_train_data.p', 'wb') as fx:
#   pickle.dump(X_train_aug, fx)

# with open('aug_data/augmented_y_train_data.p', 'wb') as fy:
#   pickle.dump(y_train_aug, fy)<jupyter_output><empty_output><jupyter_text>Loading newly created data<jupyter_code>pickle_aug_files = get_pickle_files_names('aug_data/')
pickle_aug_files
# # Loading all data
X_train_aug = load_pickle_file(pickle_aug_files[0])
y_train_aug = load_pickle_file(pickle_aug_files[1])

# # Checking the shapes
print(X_train_aug.shape)
print(y_train_aug.shape)<jupyter_output>(129000, 32, 32, 3)
(129000,)<jupyter_text>Data preprocessing<jupyter_code>def convert_to_grayscale(image):
  """Converts an image to grayscale
  
  Parameters
  ----------
  image -- a numpy array represeting an image wth 3 color channels

  Returns
  -------
  image in grayscale without a color channel;
  """
  
  return cv2.cvtColor(image.astype('uint8') * 255, cv2.COLOR_BGR2GRAY)


def normalize_image(image):
  """Normalizes an image by dividing it by 255
  
  Parameters
  ----------
  image -- a numpy array representing an image

  Returns
  -------
  Normalized image  
  """
  return image / 255


def histogram_equalization(image):
  """
  Parameters
  ----------

  image -- a numpy array representing an image in grayscale (without a color channel)

  Returns
  -------

  image of equalized grayscale
  """
  
  return cv2.equalizeHist(image)


def preprocess_image(image):
  """Applies grayscale transformation, equalization and normalization to an image

  Parameters
  ----------

  image -- a numpy array representing an image

  Returns
  -------
  preprocessed image

  """
  gray = convert_to_grayscale(image)
  equalized = histogram_equalization(gray)
  normalized = normalize_image(equalized)
  
  return normalized<jupyter_output><empty_output><jupyter_text>Grayscale<jupyter_code>plt.imshow(X_train_aug[10000], cmap='gray')
plt.axis('off')
gray_example = convert_to_grayscale(X_train_aug[10000])
plt.imshow(gray_example, cmap='gray')
plt.axis('off')
print(gray_example.shape)<jupyter_output>(32, 32)<jupyter_text>Equalization<jupyter_code>eh_image = histogram_equalization(gray_example)
plt.imshow(eh_image, cmap='gray')
plt.axis('off')
print(eh_image.shape)<jupyter_output>(32, 32)<jupyter_text>Normalization<jupyter_code>normalized_example = normalize_image(gray_example)

plt.imshow(normalized_example, cmap='gray')
plt.axis('off')
print(normalized_example.shape)<jupyter_output>(32, 32)<jupyter_text>Applying to the whole dataset<jupyter_code>X_train_preprocessed = np.array(list(map(preprocess_image, X_train_aug)))
X_test_preprocessed = np.array(list(map(preprocess_image, X_test_data)))
X_val_preprocessed = np.array(list(map(preprocess_image, X_val_data)))
plt.imshow(X_train_preprocessed[random.randint(0, len(X_train_preprocessed - 1))], cmap='gray')
plt.axis('off')<jupyter_output><empty_output><jupyter_text>Adding depth<jupyter_code>print(X_train_preprocessed.shape)
X_train = X_train_preprocessed.reshape((129000, 32, 32, 1))
X_test = X_test_preprocessed.reshape((12630, 32, 32, 1))
X_val = X_val_preprocessed.reshape((4410, 32, 32, 1))
print(X_train.shape)
print(X_test.shape)
print(X_val.shape)<jupyter_output>(129000, 32, 32, 1)
(12630, 32, 32, 1)
(4410, 32, 32, 1)<jupyter_text>OneHotEncoding labels<jupyter_code>y_train_aug.shape
y_train = to_categorical(y_train_aug, num_classes)
y_val = to_categorical(y_val_data, num_classes)
y_train.shape<jupyter_output><empty_output><jupyter_text>Creating model<jupyter_code>def create_second_model():
  """Creates DL a model with CNN's

  Returns
  -------
  A DL model with CNN's
  """

  model = Sequential()

  # Convolutional layer - extracting features
  model.add(Conv2D(filters=60, kernel_size=(5, 5), activation='relu', input_shape=(32, 32, 1), padding='same'))
  # now it will have the shape of (28, 28, 30)
  # 5 * 5 * 30 + 30 = 780 adjustable parameters
  model.add(Conv2D(filters=60, kernel_size=(5, 5), activation='relu'))

  # Pooling layers - downsampling
  model.add(MaxPooling2D(pool_size=(2, 2)))  
  # scaling the image to a 2 times smaller size (14, 14, 30)
  # 3 * 3 * 15 * 30 + 15 = 4065 adjustable parameters

  # Conv
  model.add(Conv2D(filters=30, kernel_size=(3, 3), activation='relu'))
  model.add(Conv2D(filters=30, kernel_size=(3, 3), activation='relu'))
  # now it will have the shape of (12, 12, 15)

  # Pooling
  model.add(MaxPooling2D(pool_size=(2, 2)))  
  # producing an image of size (6, 6, 15)

  model.add(Dropout(0.5))

  # Flattening
  model.add(Flatten())  # reshaping an image from (6, 6, 15) to (6*6*15, 1)

  # Dense layer
  model.add(Dense(units=700, activation='relu'))
  
  # Dropout layer
  model.add(Dropout(0.5))
  
  # Output layer
  model.add(Dense(units=num_classes, activation='softmax'))
  
  # Displaying model summary
  print(model.summary())

  # Compile model
  model.compile(Adam(lr=0.001), loss='categorical_crossentropy', metrics=['accuracy'])

  return model
model = create_second_model()
# checkpoint_path = "training_1/cp.ckpt"
# checkpoint_dir = os.path.dirname(checkpoint_path)

# # Create a callback that saves the model's weights
# cp_callback = ModelCheckpoint(filepath=checkpoint_path,
#                                                  save_weights_only=True,
#                                                  verbose=1)
history = model.fit(X_train,
                    y_train, 
                    epochs=10, 
                    validation_data=(X_val, y_val), 
                    batch_size=200, 
                    verbose=1, 
                    shuffle=1
                    )
plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])

plt.xlabel('Epoch')
plt.title('Accuracy')
plt.legend(['training', 'validation'])
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])

plt.xlabel('Epoch')
plt.title('Loss')
plt.legend(['training', 'validation'])
score, acc = model.evaluate(X_test, y_test, verbose=0)

print('Test score:', score)
print('Test accuracy:', acc)
def get_pic_names(path=''):
  """  
  Parameters
  ----------
  path -- path to a directory
    default = ''

  Returns
  -------
  List of all files ending with .jpg from a specific directory 
  """
  
  return glob.glob(path + '/*.jpg')
test_pics = get_pic_names('test_sign_images')
test_pics
fig, ax = plt.subplots(nrows=len(test_pics), ncols = 1, figsize=(5, 50))
fig.tight_layout()

for i, test_pic in enumerate(test_pics):
  img = cv2.imread(test_pic)

  img = np.asarray(img)
  resized_img = cv2.resize(img, (32, 32))
  preprocessed_img = preprocess_image(resized_img)
  # plt.imshow(preprocessed_img, cmap = plt.get_cmap('gray'))
  print(preprocessed_img.shape)

  ax[i].imshow(preprocessed_img, cmap=plt.get_cmap("gray"))
  ax[i].axis("off")

  test_image = preprocessed_img.reshape(1, 32, 32, 1)
  
  predicted_class_num = int(model.predict_classes(test_image))

  print('Predicted sign:', str(sign_names['SignName'][sign_names['ClassId'] == predicted_class_num]))<jupyter_output><empty_output><jupyter_text>Data augmentaton using ImageDataGenerator<jupyter_code>X_train_data.shape
y_train_data.shape
X_train_pre = np.array(list(map(preprocess_image, X_train_data)))
X_test_pre = np.array(list(map(preprocess_image, X_test_data)))
X_val_pre = np.array(list(map(preprocess_image, X_val_data)))
X_train = X_train_pre.reshape((34799, 32, 32, 1))
X_test = X_test_pre.reshape((12630, 32, 32, 1))
X_val = X_val_pre.reshape((4410, 32, 32, 1))
data_geterator = ImageDataGenerator(
                                    width_shift_range=0.1,
                                    height_shift_range=0.1,
                                    zoom_range=0.2,
                                    shear_range=0.1,
                                    rotation_range=10
                                    )<jupyter_output><empty_output><jupyter_text>OneHotEncoding labels<jupyter_code>y_train = to_categorical(y_train_data)
y_test = to_categorical(y_test_data, num_classes)
# Fitting the data
data_geterator.fit(X_train)
history = model.fit_generator(
                              data_geterator.flow(X_train, y_train, batch_size=300),
                              steps_per_epoch=200,
                              epochs=10,
                              validation_data=(X_val, y_val),
                              shuffle=1,
                              verbose=1
                              )
path_to_model = 'models/best_model'
model.save(path_to_model)
model = keras.models.load_model(path_to_model)
loss, acc = model.evaluate(X_test, y_test, verbose=0)

print('Test loss:', loss)
print('Test accuracy:', acc)
plt.plot(history.history['accuracy'])
plt.plot(history.history['val_accuracy'])

plt.xlabel('Epoch')
plt.title('Accuracy')
plt.legend(['training', 'validation'])
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])

plt.xlabel('Epoch')
plt.title('Loss')
plt.legend(['training', 'validation'])
fig, ax = plt.subplots(nrows=len(test_pics), ncols = 1, figsize=(5, 50))
fig.tight_layout()

for i, test_pic in enumerate(test_pics):
  img = cv2.imread(test_pic)

  img = np.asarray(img)
  resized_img = cv2.resize(img, (32, 32))
  preprocessed_img = preprocess_image(resized_img)
  # plt.imshow(preprocessed_img, cmap = plt.get_cmap('gray'))
  print(preprocessed_img.shape)

  ax[i].imshow(preprocessed_img, cmap=plt.get_cmap("gray"))
  ax[i].axis("off")

  test_image = preprocessed_img.reshape(1, 32, 32, 1)
  
  predicted_class_num = int(model.predict_classes(test_image))

  print('Predicted sign:', str(sign_names['SignName'][sign_names['ClassId'] == predicted_class_num]))

from tensorflow.keras.utils import plot_model

plot_model(model, show_shapes=True, show_layer_names=True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12338285972543923, 'fraction_numerical': 0.03400937800124287, 'mean_word_length': 3.5753424657534247, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '237580', 'license': 'MIT', 'n_tokens_mistral': 6704, 'n_tokens_neox': 6018, 'n_words': 1332, 'path': 'tsc_model.ipynb', 'repo_name': 'JakubTomaszewski/Road_signs_classification', 'size': '17701'}