Upload 21 files

SightLinks-Dev-main/DevelopmentTools/ClassUtils.py

@@ -0,0 +1,64 @@
+# All the utilities required for the inference and further tuning of the classification models
+
+import torch
+from torchvision import transforms
+from torch.utils.data import Dataset
+
+import os
+import numpy as np
+from PIL import Image
+
+
+# The custom dataset object I used to load the segmented image dataset - will only fit our format of data!!!
+class CrosswalkDataset(Dataset):
+    def __init__(self, src_dir, transform=None):
+        self.src_dir = src_dir
+        self.transform = transform
+
+        dir_files = sorted(os.listdir(src_dir))
+        self.image_paths = [file_path for file_path in dir_files if file_path.endswith((".png", ".jpg", ".jpeg"))]
+        self.label_paths = [file_path for file_path in dir_files if file_path.endswith(".txt")]
+
+    def __len__(self):
+        return len(self.image_paths)
+    
+    def __getitem__(self, index):
+        image_path = os.path.join(self.src_dir, self.image_paths[index])
+        label_path = os.path.join(self.src_dir, self.label_paths[index])
+
+        label = [0, 0]
+        try:
+            if np.array([int(open(label_path).read().strip())]) == 1:
+                label = [1, 0]
+            else:
+                label = [0, 1]
+        except:
+            pass
+        image =  Image.open(image_path)
+        
+        if self.transform is None:
+            self.transform = transforms.ToTensor()
+            
+        return (self.transform(image), torch.FloatTensor(label))
+
+
+# Mean and Std. are chosen arbitrarily - need to be tuned
+# The image do not have to be resized - the global pooling layer should technically deal with this, but I haven't tested this,
+# so resizing prevents potential inaccuracies from occuring,
+vgg_transform = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.5, 0.5, 0.5],  std=[0.3, 0.3, 0.3])
+])
+
+res_transform = transforms.Compose([
+    transforms.Resize((256, 256)),
+    transforms.ToTensor(),
+    transforms.Normalize(mean=[0.5, 0.5, 0.5],  std=[0.3, 0.3, 0.3])
+])
+
+mob3_transform = transforms.Compose([
+    transforms.Resize((224, 224)),
+    transforms.ToTensor(),
+    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225])
+])

SightLinks-Dev-main/DevelopmentTools/Classify.py

@@ -0,0 +1,168 @@
+import ClassUtils
+import LoadUtils
+
+import torch
+import torchvision
+import torchvision.models as models
+import matplotlib.pyplot as plt
+import numpy as np
+import random
+
+import warnings
+
+# Torchvision's models utils has a depreciation warning for the pretrained parameter in its instantiation but we don't use that
+warnings.filterwarnings(
+    action='ignore',
+    category=DeprecationWarning,
+    module=r'.*'
+)
+
+vgg16_state_path = "VGG16_Full_State_Dict.pth"
+# A prototype v2 version that's only been trained on 2000 images by transfer learning
+mobileNet_path = "MobileNetV3_state_dict_big_train.pth"
+data_path = "zebra_annotations/classification_data"
+
+classify = None
+transform = None
+
+# Loads a given VGG binary classifier state dictionairy into a model, for transfer learning or immediate use
+def load_vgg_classifier(state_dict_path):
+    # Ignore depreciation warnings --> It works fine for our needs
+    model = models.vgg16()
+
+    # Modifies fully connected layer to output binary class predictions
+    model.classifier[6] = torch.nn.Linear(model.classifier[6].in_features, 2)
+    state_dict = torch.load(state_dict_path, weights_only=True)
+    model.load_state_dict(state_dict)
+
+    model.eval()
+
+    return model
+
+# Only loads the classifier weights, in the case where it is transfer learning on only the top 
+# or the feature extraction has been frozen during training
+# This saves a significant amount of space
+def partial_vgg_load(classifier_state_dict_path):
+    model = models.vgg16(weights=models.VGG16_Weights.DEFAULT)
+
+    model.classifier[6] = torch.nn.Linear(model.classifier[6].in_features, 2)
+    model.classifier.load_state_dict(classifier_state_dict_path)
+
+    model.eval()
+
+    return model
+
+# Loads a given ResNet binary classifier state dictionairy into a model, for transfer learning or immediate use
+def load_resnet_classifier(state_dict_path):
+    # Ignore depreciation warnings --> It works fine for our needs
+    resnet = models.resnet18(pretrained=True)
+    resnet.fc = torch.nn.Linear(resnet.fc.in_features, 1)
+
+    state_dict = torch.load(state_dict_path, weights_only=True)
+    resnet.load_state_dict(state_dict)
+    
+    resnet.eval()
+    return resnet
+
+# Loads a given MN3 binary classifier state dictionairy into a model, for transfer learning or immediate use
+# We use this for our current version
+def load_mobileNet_classifier(state_dict_path):
+    # Ignore depreciation warnings --> It works fine for our needs
+    model = models.mobilenet_v3_small()
+    model.classifier[3] = torch.nn.Linear(model.classifier[3].in_features, 2)
+
+    state_dict = torch.load(state_dict_path, weights_only=True)
+    model.load_state_dict(state_dict)
+
+    model.eval()
+    return model
+
+# classify = load_vgg_classifier(vgg16_state_path)
+# transform = ClassUtils.vgg_transform
+
+classify = load_mobileNet_classifier(mobileNet_path)
+transform = ClassUtils.mob3_transform
+
+
+
+# Takes a numpy array representing an image and returns a boolean classification based on the given threshold value
+# Defaults to 0.35 as it errs on the side of letting images through.
+def infer(image, infer_model=classify, infer_transform=transform):
+
+    # If infer model and transform have not been initialised, the function cannot run so the program will throw an error
+    #
+    # This was chosen over intiialising default values since this way it is clear an error is occuring rather than be hidden
+    # in a bunch of system logs and potenitally continually causing errors
+    if infer_model is None or infer_transform is None:
+        raise TypeError("Error: The inference classes have not been initialised properly.")
+    if not torch.is_tensor(image):
+        image = infer_transform(image)
+    
+    # Expects batches - this adds another dimensions to properly format the data
+    if len(image.shape) <= 3:
+        image = image.unsqueeze(0)
+
+    logit_pred = infer_model(image)
+
+    probs = 1 / (1 + np.exp(-logit_pred.detach().numpy()))
+    # prob = max(0, min(np.exp(logit_pred.detach().numpy())[0], 1))
+    return probs
+
+
+# Takes a PIL image and returns a boolean classification based on the given threshold value
+# Defaults to 0.35 as it errs on the side of letting images through.
+def PIL_infer(image, threshold=0.35):
+    tensor_im = torchvision.transforms.functional.pil_to_tensor(image).float()/ 255
+    prediction = infer(tensor_im)
+    classification = prediction[0][0] > threshold
+    return classification
+
+# For testing and demo purposes
+def infer_and_display(image, threshold, actual_label, onlyWrong=False):
+    probability = infer(image)
+    prediction = probability > threshold
+    is_correct = (actual_label[0] == 1) == prediction
+
+    if onlyWrong and is_correct:
+        return prediction
+    
+    plt.imshow(torch.permute(image, (1, 2, 0)).detach().numpy())
+    plt.title(f"Prediction: {prediction[0][0]} with confidence {probability[0][0]}%, Actual: {actual_label[0] == 1}")
+    plt.axis("off")
+    plt.show()
+
+    return probability
+
+
+# Template code for how to create an inference code
+def example_init(examples=20, display=True):
+    dataset = ClassUtils.CrosswalkDataset(data_path)
+    
+    random_points = [random.randint(0, len(dataset)-1) for i in range(examples)]
+    correct, incorrect, falsepos, falseneg = 0, 0, 0, 0
+    for point in random_points:
+        image, label = dataset[point]
+
+        class_guess = [0, 1]
+        if infer(image)[0][0] > 0.5:
+            class_guess = [1, 0]
+        if class_guess == label.tolist():
+            correct += 1
+        else:
+            if class_guess[0]:
+                falsepos += 1
+            else:
+                falseneg += 1
+            incorrect += 1
+        
+        if display:
+            print(f"Prediction of {infer_and_display(image, 0.4, label)}% of a crosswalk (Crosswalk: {label[0]==1})")
+    print(f"correct: {correct}, incorrect: {incorrect}, of which false positives were {falsepos} and false negatives were {falseneg}")
+
+if __name__ == "__main__":
+    example_init(examples=200,display=False)
+
+else:
+    print(f"Module: [{__name__}] has been loaded")
+
+

SightLinks-Dev-main/DevelopmentTools/DataUtils.py

@@ -0,0 +1,39 @@
+from shapely.geometry import Polygon
+import yaml
+from pathlib import Path
+
+# Each box has format (x_1, y_1, x_2, y_2) - this does mean this is an estimate.
+# box_1 is the bounding box of the crosswalk, and box_2 is that of the tile.
+def check_box_intersection(box_1, box_2, threshold=0.6):
+    # Threshold is the min IoU required to consider it as having a crosswalk - the minimum percent area of the crosswalk that must be in the tile
+    formatted_box_1 = [[box_1[0], box_1[1]], [box_1[2], box_1[1]], [box_1[2], box_1[3]], [box_1[0], box_1[3]]]  # Formatting follows shapely clockwise system
+
+    poly_1 = Polygon(formatted_box_1)
+    poly_2 = Polygon(box_2)
+    try:
+        iou = poly_1.intersection(poly_2).area / poly_1.union(poly_2).area
+        scaled_iou = iou * ((poly_1.area) / (poly_2.area))
+        # print("--------------")
+        # print(iou, scaled_iou)
+        return (scaled_iou > threshold)
+    except ZeroDivisionError:
+        # In the case of a zero division error
+        print("ZERO DIVISION ERROR", poly_1.area, poly_2.area)
+        return False
+
+# Loads a YAML dataset into the user's device
+def load_yaml_database(yaml_path):
+    config_file = None
+
+    with open(yaml_path, "r") as file:
+        config_file = yaml.safe_load(file)
+    
+    root = Path(config_file['path'])
+    train_dir = root / config_file['train']  # Training data
+    valid_dir = root / config_file['val']  # Validation data
+    test_dir = root / config_file['test']  # Testing data
+    image_size = config_file['img_size']
+    classes = config_file['names']
+
+    # Returns in format (directories, label_description)
+    return (train_dir, valid_dir, test_dir), (image_size, classes)

SightLinks-Dev-main/DevelopmentTools/EvaluatePerformance.py

@@ -0,0 +1,126 @@
+import time
+import numpy as np
+
+import Classify
+# We import each model class inside of their respective test method to prevent long load times of initialising all of them
+
+
+# These are the default functions used for evaluation - overwrite or add more parameters as is required for your testing
+# We did not use these for our testing - we calculated it manually - but this is a much easier way
+class Results:
+    def __init__(self):
+        self.accuracy = None
+        self.precision = None
+        self.recall = None
+        self.F1 = None
+
+    # Proportion of all predictions that were right - basically what did it get right
+    def calculate_accuracy(self, correct_pos, correct_neg, total):
+        return (correct_pos + correct_neg) / total if total > 0 else 0
+
+    # Proportion of all positive predictions that were actually positive - aka if it predicted positive, how often was it
+    # actually right
+    def calculate_precision(self, correct_pos, false_pos):
+        return correct_pos / (correct_pos + false_pos) if (correct_pos + false_pos) > 0 else 0
+
+    # Proportion of all positive cases that were predicted positive - aka how many positive images did it correctly predict
+    def calculate_recall(self, correct_pos, false_neg):
+        return correct_pos / (correct_pos + false_neg) if (correct_pos + false_neg) > 0 else 0
+
+    # A combination of precision and recall that takes both of them into consideration - a decent 'summary' accuracy metric
+    def calculateF1(self, precision, recall): 
+        return 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0
+
+
+# Purely for comparing the performance of Binary Classification models
+class Evaluate:
+    def __init__(self):
+        self.model = None
+        self.dataLoader=None
+        self.threshold=0.5
+        pass
+
+    # GENERALISED EVALUATION FUNCTION FOR COMPARISION BETWEEN MODEL ARCHITECTURES - USE SPECIFIC EVAL FUNC FOR TESTING TRAINING SETUPS
+    # Tests performance of a model on unseen data - this is the function we used to evaluate our classification models during training
+    # to determine the best model architecture to use.
+    def run_test(self, verbose=True, visual=False):
+        if self.model is None or self.dataLoader is None:
+            raise AttributeError("Please choose a model to test before running the test")
+        
+        self.model.eval()
+
+        total, correct_pos, correct_neg, false_pos, false_neg = 0, 0, 0, 0, 0
+        running_average_time = 0.0
+        collated_results = Results()
+        incorrect = []
+
+        for image, gt in self.dataLoader:
+            current_start_time = time.time()
+            prediction = Classify.infer(image, self.model)
+            running_average_time += time.time() - current_start_time
+            positive, negative = prediction[prediction[:, 0] > self.threshold], prediction[prediction[:, 0] <= self.threshold]
+            positive_gt, negative_gt = gt[prediction[:, 0] > self.threshold], gt[prediction[:, 0] <= self.threshold]
+
+            correct_pos += len(positive[positive_gt[:, 0]==1])
+            correct_neg += len(negative[negative_gt[:, 0]==0])
+            false_pos += len(positive[positive_gt[:, 0]==0])
+            false_neg += len(negative[negative_gt[:, 0]==1])
+            total += min(self.dataLoader.batch_size, len(image))
+
+            false_pos_mask = (prediction[:, 0] > self.threshold) & (gt[:, 0].detach().numpy() == 0)
+            false_neg_mask = (prediction[:, 0] < self.threshold) & (gt[:, 0].detach().numpy() == 1)
+
+            if len(false_pos_mask) > 0:
+                incorrect.append((image[false_pos_mask], gt[false_pos_mask]))
+            if len(false_neg_mask) > 0:
+                incorrect.append((image[false_neg_mask], gt[false_neg_mask]))
+
+        if verbose:
+            print(f"Total Images Processed: [{total}],"
+                  f" \nAccuracy: [{((correct_pos+correct_neg)/total)*100:.2f}%],"
+                  f" \nCorrect Positives: [{correct_pos}], Correct Negatives: [{correct_neg}],"
+                  f" \nFalse Positives: [{false_pos}], False Negatives [{false_neg}],"
+                  f" \nAverage Running Time (s) per image: [{running_average_time / total}]")
+
+        if visual and incorrect:
+            for (img_set, lab_set) in incorrect:
+                for (img, lab) in zip(img_set, lab_set):
+                    if len(img) > 0:
+                        Classify.infer_and_display(img, 0.5, lab)
+
+        return (correct_pos, correct_neg, false_pos, false_neg, total)
+
+    # What we used to test how changes in our training paramters and input data affected performance
+    def test_MobileNet3_default(self, model_state_dict, test_num=1, verbose=True, visual=False) -> Results:
+        import MobileNetV3 as mn3
+
+        # We excluded the last 5% of data samples from training
+        if test_num > len(mn3.dataset) * 0.05:
+            test_num = int((len(mn3.dataset) - 1) * 0.05)
+
+        test_loader = mn3.DataLoader(
+            mn3.Subset(mn3.dataset, mn3.random.sample(list(range(int(len(mn3.dataset) * 0.95), len(mn3.dataset))), test_num)),
+            batch_size=mn3.batch_size, shuffle=False)
+        
+        test_model = Classify.load_mobileNet_classifier(model_state_dict)
+
+        self.model = test_model
+        self.dataLoader = test_loader
+
+        correct_pos, correct_neg, false_pos, false_neg, total = self.run_test(verbose=verbose, visual=visual)
+
+        self.model=None
+        self.dataLoader=None
+        test_results = Results()
+
+        test_results.accuracy = test_results.calculate_accuracy(correct_pos, correct_neg, total)
+        test_results.precision = test_results.calculate_precision(correct_pos, false_pos)
+        test_results.recall = test_results.calculate_recall(correct_pos, false_neg)
+        test_results.F1 = test_results.calculateF1(test_results.precision, test_results.recall)
+
+        return test_results
+        
+
+eval = Evaluate()
+if __name__ == "__main__":
+    mn3_test_results = eval.test_MobileNet3_default("MobileNetV3_state_dict_big_train.pth", test_num=10000, visual=True)

SightLinks-Dev-main/DevelopmentTools/LoadUtils.py

@@ -0,0 +1,158 @@
+import os
+import yaml
+from pathlib import Path
+import shutil
+from PIL import Image
+import math
+import numpy as np
+from shapely.geometry import Polygon
+import matplotlib.pyplot as plt
+from matplotlib.patches import Polygon as PltPolygon
+import random
+
+# Should eventually be moved to a settings class
+zebra_labels = "zebra_annotations/txt_annotations"
+zebra_images = "zebra_annotations/zebra_images"
+save_dir = "zebra_annotations/classification_data"
+segments = 4
+
+# Each box has format (x_1, y_1, x_2, y_2) - this does mean this is an estimate.
+# box_1 is the bounding box of the crosswalk, and box_2 is that of the tile.
+def check_box_intersection(box_1, box_2, threshold=0.6):
+    # Threshold is the min IoU required to consider it as having a crosswalk - the minimum percent area of the crosswalk that must be in the tile
+    formatted_box_1 = [[box_1[0], box_1[1]], [box_1[2], box_1[1]], [box_1[2], box_1[3]], [box_1[0], box_1[3]]]  # Formatting follows shapely clockwise system
+
+    poly_1 = Polygon(formatted_box_1)
+    poly_2 = Polygon(box_2)
+    try:
+        iou = poly_1.intersection(poly_2).area / poly_1.union(poly_2).area
+        scaled_iou = iou * ((poly_1.area) / (poly_2.area))
+        # print("--------------")
+        # print(iou, scaled_iou)
+        return (scaled_iou > threshold)
+    except ZeroDivisionError:
+        # In the case of a zero division error
+        print("ZERO DIVISION ERROR", poly_1.area, poly_2.area)
+        return False
+
+def load_yaml_database(yaml_path):
+    config_file = None
+
+    with open(yaml_path, "r") as file:
+        config_file = yaml.safe_load(file)
+    
+    root = Path(config_file['path'])
+    train_dir = root / config_file['train']  # Training data
+    valid_dir = root / config_file['val']  # Validation data
+    test_dir = root / config_file['test']  # Testing data
+    image_size = config_file['img_size']
+    classes = config_file['names']
+
+    # Returns in format (directories, label_description)
+    return (train_dir, valid_dir, test_dir), (image_size, classes)
+
+
+# Takes in a entity database, labelled for bounding box regression, and breaks down the images into
+# smaller images, with labels for classification training
+def convert_database_to_segments(image_dir, label_dir, dst_dir, overwrite=False):
+    if not os.path.exists(image_dir) or not os.path.exists(label_dir):
+        print("Error: Image or label directories do not exist")
+        return
+    
+    # If there isn't a directory created to save the new data to, create it
+    dst_dir = Path(dst_dir)
+    try:
+        if overwrite and dst_dir.exists() and dst_dir.is_dir():
+            shutil.rmtree(dst_dir)
+            dst_dir.mkdir(parents=True, exist_ok=True)
+        else:
+            dst_dir.mkdir(parents=True, exist_ok=False)
+    except FileExistsError:
+        # If the conversion has already been made, don't do anything
+        return
+    
+    # Some images are unlabelled and will be ignored
+    image_files = {Path(f).stem for f in os.listdir(image_dir) if f.endswith(('.jpg', '.jpeg', '.png'))}
+
+    file_base = 0
+    iterations = 0
+
+    for label in os.listdir(label_dir)[1:]:
+        if label.endswith('.txt'):
+            image_name = Path(label).stem
+
+            if image_name in image_files:
+                image_path = os.path.join(image_dir, image_name)
+                label_path = os.path.join(label_dir, label)
+                # Saves the broken down segments to the destination directory - no need to return
+                file_base = breakdown(image_path, label_path, dst_dir, file_base)
+
+        if iterations < 10:
+            print(file_base, end=" ")  # To make sure that it is progressing - a progress bar of sorts
+            iterations += 1
+        else:
+            print(file_base, end= '\r',)  # This doesn't work on the VS code terminal unfortunately 
+            iterations = 0
+
+def breakdown(image_path, label, dst_dir, file_base, segment=segments, targ_size = None):
+    with Image.open(image_path + ".jpg") as image:
+        img_size, take_size = image.size[0], None
+
+        # Can segment images by quantity (how many images you want) or by quantity (how large do you want the images)
+        if targ_size is None:
+            take_size = math.floor(img_size / segment)
+
+        else:
+            segment = math.floor(img_size / targ_size)
+            take_size = img_size / segment
+
+        # Particular to the Zebra label format
+        label_data = []
+        with open(label, 'r') as label_file:
+            for line in label_file.readlines():
+                if line[0] == '0':
+                    parsed = list(map(float, line.split()[1:]))
+                    entity_box = np.array([(parsed[i], parsed[i + 1]) for i in range(0, len(parsed), 2)])
+                    label_data.append(entity_box * img_size)
+
+
+        img = np.array(image)
+        for i in range(segment):
+            for j in range(segment):
+                box_coordinates = [i * take_size, j * take_size, min((i + 1) * take_size, len(img[0])), min((j + 1) * take_size, len(img))]
+                new_img = img[box_coordinates[1]: box_coordinates[3], box_coordinates[0]: box_coordinates[2]]
+                
+                # Some images do not load so have size zero, which causes crashes when loading them from the database during training
+                # because tensor conversions cannot take empty lists, For that reason we threshold at an arbitrarily low number here.
+                if new_img.size <= 32:
+                    continue
+
+                new_image = Image.fromarray(new_img)
+
+                crosswalk_intersection = False
+                for crosswalk_box in label_data:
+                    if check_box_intersection(box_coordinates, crosswalk_box):
+                        # 1 means an image that contains a crosswalk (a significant portion of it)
+                        crosswalk_intersection = True
+                    else:
+                        # 0 means background image (does not contain any significant portion of a crosswalk)
+                        pass
+                
+                if crosswalk_intersection: 
+                    with open(os.path.join(dst_dir, str(file_base)) + ".txt", 'w') as new_label_file:
+                        new_label_file.write("1")
+                        new_image.save(str(os.path.join(dst_dir, str(file_base))) + ".png")
+
+                # As the crosswalks are sparse - this improves the balance of positive to negative cases for training
+                else:
+                    if random.randint(0, 4) >= 1:
+                        with open(os.path.join(dst_dir, str(file_base)) + ".txt", 'w') as new_label_file:
+                            new_label_file.write("0")
+                            new_image.save(str(os.path.join(dst_dir, str(file_base))) + ".png")
+                    
+                # This ensures we don't overwrite previous segment files
+                file_base += 1
+
+    return file_base
+
+convert_database_to_segments("zebra_annotations/zebra_images", "zebra_annotations/txt_annotations", "zebra_annotations/classification_data")

SightLinks-Dev-main/DevelopmentTools/MobileNetV3.py

@@ -0,0 +1,117 @@
+# This is unquantised - for comparision
+
+import ClassUtils
+import torch
+import torch.nn as nn
+import torch.optim as optim
+from torch.utils.data import Subset
+import random
+from torchvision import models, transforms
+from torch.utils.data import DataLoader
+import time
+
+import matplotlib.pyplot as plt
+import numpy as np
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+print(device)
+
+# These define the model that will be trained
+num_classes = 2
+batch_size = 256
+epochs = 25
+learning_rate = 5e-4
+train_data_size = 25000
+saved_state_dict_path = "MobileNetV3_test.pth"
+
+model = models.mobilenet_v3_small(weights=models.MobileNet_V3_Small_Weights.DEFAULT)
+
+model.classifier[3] = nn.Linear(model.classifier[3].in_features, num_classes)
+model = model.to(device)
+
+dataset = ClassUtils.CrosswalkDataset("zebra_annotations/classification_data")
+
+train_loader = DataLoader(
+    Subset(dataset, random.sample(list(range(0, int(len(dataset) * 0.95))), train_data_size)),
+      batch_size=batch_size, shuffle=True)
+test_loader = DataLoader(
+    Subset(dataset, random.sample(list(range(int(len(dataset) * 0.95), len(dataset))), 12)),
+      batch_size=batch_size, shuffle=False)
+
+criterion = nn.BCELoss()
+optimizer = optim.Adam(model.parameters(), lr=learning_rate)
+
+
+# Generalised training function that uses a training-testing split defined in the training variables.
+# Works best with transfer learning as is shown in the 'MobileNetV3.py' function where it is defined - check it out.
+def train_model():
+    model.train()
+    start_time = time.time()
+    for epoch in range(epochs):
+        to_do = train_data_size
+        running_loss = 0.0
+        for inputs, labels in train_loader:
+            try:
+                inputs, labels = inputs.to(device), labels.to(device)
+            except:
+                continue
+            
+            optimizer.zero_grad()
+            outputs = torch.sigmoid(model(inputs))
+            loss = criterion(outputs, labels)
+            loss.backward()
+            optimizer.step()
+            
+            running_loss += loss.item()
+            to_do -= batch_size
+
+        print(f"Epoch {epoch+1}/{epochs}, Loss: {running_loss/len(train_loader)}, time {time.time()- start_time}")
+        start_time = time.time()
+
+# Do not use to actually evaluate performance, this is for quick checks - 'EvaluatePerformance.py' has actual quantified evaluation tools.
+def test_model():
+    model.eval()
+    correct = 0
+    total = 0
+    with torch.no_grad():
+        for inputs, labels in test_loader:
+            try:
+                inputs, labels = inputs.to(device), labels.to(device)
+            except:
+                continue
+            outputs = torch.sigmoid(model(inputs))
+
+            predicted = (outputs/100) > 0.5
+            for i in range(len(inputs)):
+                    plt.close()
+                    plt.imshow(torch.permute(inputs[i], (1, 2, 0)).cpu().detach().numpy())
+                    plt.title(f"prediction of {outputs[i].tolist()[0]:.3f}%, {100 * predicted[i].tolist()[0]:.3f}%,\nactual: {labels[i].tolist()}")
+                    plt.axis("off")
+                    plt.show()
+            
+            total += labels.size(0)
+            # print(predicted, labels)
+
+            for prediction, label in zip(predicted, labels):
+                correct += ((prediction[0]>50) == label[0])
+    
+    print(f"Accuracy: {100 * correct / total}%")
+
+
+
+train = True
+if __name__ == "__main__":
+    if train:
+        train_model()
+        torch.save(model.state_dict(), "mn3_vs55.pth")
+    else:
+        state_dictionairy = torch.load(saved_state_dict_path, weights_only=True)
+        print(type(state_dictionairy))
+        model.load_state_dict(state_dictionairy)
+
+    test_model()
+
+else:
+    state_dictionairy = torch.load(saved_state_dict_path, weights_only=True)
+    model.load_state_dict(state_dictionairy)
+    print(f"Module: [{__name__}] has been loaded")

SightLinks-Dev-main/DevelopmentTools/VGG_Custom.py

@@ -0,0 +1,45 @@
+import torch
+import torch.nn as nn
+import numpy as np
+from torchvision import transforms, datsets
+from torch.utils.data.sampler import SubsetRandomSampler
+
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+
+# This is a VGG-16 architecture (16 layers) - just for reference or if you want to build on top of it
+class VGG16(nn.Module):
+    def __init__(self, num_classes = 2):
+        # Has two classes, crosswalk or background.
+        self.layer1 = nn.Sequential(
+            nn.Conv2d(3, 342, kernel_size=3, stride=1, padding=1),
+            nn.BatchNorm2d(64),
+            nn.ReLU()
+        ),
+        self.layer2 = nn.Sequential(
+            nn.Conv2d(3, 342, kernel_size=3, stride=1, padding=1),
+            nn.BatchNorm2d(64),
+            nn.ReLU(),
+            nn.MaxPool2d(kernel_size=2)
+        ),
+        self.layer3 = nn.Sequential(
+            nn.Conv2d(3, 342, kernel_size=3, stride=1, padding=1),
+            nn.BatchNorm2d(64),
+            nn.ReLU()
+        ),
+        self.layer4 = nn.Sequential(
+            nn.Conv2d(3, 342, kernel_size=3, stride=1, padding=1),
+            nn.BatchNorm2d(64),
+            nn.ReLU()
+        ),
+        self.layer5 = nn.Sequential(
+            nn.Conv2d(3, 342, kernel_size=3, stride=1, padding=1),
+            nn.BatchNorm2d(64),
+            nn.ReLU()
+        ),
+        self.layer6 = nn.Sequential(
+            nn.Conv2d(3, 342, kernel_size=3, stride=1, padding=1),
+            nn.BatchNorm2d(64),
+            nn.ReLU()
+        )

SightLinks-Dev-main/DevelopmentTools/VGG_TL.py

@@ -0,0 +1,83 @@
+import torch
+import torch.nn as nn
+from torchvision import transforms, models
+from torch.utils.data import DataLoader, Subset
+from torchvision.datasets import ImageFolder
+from ClassUtils import CrosswalkDataset
+import numpy as np
+import random
+import time
+
+
+import warnings
+# Torchvision's models utils has a depreciation warning for the pretrained parameter in its instantiation but we don't use that
+warnings.filterwarnings(
+    action='ignore',
+    category=DeprecationWarning,
+    module=r'.*'
+)
+
+# In a later version, this could be moved to a configuration file.
+learning_rate = 4e-3
+epoch_num = 25
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+vgg16 = models.vgg16(weights = models.VGG16_Weights)
+# Modifies fully connected layer to output binary class predictions
+vgg16.classifier[6] = nn.Linear(vgg16.classifier[6].in_features, 2)
+
+# Freeze as you see fit depending on the application you want ot design
+# for param in vgg16.features.parameters():
+#     param.requires_grad = False
+# for param in vgg16.classifier[:6].parameters():
+#     param.requires_grad = False
+
+vgg16 = vgg16.to(device)
+loss_function = nn.BCELoss()
+
+# Prevents accidental loading of the whole training process in the background
+if __name__ == "__main__":
+    # Takes only the classifier layers, which have not been frozen
+    optimiser = torch.optim.Adam(params=
+                                filter(lambda p: p.requires_grad, vgg16.parameters()),
+                                lr=learning_rate)
+
+
+    training_dataset = CrosswalkDataset("zebra_annotations/classification_data")  
+    training_loader = DataLoader(Subset(training_dataset, random.sample(range(len(training_dataset)-1), 25000)), batch_size=128, shuffle=True)
+
+    for param in vgg16.features.parameters():
+        param.requires_grad = False
+
+
+    vgg16.train()
+    print(len(training_dataset))
+    for epoch in range(epoch_num):
+        running_loss = 0.0
+        start_time = time.time()
+        last_time = start_time
+        for images, gt in training_loader:
+            images, gt = images.to(device), gt.to(device)
+
+            classifications = torch.sigmoid(vgg16(images))
+            loss = loss_function(classifications, gt)
+            optimiser.zero_grad()
+            loss.backward()
+            optimiser.step()
+
+            batch_time = time.time()
+
+            running_loss += loss.item()
+
+            last_time = batch_time
+            print(",,, ---")
+
+        
+        print(f"\nEpoch {epoch + 1} of {epoch_num} has a per image loss of [{running_loss/len(training_loader):.4f}]")
+        print(f"{(last_time - start_time):.6f}")
+
+    # Includes the feature extraction layers
+    torch.save(vgg16.state_dict(), "VGG16_Full_State_Dict.pth")
+    # Only includes the classifier layer 
+    # - the 'head' whose weights you can use to overwrite if you don't want to store the whole state dict file
+    torch.save(vgg16.classifier[6].state_dict(), "vgg16_binary_classifier_onlyHead.pth")

SightLinks-Dev-main/DevelopmentTools/automate_annotation.py

@@ -0,0 +1,81 @@
+# A set of functions to automatically annotate images for training a classification model of crosswalks. Requires a paid API key for the images but tile locations are free to access.
+import overpy
+from geopy.geocoders import Nominatim
+import requests
+
+MAPBOX_API_KEY = None  
+crosswalk_images_folder = "crosswalk_images"
+
+# Returns a list of crosswalk locations based on a given location geo-code, such as "Bloomsbury, London, UK"
+# Geo-codes for a location may differ from the common name for a location, so check before putting in a location
+def get_cross_walk_location(location_geo_code, bounds=0.001):
+    over_api = overpy.Overpass()
+    geo_locator = Nominatim(user_agent="crosswalk_locator")
+    location = geo_locator.geocode(location_geo_code)
+
+    if not location:
+        raise ValueError("This location could not be geocoded - try an alternative name?.")
+
+    geo_bounding_box = (location.latitude - bounds, location.longitude - bounds,
+                        location.latitude + bounds, location.longitude + bounds)  # South, West, North, East
+
+    query = f"""
+    [out:json][timeout:25];
+    (
+      node["highway"="crossing"]({geo_bounding_box[0]},{geo_bounding_box[1]},{geo_bounding_box[2]},{geo_bounding_box[3]});
+    );
+    out body;
+    """  # Should return all crosswalks within {bounds} longitude/ latitude points, in a square shape
+
+    result = over_api.query(query)
+
+    crosswalks = []
+    for node in result.nodes:
+        crosswalks.append((node.lat, node.lon))
+    # Now we have a list of all the (latitude, longitude) pairs in that area,
+    # and can use it access the images of those areas for annotation
+
+    return crosswalks
+
+
+# Gets the aerial image associated with a particular longitude and latitude.
+# Uses the mapbox api, which is cheaper alternative to google maps static tiles API, but has lower res. images.
+def get_mapbox_aerial_image(location, output_file_name, zoom=18, size="256x256", style="satellite-v9",
+                            api_key=MAPBOX_API_KEY):
+    longitude, latitude = location  # We split here, so it's easier to map this function onto a set of (lon, lat) coords
+    base_url = f"https://api.mapbox.com/styles/v1/mapbox/{style}/static"
+
+    coords = f"{latitude},{longitude},{zoom}"
+    url = f"{base_url}/{coords}/{size}?access_token={api_key}"
+
+    response = requests.get(url)
+
+    if response.status_code == 200:
+        with open(output_file_name, "wb") as file:
+            file.write(response.content)
+        print(f"Retrieved aerial image and saved it as {output_file_name}")
+
+    else:
+        print("Error, Error, Error!", response.status_code, response.text)
+
+
+# Saves a list of crosswalk locations for the given geo-code to a folder, and returns the names of the saved files
+# This function executes the setup required for training a classification model requiring 256x256 images as ours did
+# If you want to change the model you are training, adjsut the get_mapbox_aerial_image function
+def get_crosswalk_images(geo_code, folder=crosswalk_images_folder, file_extension="png"):
+    crosswalk_image_files = []
+    crosswalk_set = get_cross_walk_location(geo_code, bounds=0.1)
+    for crosswalk in crosswalk_set:
+        formatted_crosswalk = list(map(float, crosswalk))
+        stored_file_name = str(formatted_crosswalk)
+        try:
+            get_mapbox_aerial_image(formatted_crosswalk,
+                                    f"{folder}/{formatted_crosswalk}.{file_extension}")
+            crosswalk_image_files.append(stored_file_name)
+        except Exception as e:
+            print(e)
+
+    return crosswalk_image_files
+
+
+#print(len(get_cross_walk_location("Bloomsbury, London, UK")))

SightLinks-Dev-main/DevelopmentTools/experimental_resources/FeatureExtractionMethods.py

@@ -0,0 +1,338 @@
+# Feature Extraction Experimental Methods - Our method actually does not signficiantly benefit from 
+# common preprocessing techniques, so we created a set of more uncommon/ niche methods as well as some
+#  other untested methods to experiment with which features best represent the crosswalk features.
+
+# This hopefully should be of great use to anyone trying to train their own specialised classifier using 
+# our pipeline - hopefully one of these works for you!
+
+# All these methods work on a numpy array - if you want to use them convert your image to numpy first
+# Most of the methods assume the channels of the image are at the end
+
+import numpy as np
+from scipy.signal import convolve2d
+import skimage.feature as skf
+from cv2 import NMSBoxes
+
+from scipy.ndimage import maximum_filter
+from collections import deque
+
+
+
+# Was required for keeping several of the methods optimised.
+def matrixConvolution(image, kernel, padding=True):
+    convolvedImage = np.zeros_like(image)
+
+    # Again, this assumes the channels are placed at the end of the image
+    for c in range(image.shape[2]):
+        convolvedImage[..., c] = convolve2d(image[..., c], kernel, mode='same', boundary='wrap')
+
+    return convolvedImage
+
+
+# Different more traditional image processing methods, these may not be of particular use in our project, 
+# but could be useful for other developers using this pipeline.
+class ImageProcessingFeatures:
+    def __init__(self):
+        self.gaussianKernel = self.generateGaussianKernel(5, 1)
+
+    # Converts an image to grayscale - but using binary thresholding of the averaged colour channels
+    # Image in format [n, m, k] where k is the channels and n, m are the dimensions
+    def binaryThresholding(self, image, threshold):
+
+        averaged = np.mean(image, axis=2)
+        thresholded = averaged > threshold
+
+        return thresholded
+    
+    # A bit unintuitively named, but this takes an already grayscale image and applies binary thresholding to it
+    def grayscaleBinaryThresholding(self, grayscaleImage, threshold):
+        return grayscaleImage > threshold
+
+    
+    # Converts an image to grayscale - it has several possible schema it can use but defaults to the lightness method
+    def grayscaleConversion(self, image, schema="lightness"):
+        # https://tannerhelland.com/2011/10/01/grayscale-image-algorithm-vb6.html
+        grayImage = np.zeros(np.shape(image)[:-1]) 
+
+        if schema == "average":
+            grayImage = np.mean(image, axis=2)
+
+        if schema == "lightness":
+            # (max(R, G, B) + min(R, G, B)) / 2 --> Sometimes called desaturation
+            grayImage = (np.max(image, axis=2) + np.min(image, axis=2)) / 2
+
+        if schema == "luma":
+            # I assume RGB colour ordering in the image here -- feel free to overwrite in your implementation
+            # (Red * 0.2126 + Green * 0.7152 + Blue * 0.0722)
+            colourWeighting = np.array([0.2126, 0.7152, 0.0722])
+            grayImage = np.dot(image[..., :3], colourWeighting)
+
+        if schema == "decomposition":
+            # This is a maximum decomposition - minumum decomposition can be implemented by just switching our for min...
+            grayImage = np.max(image, axis=2)
+
+        return grayImage
+
+
+    # Sharpens the image by convolving a preset laplace kernel with it
+    # My chosen kernel is a pretty standard one that takes into consideration the corners - the 8 adjacencies one
+    # ​[-1, -1, -1], [-1, 8, -1], [-1, -1, -1]
+    def laplaceTransform(self, image):
+        laplaceKernel=[[-1, -1, -1], [-1, 8, -1], [-1, -1, -1]]
+        convolvedImage = None
+
+        # RGB image
+        if len(np.shape(image)) == 3:
+            convolvedImage = matrixConvolution(image, laplaceKernel)
+
+        # GrayScale image
+        if len(np.shape(image)) == 2:
+            convolvedImage = convolve2d(image, laplaceKernel, mode='same', boundary='wrap')
+
+        return convolvedImage
+
+    # Assumes Grayscale images
+    # https://en.wikipedia.org/wiki/Sobel_operator - naive implementation by me, can probably be massively improved
+    # Basically takes the vertical and horizontal gradients using convolution with a sobel operator and combines them.
+    def sobelConvolution(self, image):
+        # Gx
+        verticalKernel = [[-1, 0, 1],
+                            [-2, 0, 2],
+                            [-1, 0, 1]]
+
+        # Gy
+        horizontalKernel = [[-1, -2, -1],
+                        [0, 0, 0],
+                        [1, 2, 1]] 
+
+        horizontalGrad = convolve2d(image, horizontalKernel, mode='same', boundary='wrap')
+        verticalGrad = convolve2d(image, verticalKernel, mode='same', boundary='wrap')
+
+        sobelConvolvedImage = np.sqrt(np.square(horizontalGrad) + np.square(verticalGrad))
+
+        # You could go further to find the gradient direction by calculating angle with some trigonometry - atan2(Gy, Gx)
+        return sobelConvolvedImage
+    
+    # Part of the calculations for canny edge detection, a tri-threshold operations. Requires greyscale images.
+    # Classifies into strong edges and weak edges based on pixel intensity
+    def doubleFiltering(self, image, weakThreshold=75, strongThreshold=200):
+        weak, strong = 125, 255
+
+        strong_edges = image >= strongThreshold
+        weak_edges = (image >= weakThreshold) & (image < strongThreshold)
+        
+        result = np.zeros_like(image, dtype=np.uint8)
+        result[strong_edges] = strong
+        result[weak_edges] = weak
+        
+        return result, strong, weak
+    
+    # Technical name: edge tracking by hysteresis. This is an 'optimised' dequeue approach (to the best of my ability)
+    def followEdges(self, weakEdges, strongEdges):
+        h, w = weakEdges.shape
+        directions = [(-1, -1), (-1, 0), (0, -1), (1, -1), (-1, 1), (0, 0), (0, 1), (1, 0), (1, 1)]
+        finalEdges = (strongEdges.copy() > 0) # must include strong edges, it's guaranteed, and we want a binary output.
+        edgeQueue = deque(np.argwhere(strongEdges == 1)) 
+        
+        while edgeQueue:
+            y, x = edgeQueue.popleft()  
+            for dy, dx in directions:
+                ny, nx = y + dy, x + dx
+                
+                # Expand the branch if they're connected to a weak edge, else kill this branch
+                if 0 <= ny < h and 0 <= nx < w and weakEdges[ny, nx] > 0:
+                    weakEdges[ny, nx] = 0
+                    finalEdges[ny, nx] = 1  
+                    edgeQueue.append((ny, nx))
+        
+        return finalEdges
+
+
+    # Much better than sobel in terms of accuracy and removing false edges produced due to noise, but slower.
+    # https://en.wikipedia.org/wiki/Canny_edge_detector + a lot of chatGPT prompts
+    def cannyEdgeDetection(self, image):
+        # Ensures that the image is in greyscale so we don't have any issues.
+        if (np.shape(image) == 3):
+            image = ImageProcessingFeatures.grayscaleConversion(image, schema="average")
+
+        # Takes a greyscale image and returns a set of strong and weak edges (see wikipedia page)
+        blurred = convolve2d(image, self.gaussianKernel, mode='same', boundary='wrap')
+        gradientMagnitude = self.sobelConvolution(blurred)
+        maxFiltered = maximum_filter(gradientMagnitude, size=3, mode='constant')
+        suppressed = np.where(gradientMagnitude==maxFiltered, gradientMagnitude, 0)
+        thresholded, strong, weak = self.doubleFiltering(suppressed)
+
+        final_edges = self.followEdges(strong, weak)
+        # We expand the definition of strong edges to include weak edges that are adjacent to strong edges
+
+        return final_edges
+
+    
+    # Convolve this with the image to produce a Gaussian Blur effect 
+    # In a real application you should precompute this, it's really unoptimised
+    def generateGaussianKernel(self, size, sigma):
+        center = size // 2
+        kernel = np.zeros((size, size))
+
+        for i in range(size):
+            for j in range(size):
+                sides = np.sqrt((i - center) ** 2 + (j - center) ** 2)
+                kernel[i, j] = np.exp(-(sides ** 2) / (2 * sigma ** 2))  # According to the formula
+
+        return kernel / np.sum(kernel)
+
+
+    # This should be applied to a grayscale image! 
+    def differenceOfGaussians(self, image, size=5, sigmaOne=1, sigmaTwo=1.5, brightFeatureFocus=True):
+        # This by default uses a 5x5 kernel with sigmaOne = 1 and sigmaTwo = 2. These are not finetuned values but follow the general
+        # principle that sigma should not be greater than approx. 3*dims. Be careful to consider the relationship between the dimension
+        # and sigma values (and between the two sigmas themselves) to preserve the gaussian property!
+        kernelOne, kernelTwo = self.generateGaussianKernel(size, sigmaOne), self.generateGaussianKernel(size, sigmaTwo)
+
+        primaryImage = convolve2d(image, kernelOne, mode='same', boundary='wrap')
+        backgroundImage = convolve2d(image, kernelTwo, mode='same', boundary='wrap')
+
+        diffOfGaussians = primaryImage - backgroundImage
+        
+        # Enhances bright feature edges (our focus in crosswalks), but this is a thing that might vary based on your focus. Can be disabled!
+        if brightFeatureFocus:
+            diffOfGaussians = diffOfGaussians[diffOfGaussians > 0]
+
+        return diffOfGaussians
+
+
+# Different feature extraction methods that attempt to quantify the complexity in an image. 
+# Generally seperated into local complexity per region, and complexity of an image as a whole.
+# Intended to potentially detect the occlusion in an image, e.g. more treees --> more complexity hopefully.
+class ComplexityFeatures:
+    def __init__(self):
+        pass
+
+    # Definition: Mathematical measure of how complex an image or pattern is
+    # In our case (Grayscale image), it measures how much the detail in an image changes with the scale it is perceived at.
+    # The particular method we will the basic box-counting method
+    # Our definition of a useful feature map is a binary-thresholded line detection method using difference of Gaussians
+    def fractalDimension(self, image, minimumBoxSize=2, imageStructureThreshold=0.9):
+
+        if len(image.shape) > 2:
+            image = ImageProcessingFeatures.grayscaleConversion(image, schema="average")
+
+        image = ImageProcessingFeatures.sobelConvolution(image) # Could also go for DoG or Laplace too - most edge detectors work
+        image = np.int_(255 * (image - np.min(image)) / (np.max(image) - np.min(image)))
+        image = ImageProcessingFeatures.grayscaleBinaryThresholding(image, imageStructureThreshold) 
+
+        N, M = image.shape
+
+        startingBoxSize = min(N, M) // 4
+        boxSizes = [s for s in range(minimumBoxSize, startingBoxSize + 1, 2) if min(N, M) % s == 0]
+        logSizes, logCounts = [], []
+
+        for boxSize in boxSizes:
+            numBoxes = 0
+
+            for i in range(0, N, boxSize):
+                for j in range(0, M, boxSize):
+                    if np.any(image[i:i+boxSize, j:j+boxSize]):  # Check if box contains a 1 (part of the image structure)
+                        numBoxes += 1
+                        
+            if numBoxes > 0:
+                logSizes.append(np.log(1.0 / boxSizes)) 
+                logCounts.append(np.log(numBoxes))
+        
+        slope, _ = np.polyfit(logSizes, logCounts, 1)
+        # Finds the relationship of the sizes and the image structure contained as the image scale decreases
+
+        return slope
+
+    # Come back to
+    def waveletBasedFractalTransform(self, image):
+        pass
+
+
+# A set of feature extraction methods inspired by texture analysis methods
+# just a wild throw in the dark, not sure how useful these could be as a feature
+# Many of these should be applied in sliding window approaches or in regions, or to the whole image if you have a feature vector
+class TextureFeatures:
+    def __init__(self):
+        pass
+
+    def lbpCompare(self, threshold, value):
+        return 0 if value < threshold else 1
+
+    # A matter of personal preference, but this is a spiral concatenation for the local binary pattern signature generation
+    # You could do row by row, but this is my preferred method. Feel free to overwrite, it shouldn't make a difference as long as you're consistent
+    def spiral_concatenation(self, vals, dims):
+        summed = []
+        dirs = [(0, 1), (1, 0), (0, -1), (-1, 0)] 
+        cur_dir = 0  
+        cur_x, cur_y = 0, 0  
+        bounds = dims  
+
+        while bounds > 0:
+            for i in range(bounds):
+                summed.append(str(vals[cur_y * dims + cur_x])) 
+                cur_y, cur_x = cur_y + dirs[cur_dir][0], cur_x + dirs[cur_dir][1]  
+            cur_dir = (cur_dir + 1) % 4 
+            
+            bounds -= 1
+
+            if bounds > 0:
+                for i in range(bounds):
+                    summed.append(str(vals[cur_y * dims + cur_x]))
+                    cur_y, cur_x = cur_y + dirs[cur_dir][0], cur_x + dirs[cur_dir][1]
+                cur_dir = (cur_dir + 1) % 4
+                bounds -= 1
+
+        return "".join(summed)
+
+    # Basically captures the local binary changes in texture in an image - a potentially useful feature for our crosswalk detector that
+    # works on a very similar principle with the local regions of interest found by the first.
+    # It generates a signature for each local region that can be used to compare them quite easily in applications like texture analysis.
+    # CAREFUL - THIS IS A STRING FEATURE, NOT A NUMERICAL ONE
+    def localBinaryPattern(self, image, dims):
+        imgWidth, imgLength = len(image[0]), len(image)
+        edge = dims // 2 
+        lbpList = []
+        for row in range(edge, imgLength - edge):
+            for pixel in range(edge, imgWidth - edge):
+                neighborhood = image[row - edge: row + edge + 1, pixel - edge: pixel + edge + 1]
+                
+                centralPixel = image[row][pixel]
+                binaryVals = [self.lbpCompare(centralPixel, val) for val in neighborhood.flatten()]
+
+                lbpSignature = self.spiral_concatenation(binaryVals, dims)
+                lbpList.append(lbpSignature)
+
+        return lbpList
+
+    # https://medium.com/@girishajmera/feature-extraction-of-images-using-glcm-gray-level-cooccurrence-matrix-e4bda8729498
+    # Link above explains the function quite well and succinctly. This takes in both coloured (n, m, k) and grayscale arrays (n, m).
+    # Captures the spatial relationships between neighbouring gray levels/ Intensities
+    def grayLevelCoOccurenceMatrix(self, image, pixelOffset=5, preserveMatrix=False):
+        if not (np.shape(image) == 2):
+            # P.S - this assumes that RGB and [n, m, k] formats are followed. Can throw errors otherwise.
+            image = ImageProcessingFeatures.grayscaleConversion(image, schema="average")
+
+        transformedArray = np.int_(255 * (image - np.min(image)) / (np.max(image) - np.min(image)))
+
+        distances = [pixelOffset]
+        angles = [0, np.pi/4, np.pi/2, 3*np.pi/4]
+
+        # A bit of a cop out, but this just does it all for us. The metrics you choose to extract from this depend.
+        glcm = skf.graycomatrix(image, distances=distances, angles=angles, levels=256, symmetric=True, normed=True)
+
+        # This is a set of features that I personally thought might be useful for the crosswalks - but there are many other extractable features
+        contrast = skf.graycoprops(glcm, 'contrast')
+        energy = skf.graycoprops(glcm, 'energy')
+        homogeneity = skf.graycoprops(glcm, 'homogeneity')
+        correlation = skf.graycoprops(glcm, 'correlation')
+
+        # Single value metrics for the image
+        metrics = (np.mean(contrast.flatten()), np.mean(energy.flatten()), np.mean(homogeneity.flatten()), np.mean(correlation.flatten()))
+
+        if preserveMatrix:
+            return metrics, (contrast, energy, homogeneity, correlation)
+        
+        return metrics
+
+

SightLinks-Dev-main/DevelopmentTools/experimental_resources/SelfSupervisedFeatureExtraction.py

@@ -0,0 +1,137 @@
+# Besides the usual feature extraction/ preprocessing methods, or statistical ones - we can also employ ML models to automatically
+# build feature extraction layers, using self-supervised learning for example.
+
+# This is an implementation of simCLR
+
+import torch
+import torch.nn as nn
+import torchvision.transforms as transforms
+import torchvision.models as models
+import torchvision.datasets as datasets
+from torch.utils.data import DataLoader
+import numpy as np
+import random
+
+# Applies random data augmentation techniques to the image, following the "simCLR" methodology
+# https://arxiv.org/pdf/2002.05709 
+class SimCLRAugmentationTransform:
+    def __init__(self):
+        # Following the default augmentations listed out in their paper's Data Augmentation Details
+        # If you want to add in more transforms, remember to use transforms.RandomApply
+        self.transform = transforms.Compose([
+            transforms.RandomResizedCrop(size=224,),
+            transforms.RandomHorizontalFlip(),
+            transforms.RandomApply(
+                [transforms.ColorJitter(brightness=0.8, contrast=0.8, saturation=0.8, hue=0.25)],
+                  p=0.8),
+            transforms.RandomGrayscale(p=0.2),
+            transforms.GaussianBlur(kernel_size=3),
+            transforms.ToTensor(),
+            transforms.Normalize(mean=(0.5,), std=(0.5,), inplace=False),
+        ])
+
+    # We want to produce two contrasting views of the image
+    def __call__(self, x):
+        return self.transform(x), self.transform(x)
+    
+
+# See: https://medium.com/data-science/nt-xent-normalized-temperature-scaled-cross-entropy-loss-explained-and-implemented-in-pytorch-cc081f69848
+# for a pretty in-depth and clear explanation of how it works. TLDR: Normalise, temperature scale, Cross-Entropy loss.
+class NTXentLoss(nn.Module):
+    def __init__(self, temperature = 0.5):
+        super().__init__()
+        self.temperature = temperature
+        self.cosSimilarity = nn.CosineSimilarity(dim=-1)
+        self.CEL = nn.CrossEntropyLoss(reduction="mean")
+
+    def forward(self, z_i, z_j, temperature=None):
+        if temperature is None:
+            temperature = self.temperature
+        x = torch.cat([z_i, z_j], dim=0)
+        xcs = self.cosSimilarity(x[None,:,:], x[:,None,:])
+
+        # Naturally, each embedding will have a 1-1 similarity with itself, so will contribute nothing useful,
+        # so we zero these contributions.
+        xcs[torch.eye(xcs.shape[0], dtype=torch.bool)] = -float('inf')
+
+        # Math works out on these I think, because they're in contigious pairs as an input.
+        target = torch.arange(len(z_i) * 2)
+        target[0::2] += 1
+        target[1::2] -= 1
+
+        return self.CEL(xcs/ temperature, target)
+
+
+# A SimCLR model - to use just override the getModel function to whatever model you want to use, 
+# and post-training remove the projection layer and add in your classifier head (overwrite model's fully connected layer)
+class SimCLR(nn.Module):
+    def __init__(self, encoderModel="MobileNet_V3", outDim=128):
+        super().__init__()
+        self.encoder = self.getModel(encoderModel)
+        self.projectionHead = nn.Sequential(
+            nn.Linear(1000, 256),
+            nn.LeakyReLU(),
+            nn.Linear(256, outDim)
+        )
+
+
+    def getModel(self, modelName):
+        modelDictionairy = {
+            "ResNet-50": models.resnet50(progress=True),
+            "ResNet-18": models.resnet18(progress=True),
+            "MobileNet_V3": models.mobilenet_v3_small(progress=True),
+        }
+        # This should only be evaluated upon the getting of a particular model (calling of a particular index)
+        # - shouldn't have to install all of these until used unless there's a bug
+        model = modelDictionairy[modelName]
+        model.fc = nn.Identity()  # This should be overwritten in your downstream application
+
+        return model
+    
+    def forward(self, x):
+        features = self.encoder(x)
+        featureVector = self.projectionHead(features)
+        return featureVector
+
+
+# Since this is completely dependent on what dataset you use, this serves PURELY as a tutorial for how
+# to put your own code in
+
+if __name__ == "__main__":
+    batchSize = 128
+    transform = SimCLRAugmentationTransform()
+    dataset = datasets.CIFAR10(root="./data", train=True, transform=transform, download=True)
+    dataloader = DataLoader(dataset, batch_size=batchSize, shuffle=True)
+
+    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+    model = SimCLR().to(device)
+    optimiser = torch.optim.Adam(model.parameters(), lr=2e-3)
+    lossFunction = NTXentLoss(temperature=0.5)
+
+    epochs = 2
+    lossTracker = []
+    print(len(dataloader))
+    for epoch in range(epochs):
+        eraLoss = 0.0
+        model.train()
+
+        for (x_i, x_j), _ in dataloader:
+            print(".")
+            x_i, x_j = x_i.to(device), x_j.to(device)
+
+            optimiser.zero_grad()
+
+            z_i, z_j = model(x_i), model(x_j)
+
+            loss = lossFunction(z_i, z_j)
+            loss.backward()
+            optimiser.step()
+
+            eraLoss += loss.item()
+        print("\n")
+
+        avgLoss = eraLoss / len(dataloader)
+        lossTracker.append(avgLoss)
+        print(f"In epoch {epoch+1} of {epoch+1}, there was a loss of {avgLoss:.4f}")
+
+    print("Completed Training!!!")

SightLinks-Dev-main/DevelopmentTools/experimental_resources/quantisedMobileNet.py

@@ -0,0 +1,352 @@
+# BETA VERSION - NEEDS FURTHER DEVELOPMENT
+
+
+# Read these to catch up on what is (trying to at least) being done here
+# https://pytorch.org/tutorials/advanced/static_quantization_tutorial.html
+# https://pytorch.org/docs/stable/quantization.html#model-preparation-for-eager-mode-static-quantization
+
+# Torch implementation of these models - mine is heavily based on these with some minor adjustments
+#
+# I've added squeeze and excitation layers to the MobileNetV2, a feature of MobileNetV3, but I did not put in 
+# NAS (unnecessary since we're not optimising for mobile) or hardswish (because I prefer ReLU/ think it is better)
+# https://github.com/pytorch/vision/blob/main/torchvision/models/mobilenetv3.py#L117
+# https://github.com/pytorch/vision/blob/11bf27e37190b320216c349e39b085fb33aefed1/torchvision/models/mobilenetv3.py#L56
+
+# This is an adapted version of MobileNet, somewhere between versions 2/3, as some features of 3 were not required. There are
+# also some additions for our particular use case from miscallaneous sources
+
+from torchvision import transforms
+import torch
+from torch import nn, Tensor
+from torch.nn import functional as F
+from torch.utils.data import DataLoader, Subset
+import ClassUtils
+
+from torch.ao.quantization import QuantStub, DeQuantStub
+from torchvision.models.mobilenetv2 import _make_divisible
+
+import time
+import random
+import os
+import matplotlib.pyplot as plt
+
+# Squeeze: summarising global context by pooling feature maps into a single value
+# Excitation: Learning attention weights for each channel to prioritise the most relevant ones
+class SqueezeExcitation(nn.Module):
+    def __init__(self, input_channels:int, squeeze_factor: int = 4):
+        super().__init__()
+        # If channels are a multiple of 8, they're optimised by the hardware
+        squeeze_channels = _make_divisible(input_channels // squeeze_factor, 8)
+        self.squeeze = nn.Conv2d(input_channels, squeeze_channels, 1)
+        self.relu = nn.ReLU(inplace=True)
+        self.unsqueeze = nn.Conv2d(squeeze_channels, input_channels, 1)
+        self.quant = nn.quantized.FloatFunctional()
+    
+    # Scale returns the feature attention map, how much attention should be payed to each input layer, in range [0, 1]
+    # Inplace is used to save memory on operations - it might not be necessary in our case since we aren't using edge devices
+    def _scale(self, input: Tensor, inplace=bool) -> Tensor:
+        # Squeeze
+        scale = F.adaptive_avg_pool2d(input, 1)
+        scale = self.squeeze(scale)
+        # Excite
+        scale = self.relu(scale)
+        scale = self.unsqueeze(scale)
+        return F.hardsigmoid(scale, inplace=inplace)
+    
+    def forward(self, input: Tensor) -> Tensor:
+        # print(self._scale(input, True))
+        # print(input)
+        return self.quant.mul(self._scale(input, True), input)
+
+
+# The basic building block of our convolutional neural network
+# - qconfig should automatically insert fakeQuantisation operations during training, so there is no need to manually place them now
+class ConvBNReLu(nn.Sequential):
+    def __init__(self, in_planes, out_planes, kernel_size=3, stride=1, groups=1):
+        padding = (kernel_size - 1) // 2
+        super().__init__(
+            nn.Conv2d(in_planes, out_planes, kernel_size, stride, padding, groups=groups, bias=False),
+            # No point applying a bias (constant addative term) if the next layer is a batch normalisation layer
+            nn.BatchNorm2d(out_planes, momentum=0.1),
+            nn.ReLU(inplace=True)
+        )
+ 
+
+# Like typical residual blocks but uses inverse narrow->wide->narrow, with Depth-wise convolutions instead of normal,
+# to reduce the number of parameters required compared to the usual residual blocks
+class InvertedResidual(nn.Module):
+    def __init__(self, inpt, oupt, stride, expnd_ratio, kernel_size=3, se_layer=None):
+        super().__init__()
+
+        self.stride = stride
+        assert stride in [1, 2]
+
+        intermediate_channels = int(round(inpt * expnd_ratio))
+        # If the stride != 1, downsampling occurs so cannot be true. 
+        self.use_residual = (stride==1) and (inpt==oupt)
+        # Squeeze and excitation layer - applied after the dw and pw convolutions, but before the residual
+        self.se_layer = se_layer if se_layer else None
+
+        layers = []
+
+        if expnd_ratio != 1:
+            # Pointwise convolution to increase the channels
+            layers.append(ConvBNReLu(inpt, intermediate_channels, kernel_size=1))
+
+        layers.extend([
+            # Depthwise convolution - each channel is convoled on an independent basis 
+            ConvBNReLu(intermediate_channels, intermediate_channels, stride=stride, groups=intermediate_channels),
+            # point-wise convolution - linear combination to reduce layers back to the expected number
+            nn.Conv2d(intermediate_channels, oupt, 1, 1, 0, bias=False),
+            nn.BatchNorm2d(oupt, momentum=0.25)
+        ])
+        self.conv = nn.Sequential(*layers)
+
+    def forward(self, x):
+        outpt = self.conv(x)
+        if self.se_layer is not None:
+            outpt = self.se_layer(outpt)
+
+        if self.use_residual:
+            return x + outpt
+        else:
+            return outpt
+        
+
+# Same as the inverted residual, but replaces addition with a quantizable friendly operation 
+class QuantizableInvertedResidual(InvertedResidual):
+    def __init__(self, inpt, outpt, stride, expnd_ratio, se_layer=None):
+        super().__init__(inpt, outpt, stride, expnd_ratio, se_layer=se_layer)
+        self.skip_add = nn.quantized.FloatFunctional()
+
+    # Overwrites the forwarding to use a quantizable friendly version of the addition
+    def forward(self, x):
+        outpt = self.conv(x)
+        if self.se_layer is not None:
+            outpt = self.se_layer(outpt)
+
+        if self.use_residual:
+            return self.skip_add.add(x, outpt)
+        else:
+            return outpt
+
+# The MobileNetV2 Architecture + some features from V3 (squeeze and excitation) but I didn't add NAS since we aren't running this on mobile
+# And I prefer ReLU over hardswish
+class MobileNetV2_5(nn.Module):
+    def __init__(self, class_num=2, width_mult=1.0, round_nearest=8):
+        super().__init__()
+
+        layers = []
+
+        input_channel = 32
+        last_channel = 1280
+
+        # Just straight up copying this from the torchvision implementation
+        self.residual_params =  [
+                # expnd_ratio, outpt_channels, num_blocks, stride
+                [1, 16, 1, 1],
+                [6, 24, 2, 2],
+                [6, 32, 3, 2],
+                [6, 64, 4, 2],
+                [6, 96, 3, 1],
+                [6, 160, 3, 2],
+                [6, 320, 1, 1],
+            ]
+        
+        first_conv_output_channels = _make_divisible(self.residual_params[0][1] *width_mult, round_nearest)
+        layers.append(
+            ConvBNReLu(3,
+                first_conv_output_channels,
+                kernel_size=3,
+                stride=2,
+                )
+        )
+        prev_input_channels = first_conv_output_channels
+
+        # Main body of feature extraction
+        for expnd, oupt_c, num_blocks, strd in self.residual_params:
+            # output channels must be a multiple of 8 for hardware optimisation
+            output_channel = _make_divisible(oupt_c * width_mult, round_nearest)
+            
+            for i in range(num_blocks):
+                stride = strd if i == 0 else 1
+                se_layer = SqueezeExcitation(oupt_c) if i == 0 else None
+                layers.append(QuantizableInvertedResidual(prev_input_channels, output_channel, stride, expnd_ratio=expnd, se_layer=se_layer))
+                prev_input_channels = output_channel
+            
+        
+        self.last_channel = _make_divisible(last_channel * max(width_mult, 1.0), round_nearest)
+
+        # We could put this in the classifier, but I want that to be lightweight so that we could do transfer learning only on the head and 
+        # the feature extraction part of the model.
+        layers.append(
+            ConvBNReLu(prev_input_channels, self.last_channel, kernel_size=1)
+        )
+
+        self.feature_extraction = nn.Sequential(*layers)
+        self.avg_pooling = nn.AdaptiveAvgPool2d(1)
+        self.classifier = nn.Sequential(
+            nn.Dropout(0.125),
+            nn.Linear(last_channel, class_num)
+        )
+
+        # This bit is also just straight up copied from torch's implementation - I'm not touching it in case it gets messed up
+        # weight initialization
+        for m in self.modules():
+            if isinstance(m, nn.Conv2d):
+                nn.init.kaiming_normal_(m.weight, mode='fan_out')
+                if m.bias is not None:
+                    nn.init.zeros_(m.bias)
+            elif isinstance(m, nn.BatchNorm2d):
+                nn.init.ones_(m.weight)
+                nn.init.zeros_(m.bias)
+            elif isinstance(m, nn.Linear):
+                nn.init.normal_(m.weight, 0, 0.01)
+                nn.init.zeros_(m.bias)
+
+    def forward(self, x: Tensor) -> Tensor:
+        x = self.feature_extraction(x)
+        x = self.avg_pooling(x)
+        x = torch.flatten(x, 1)
+        print("eyo")
+        x = self.classifier(x)
+
+        return x
+
+class QuantizableMobileNetV2_5(MobileNetV2_5):
+    def __init__(self, class_num=2, width_mult=1.0, round_nearest=8):
+        super().__init__(class_num=class_num, width_mult=width_mult, round_nearest=round_nearest)
+        self.quant = QuantStub()
+        self.dequant = DeQuantStub()
+
+    def _forward_impl(self, x: Tensor) -> Tensor:
+        x = self.feature_extraction(x)
+
+        # This was for debugging errors in shape of feature maps as they pass through - not deleting incase useful later
+        # for idx, layer in enumerate(self.feature_extraction):
+        #     x = layer(x)
+        #     print(f"Feature extraction layer {idx}, output shape: {x.shape}")
+    
+        x = self.avg_pooling(x)
+        x = torch.flatten(x, 1)
+        x = self.classifier(x)
+
+        return x
+
+    def forward(self, x: Tensor) -> Tensor:
+        x = self.quant(x)
+        x = self._forward_impl(x)
+        x= self.dequant(x)
+        return x
+
+
+def train_single_epoch(model, loss_fnc, optimiser, data_loader, device):
+    model.train()
+    running_loss = 0
+    running_time = 0.0
+    for images, labels in data_loader:
+        start_time = time.time()
+        print(".", end=" ")
+        
+        images, labels = images.to(device), labels.to(device)
+        preds = model(images)
+        loss = loss_fnc(preds, labels)
+        loss.backward()
+        optimiser.step()
+
+        running_loss += loss.item()
+        running_time += time.time() - start_time
+        
+        print(f"{(time.time() - start_time):.2f}, {(running_time):.2f}", end=" ")
+
+    print(f"loss of {running_loss}")
+    return
+
+def print_size_of_model(model):
+    torch.save(model.state_dict(), "temp.p")
+    print('Size (MB):', os.path.getsize("temp.p")/1e6)
+    os.remove('temp.p')
+
+def adjust_quantisation_engine():
+    # Adjust according to what your device supports
+    print(torch.backends.quantized.supported_engines)
+    torch.backends.quantized.engine = 'qnnpack'
+
+def train_model(model, dataloader, loss_function, optimiser, epoch_number=25, const_save=False, save=True):
+    for epoch in range(epoch_number):
+        print("IT IS EPOCH", epoch)
+        train_single_epoch(model, loss_function, optimiser, dataloader, torch.device('cpu'))
+
+        # Gradually freezes the unrequired observer parameters for quantisation and batch normalisation after a few epochs
+        if epoch > 3:
+            # Freeze quantizer parameters
+            model.apply(torch.ao.quantization.disable_observer)
+        if epoch > 2:
+            # Freeze batch norm mean and variance estimates
+            model.apply(torch.nn.intrinsic.qat.freeze_bn_stats)
+
+        if const_save:
+            quantized_model = torch.ao.quantization.convert(model.eval(), inplace=False)
+            quantized_model.eval()
+
+            # Saving each intermediary model since they're so small, and this lets load up any of them for performace difference examples later
+            torch.save(quantized_model.state_dict(), "quantStateDict"+str(epoch+1)+".pth")
+
+            print(f"the above was Epoch {epoch+1} of {epoch_number} \nThe model has a size of", end=" ")
+            print_size_of_model(quantized_model)
+
+        else:
+            print(f"the above was Epoch {epoch} of {epoch_number}")
+    
+    if save:
+        torch.save(quantized_model.state_dict(), "full_quantStateDict.pth")
+                   
+    return model
+
+
+learning_rate = 1e-3
+batch_size = 64   
+data_size = 2560
+
+device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
+
+# Update to whatever you call your model
+modelName = "quantStateDict8.pth"
+load = False
+
+model = QuantizableMobileNetV2_5()
+
+# Adjust according to what your device supports
+torch.backends.quantized.engine = 'qnnpack'
+model.qconfig = torch.ao.quantization.default_qconfig
+
+optimiser = torch.optim.SGD(model.parameters(), lr= learning_rate)
+torch.ao.quantization.prepare_qat(model, inplace=True)
+
+
+dataset = ClassUtils.CrosswalkDataset("zebra_annotations/classification_data")
+train_loader = DataLoader(
+    Subset(dataset, random.sample(list(range(0, int(len(dataset) * 0.95))), data_size)),
+      batch_size=batch_size, shuffle=True)
+test_loader = DataLoader(
+    Subset(dataset, random.sample(list(range(int(len(dataset) * 0.95), len(dataset))), 256)),
+      batch_size=batch_size, shuffle=False)
+
+loss_function = nn.BCEWithLogitsLoss()
+model_updated = train_model(model, train_loader, loss_function, optimiser, epoch_number=8, const_save=True)
+
+quantized_model = torch.ao.quantization.convert(model_updated.eval(), inplace=True)
+
+if load:
+    model_loaded_state_dict = torch.load(modelName)
+    quantized_model.load_state_dict(model_loaded_state_dict)
+
+
+for images, labels in test_loader:
+    preds = torch.sigmoid(quantized_model(images))
+    for i in range(len(preds)):
+        print(preds)
+        # plt.imshow(torch.permute(images[i], (1, 2, 0)).detach().numpy())
+        # plt.title(f"Prediction: {preds[i]}, Actual: {labels[i][0] == 1}")
+        # plt.axis("off")
+        # plt.show()

SightLinks-Dev-main/LegacyVersion/ClassifierModel.py

@@ -0,0 +1,54 @@
+import numpy as np
+import torch.nn as nn
+
+
+# A really basic classifier model to detect points of interest (any likely crosswalks) for further investigation
+class BasicClassificationModel(nn.Module):
+    def __init__(self, image_size):
+        super().__init__()
+        self.image_size = image_size
+
+        # In channels are 3 - the RGB colours.
+        self.first_convolutional_layer = nn.Conv2d(3, 16, 5, padding=0)
+        self.second_convolutional_layer = nn.Conv2d(16, 32, 5, padding=0)
+        self.third_convolutional_layer = nn.Conv2d(32, 64, 5, padding=0)
+
+        self.fully_connected_layer = self.create_dynamic_output_layer()
+        # Actually predicts class probability
+
+        self.pooling_layer = nn.MaxPool2d(5)
+        self.activation_layer = nn.LeakyReLU(0.01)
+
+    # It is necessary to be able to take in images of a dynamic size, since may rescale depending on local regulations
+    # for crosswalk size. We can also assume squareness since this takes in the results of the segmentation model.
+    def create_dynamic_output_layer(self):
+        output_image_size = self.image_size
+
+        for layer in range(2):
+            output_image_size = ((output_image_size - 4) // 5)
+            # Convolution, then padding
+
+        output_image_size = output_image_size - 4
+        fully_connected_layer = nn.Linear(output_image_size * output_image_size * 64, 2)
+        return fully_connected_layer
+
+    def forward(self, x):
+        x = self.first_convolutional_layer(x)
+        x = self.pooling_layer(x)
+
+        x = self.second_convolutional_layer(x)
+        x = self.pooling_layer(x)
+
+        x = self.third_convolutional_layer(x)
+        x = self.activation_layer(x)
+
+        x = x.view(x.size(0), -1)
+        # Flattens the feature maps to pass into the fully connected layer - from 4D to 2D (batch, lin_tensor)
+
+        class_predictions = self.fully_connected_layer(x)
+
+        return class_predictions
+
+
+classifierModel = BasicClassificationModel(image_size=416)  # assumed square
+print(classifierModel)

SightLinks-Dev-main/LegacyVersion/ClassifierTraining.py

@@ -0,0 +1,73 @@
+import torch
+
+import CrosswalkDataset as Dataset
+import ClassifierModel as Model
+import Utilities as Utils
+
+
+def train_model_v0(model_to_train, dataset, epoch_number=25, loss_func=Utils.BasicClassificationLoss,
+                   batch_size=16, save=False):
+    optimiser = torch.optim.Adam(model_to_train.parameters(), lr=0.001)
+    dataloader = torch.utils.data.DataLoader(dataset, shuffle=True, batch_size=batch_size)
+
+    loss_function = loss_func()
+
+    for epoch in range(epoch_number):
+        model_to_train.train()
+        running_loss = 0.0
+
+        for images, gt_labels in dataloader:
+            optimiser.zero_grad()
+
+            predictions = model_to_train(images)
+
+            batch_loss = loss_function(predictions, gt_labels)
+            batch_loss.backward()
+
+            running_loss += batch_loss
+            optimiser.step()
+
+        print(f"Epoch [{epoch + 1} of {epoch_number}] finished, with loss {running_loss / len(dataloader)} in "
+              f"len {len(dataloader) * batch_size}")
+
+    Utils.save_model(model_to_train, optimiser)
+    return model_to_train
+
+
+# Additionally incorporated a learning rate scheduler
+def train_model_v1(model_to_train, dataset, epoch_number=10, loss_func=Utils.BasicClassificationLoss,
+                   batch_size=16, save=False):
+    optimiser = torch.optim.Adam(model_to_train.parameters(), lr=0.001)
+    scheduler = torch.optim.lr_scheduler.ExponentialLR(optimiser, gamma=0.95)
+    dataloader = torch.utils.data.DataLoader(dataset, shuffle=True, batch_size=batch_size)
+    loss_function = loss_func()
+
+    for epoch in range(epoch_number):
+        model_to_train.train()
+        running_loss = 0.0
+
+        for images, gt_labels in dataloader:
+            optimiser.zero_grad()
+
+            predictions = model_to_train(images)
+
+            batch_loss = loss_function(predictions, gt_labels)
+            batch_loss.backward()
+
+            running_loss += batch_loss
+            optimiser.step()
+
+        scheduler.step()
+
+        print(f"Epoch [{epoch + 1} of {epoch_number}] finished, with loss {running_loss / len(dataloader)} in "
+              f"len {len(dataloader) * batch_size}")
+
+    Utils.save_model(model_to_train, optimiser)
+    return model_to_train
+
+
+model = Model.BasicClassificationModel(image_size=416)
+# size should be dynamically obtained later on
+crosswalk_dataset = Dataset.CrosswalkDataset("Crosswalk.v7-crosswalk-t3.tensorflow/train/_annotations.csv",
+                                             "Crosswalk.v7-crosswalk-t3.tensorflow/train")
+model = train_model_v1(model, crosswalk_dataset, save=True)

SightLinks-Dev-main/LegacyVersion/CrosswalkDataset.py

@@ -0,0 +1,66 @@
+import torch
+import pandas as pd
+import numpy as np
+from PIL import Image
+
+
+class CrosswalkDataset:
+    def __init__(self, annotation_path, image_path, transform=None):
+        self.annotations = pd.read_csv(annotation_path)
+        self.image_dir = image_path
+
+        self.transform = transform
+        # In case later one we want to do normalisation, pre-processing etc. --> Someone else will look at this
+
+        self.unique_labels = sorted(self.annotations['class'].unique())
+        # There might be a more efficient method to do this -- come back to
+        self.type_mapping = {type_value: 1 + idx for idx, type_value in enumerate(self.unique_labels)}
+        # Labels have been converted to numerical class labels mapped by type mapping - for tensor conversion, 0 is
+        # the background object/ none type. For the current binary classifier everything other than zebra is 0
+
+        self.image_data = []
+        self.labels = []  # One-to-One mapping with bounding boxes by the way
+
+        self.process_annotations()
+
+    def __len__(self):
+        return len(self.image_data)
+
+    def __getitem__(self, index):
+        image = self.image_data[index]
+        class_label = self.labels[index]
+
+        tensor_image = torch.tensor(image, dtype=torch.float32).permute(2, 0, 1)
+        # From (H, W, C) to (C, H, W)  --> This is the format that pytorch uses.
+        tensor_label = torch.tensor(class_label, dtype=torch.float32)
+
+        if self.transform:
+            tensor_image = self.transform(tensor_image)
+
+        return tensor_image, tensor_label
+
+    def process_annotations(self):
+        for filename, group in self.annotations.groupby('filename'):
+            completed_image_path = f"{self.image_dir}/{filename}"
+            image = Image.open(completed_image_path)
+            image_array = np.array(image)
+            entity_annotations = [1, 0]
+
+            for _, row in group.iterrows():
+                # We can add additional classes in here - come back to later
+                if row['class'] == "ZebraStyle":
+                    # numerical_class = self.type_mapping[row['class']]
+                    entity_annotations = [0, 1]
+                    # print([numerical_class, (row['xmin'], row['ymin'], row['xmax'], row['ymax'])])
+
+                else:
+                    pass
+                    # Classified as a background object
+
+            self.image_data.append(image_array)
+            self.labels.append(entity_annotations)
+
+
+crosswalk_dataset = CrosswalkDataset("Crosswalk.v7-crosswalk-t3.tensorflow/train/_annotations.csv",
+                           "Crosswalk.v7-crosswalk-t3.tensorflow/train")
+

SightLinks-Dev-main/LegacyVersion/EvaluateModel.py

@@ -0,0 +1,38 @@
+import matplotlib.pyplot as plt
+import torch
+import Utilities as Utils
+import classifierModel
+import CrosswalkDataset as Dataset
+
+
+model = Utils.load_model_for_eval('trainedClassifier_weights.pth', classifierModel.BasicClassificationModel)
+dataset = Dataset.CrosswalkDataset("Crosswalk.v7-crosswalk-t3.tensorflow/test/_annotations.csv",
+                                   "Crosswalk.v7-crosswalk-t3.tensorflow/test")
+
+
+with torch.no_grad():
+    loss = 0.0
+    batch_size = 3
+    dataloader = torch.utils.data.DataLoader(dataset, shuffle=True, batch_size=batch_size)
+    loss_function = Utils.BasicClassificationLoss()
+    count, notCount = 0, 0
+    for images, gt_labels in dataloader:
+        predictions = model(images)
+        softmax_probabilities = Utils.softmax(predictions)
+        for i in range(len(images)):
+            plt.imshow(images[i].permute(1, 2, 0).numpy() / 255.0)
+            classif = False
+            if (gt_labels[i][1] > gt_labels[i][0] and softmax_probabilities[i][1] > softmax_probabilities[i][0]) or (gt_labels[i][1] <= gt_labels[i][0] and softmax_probabilities[i][1] <= softmax_probabilities[i][0]):
+                classif = True
+                count += 1
+            else:
+                notCount += 1
+            plt.title(str(softmax_probabilities[i]) + " " + str(gt_labels[i]) + str(classif))
+            plt.show()
+            print(softmax_probabilities[i])
+
+        batch_loss = loss_function(predictions, gt_labels)
+        loss += batch_loss
+
+    print("Loss is: ", loss / (len(dataloader) * batch_size))
+    print(count, notCount)

SightLinks-Dev-main/LegacyVersion/Utilities.py

@@ -0,0 +1,34 @@
+# This just contains loss functions and other things required for the classifier training process
+import numpy as np
+import torch
+import torch.nn as nn
+
+
+class BasicClassificationLoss(nn.Module):
+    def __init__(self):
+        super().__init__()
+        self.classification_loss = nn.CrossEntropyLoss()
+
+    def forward(self, pred_labels, gt_labels):
+        return self.classification_loss(pred_labels, gt_labels)
+
+
+def save_model(trained_model, optimiser_used):
+    torch.save(trained_model, 'trainedClassifier.pth')
+    print(",")
+    torch.save(trained_model.state_dict(), 'trainedClassifier_weights.pth')
+    torch.save(optimiser_used, 'optimiserUsed.pth')
+
+
+def load_model_for_eval(file_path, model_type):
+    model_template = model_type(416)
+    model_template.load_state_dict(torch.load(file_path, weights_only=True))
+    model_template.eval()
+    return model_template
+
+
+def softmax(unprocessed_logits):
+    logits = np.array(unprocessed_logits)
+    exponentials = np.exp(logits)
+    softmax_arr = exponentials / sum(exponentials)
+    return softmax_arr

SightLinks-Dev-main/README.md

@@ -0,0 +1,63 @@
+# SightLinks Development Repository
+
+Welcome to the SightLinks Development Repository! We hope you can use these tools to build accessible systems and make the world a more welcoming place. This repository contains various tools and resources designed to facilitate the development of the SightLinks project, focusing on data generation, annotation, model training, feature extraction and much more. These are the tools used to develop the SightLinks system, as well as some addiitional Beta features that can bring the project even further.
+
+## Project Overview
+
+SightLinks is an initiative aimed at enhancing accessibility mapping through automated analysis of satellite imagery. By accurately detecting features relevant to wheelchair users and individuals with mobility needs, SightLinks contributes to the development of more inclusive navigation solutions. The system achieves high reliability in feature detection while maintaining strict safety standards to prevent false positives. ([students.cs.ucl.ac.uk](https://students.cs.ucl.ac.uk/2024/group15/index.html))
+
+## Repository Structure
+
+- **DevelopmentTools/**: This directory includes tools and scripts used for data generation, annotation, model training, and feature pre-processing. These resources are essential for processing satellite imagery and training machine learning models to detect accessibility features, but that's not the only thing they're limited to being used for!
+
+- **LegacyVersion/**: This directory contains previous versions of the tools and scripts, preserved for reference and potential reuse.
+
+## Getting Started
+
+To begin using the tools in this repository:
+
+1. **Clone the Repository**:
+   ```bash
+   git clone https://github.com/UCL-SightLinks/SightLinks-Dev.git
+   ```
+2. **Navigate to the Development Tools Directory**:
+   ```bash
+   cd SightLinks-Dev/DevelopmentTools
+   ```
+3. **Install Dependencies**: Ensure you have Python installed, then install the necessary packages:
+   ```bash
+   pip install -r requirements.txt
+   ```
+4. **Run the Tools**: Follow the instructions provided in each file's documentation to generate data, annotate images, or train models. 
+
+## Contributing
+
+We welcome contributions to enhance the functionality and efficiency of the tools in this project, in fact we encourage it. To contribute:
+
+1. Fork the repository.
+2. Create a new branch for your feature or bug fix:
+   ```bash
+   git checkout -b feature-name
+   ```
+3. Commit your changes:
+   ```bash
+   git commit -m "Description of feature or fix"
+   ```
+4. Push to your branch:
+   ```bash
+   git push origin feature-name
+   ```
+5. Open a pull request detailing your changes. (we will most likely accept, there's a lot of improvements to be made!)
+
+## License
+
+This project is licensed under the MIT License, and owned by the UCL Computer Science department.
+
+## Contact
+
+For questions or further information, please contact the SightLinks development team at [zcabkde@ucl.ac.uk], or the university's computer science department as they have taken over this project ([UCL CS Department Staff Email List](https://www.ucl.ac.uk/computer-science/people/computer-science-professional-services-staff)).
+
+---
+
+*Note: For a comprehensive overview of the SightLinks project, including key features, functionalities, and team members, please visit our [project website](https://students.cs.ucl.ac.uk/2024/group15/index.html)!!* ([students.cs.ucl.ac.uk](https://students.cs.ucl.ac.uk/2024/group15/index.html))
+

SightLinks-Dev-main/requirements.txt

@@ -0,0 +1,14 @@
+overpy
+geopy
+requests
+torch
+torchvision
+matplotlib
+numpy
+pillow
+shapely
+pyyaml
+scipy
+scikit-image
+opencv-python
+pandas