Merge pull request #6 from Flowtter/settings

Settings

.gitignore

@@ -22,3 +22,5 @@ build/
 tmp/
 outputs/
 issues/
+
+settings.json

.pylintrc

@@ -15,4 +15,11 @@ disable=
     E1101,
     R0914,
     C0200,
-    W0614
+    W0614,
+    W1203,
+    C3001
+
+
+string-quote=double
+triple-quote=double
+docstring-quote=double

backend/requirements.txt

@@ -17,6 +17,7 @@ ffmpeg-python==0.2.0
 yapf==0.32.0
 mypy==0.961
 pylint==2.14.3
+pylint-quotes==0.2.3
 
 # Tests
 pytube==12.1.0

backend/src/app.py

@@ -79,7 +79,7 @@ def new_json() -> None:
     save_json()
 
 
-@app.get('/')
+@app.get("/")
 async def main_root():
     return JSON_INFO
 

backend/src/crispy/AI/network.py

@@ -2,11 +2,12 @@
 
 import numpy as np
 import scipy.special
-import scipy.ndimage
 
 
 class NeuralNetwork:
-    """Neural network to predict if a kill is on the image"""
+    """
+    Neural network to predict if a kill is on the image
+    """
 
     def __init__(self, nodes: List[int], learning_rate: float) -> None:
         self.nodes = nodes
@@ -15,15 +16,19 @@ def __init__(self, nodes: List[int], learning_rate: float) -> None:
         self.activation_function = lambda x: scipy.special.expit(x)
 
     def initialize_weights(self) -> None:
-        """Initialize the weights of the neural network"""
+        """
+        Initialize the weights of the neural network
+        """
         for i in range(len(self.nodes) - 1):
             w = np.random.normal(0.0, pow(self.nodes[i], -0.5),
                                  (self.nodes[i + 1], self.nodes[i]))
             self.weights.append(w)
 
     def _train(self, inputs: List[float],
                targets: Any) -> Tuple[int, int, int]:
-        """Train the neural network"""
+        """
+        Train the neural network
+        """
 
         inputs = np.array(inputs, ndmin=2).T
         targets = np.array(targets, ndmin=2).T
@@ -65,12 +70,16 @@ def _train(self, inputs: List[float],
         return expected == got, expected, got
 
     def mean_weights(self, N: "NeuralNetwork") -> None:
-        """Mitigate the weights of the current network with the weights of N"""
+        """
+        Mitigate the weights of the current network with the weights of N
+        """
         for i in range(len(self.weights)):
             self.weights[i] = (self.weights[i] + N.weights[i]) / 2
 
     def query(self, inputs: List[float]) -> List[float]:
-        """Query the neural network on a given input"""
+        """
+        Query the neural network on a given input
+        """
         inputs = np.array(inputs, ndmin=2).T
 
         outputs = []
@@ -84,12 +93,16 @@ def query(self, inputs: List[float]) -> List[float]:
         return outputs[-1]
 
     def save(self, filename: str) -> None:
-        """Save the weights of the neural network in numpy format"""
+        """
+        Save the weights of the neural network in numpy format
+        """
         print(f"Saving weights to {filename}.npy")
         np.save(filename, self.weights)
 
     def load(self, filename: str) -> None:
-        """Load the weights of the neural network from numpy format"""
+        """
+        Load the weights of the neural network from numpy format
+        """
         self.weights = np.load(filename, allow_pickle=True)
 
     def __str__(self) -> str:

backend/src/crispy/AI/trainer.py

@@ -18,7 +18,9 @@
 
 
 class Trainer(NeuralNetwork):
-    """Trainer for the neural network"""
+    """
+    Trainer for the neural network
+    """
 
     def __init__(self, nodes: List[int], learning_rate: float) -> None:
         super().__init__(nodes, learning_rate)
@@ -28,7 +30,9 @@ def __init__(self, nodes: List[int], learning_rate: float) -> None:
 
     @staticmethod
     def move_images_from_histogram(histogram: List[int], path: str) -> None:
-        """Debugging method to move images to the folder `issues`"""
+        """
+        Debugging method to move images to the folder `issues`
+        """
         assert DEBUG
         maximum = max(histogram)
 
@@ -43,7 +47,9 @@ def move_images_from_histogram(histogram: List[int], path: str) -> None:
 
     def train(self, epochs: int, inputs: List[List[float]],
               targets: List[Any]) -> None:
-        """Train the neural network for a given number of epochs"""
+        """
+        Train the neural network for a given number of epochs
+        """
         for epoch in range(epochs):
             print("===\n\tEpoch:", epoch)
             progress_bar = progressbar.ProgressBar(max_value=len(inputs))
@@ -70,7 +76,9 @@ def train(self, epochs: int, inputs: List[List[float]],
         self.save("./outputs/trained_network_" + self.hash + "_end")
 
     def test(self, inputs: List[List[float]], targets: List[Any]) -> bool:
-        """Test the neural network"""
+        """
+        Test the neural network
+        """
         print("Testing...")
         accuracy_score = 0
         failed = []
@@ -102,15 +110,17 @@ def __str__(self) -> str:
 
 
 def get_inputs_targets(path: str) -> Tuple[List[List[float]], List[Any]]:
-    """Read the path csv file and return the inputs and targets"""
-    with open(path, 'r') as f:
+    """
+    Read the path csv file and return the inputs and targets
+    """
+    with open(path, "r") as f:
         test_data_list = f.readlines()
 
         final_inputs = []
         final_targets = []
 
         for record in test_data_list:
-            all_values = record.split(',')
+            all_values = record.split(",")
             inputs = (np.asfarray(all_values[1:]) / 255.0 * 0.99) + 0.01
 
             targets = np.zeros(2) + 0.01
@@ -123,22 +133,25 @@ def get_inputs_targets(path: str) -> Tuple[List[List[float]], List[Any]]:
 
 
 def test(trainer: Trainer, path: str) -> bool:
-    """Wrapper for the test method"""
+    """
+    Wrapper for the test method
+    """
     final_inputs, final_targets = get_inputs_targets(path)
     return trainer.test(final_inputs, final_targets)
 
 
 def train(epoch: int, trainer: Trainer, path: str) -> None:
-    """Wrapper for the train method"""
+    """
+    Wrapper for the train method
+    """
     final_inputs, final_targets = get_inputs_targets(path)
     trainer.train(epoch, final_inputs, final_targets)
 
 
 if __name__ == "__main__":
     t = Trainer([4000, 120, 15, 2], 0.01)
-
-    csv_path = "./backend/dataset/result.csv"
-    csv_test_path = "./backend/dataset/test.csv"
+    csv_path = os.path.join("backend", "dataset", "result.csv")
+    csv_test_path = os.path.join("backend", "dataset", "test.csv")
 
     parser = argparse.ArgumentParser()
     parser.add_argument("--train",
@@ -152,10 +165,7 @@ def train(epoch: int, trainer: Trainer, path: str) -> None:
     parser.add_argument("--load",
                         help="Load a trained network",
                         action="store_true")
-    parser.add_argument("--path",
-                        help="Path to the network",
-                        type=str,
-                        default="./backend/assets/trained_network_latest.npy")
+    parser.add_argument("--path", help="Path to the network", type=str)
 
     parser.add_argument("--debug", help="Debug mode", action="store_true")
 

backend/src/crispy/main.py

@@ -1 +1,61 @@
-print("Hello World!")
+import logging
+from typing import List
+
+from utils.arguments import args
+from utils.constants import NEURAL_NETWORK_PATH
+from utils.IO import io
+import video.video as vid
+from AI.network import NeuralNetwork
+
+logging.getLogger("PIL").setLevel(logging.ERROR)
+
+
+def main(videos: List[str]) -> None:
+    io.generate_tmp_folder(not args.no_extract)
+
+    nn = NeuralNetwork([4000, 120, 15, 2], 0.01)
+    nn.load(NEURAL_NETWORK_PATH)
+    l.debug(f"Neural network: {nn}")
+
+    for video in videos:
+        l.info(f"Currently processing {video}")
+        video_no_ext = io.remove_extension(video)
+        video_clean_name = io.generate_clean_name(video_no_ext)
+        l.debug(f"Clean name: {video_clean_name}")
+
+        if not args.no_extract:
+            io.generate_folder_clip(video_clean_name)
+            images_path = vid.extract_frames_from_video(video)
+        else:
+            images_path = vid.get_saving_path(video_clean_name)
+
+        if not args.no_segmentation:
+            io.clean_cuts(video_clean_name)
+
+            query_array = vid.get_query_array_from_video(nn, images_path)
+            l.debug(query_array)
+            kill_array = vid.get_kill_array_from_query_array(query_array)
+            l.debug(kill_array)
+            kill_array = vid.post_processing_kill_array(kill_array)
+            l.debug(kill_array)
+            vid.segment_video_with_kill_array(video, kill_array)
+
+    if not args.no_merge:
+        vid.merge_cuts()
+
+
+if __name__ == "__main__":
+    l = logging.getLogger()
+
+    print("Welcome to crispy!")
+
+    l.info("Starting the program crispy")
+
+    l.debug(f"Arguments: {args}")
+
+    videos_path = ["4.mp4", "quadra chrlie mice.mp4"]
+
+    # FIXME: should be sort with the frontend ?
+    videos_path.sort()
+
+    main(videos_path)

backend/src/crispy/utils/IO/io.py

@@ -0,0 +1,82 @@
+import os
+import shutil
+
+from utils.constants import TMP_PATH, IMAGE, CUT
+
+
+def generate_tmp_folder(overwrite: bool) -> None:
+    """
+    Generate the tmp directory
+    """
+    if not os.path.exists(TMP_PATH):
+        os.makedirs(TMP_PATH)
+    elif overwrite:
+        clear_directory(TMP_PATH)
+        os.makedirs(TMP_PATH)
+
+
+def generate_folder_clip(name: str, overwrite: bool = True) -> None:
+    """
+    Generate the folder clip
+    Will generate:
+    tmp/{name}/cut
+    tmp/{name}/images
+    """
+    path = os.path.join(TMP_PATH, name)
+    if not os.path.exists(path):
+        os.makedirs(path)
+        os.makedirs(os.path.join(path, CUT))
+        os.makedirs(os.path.join(path, IMAGE))
+    elif overwrite:
+        clear_directory(path)
+        generate_folder_clip(name)
+
+
+def clear_directory(path: str) -> None:
+    """
+    Clear the given directory
+    """
+    if os.path.exists(path):
+        shutil.rmtree(path)
+
+
+def generate_clean_name(name: str) -> str:
+    """
+    Generate the path from the name
+    Remove the delim characters from the file name
+    Add hash so (hello world) and (hello_world) are different
+    """
+
+    def custom_hash(string: str) -> str:
+        """
+        Generate a hash from the name, same for each session
+        """
+        h = 0
+        for ch in string:
+            h = (h * 281 ^ ord(ch) * 997) & 0xFFFFFFFF
+        return str(h)
+
+    init_name = name
+    name = name.replace(" ", "__")
+    name = name.replace("-", "__")
+    name = name.replace("/", "__")
+    name = name.replace("\\", "__")
+
+    return name + "-" + custom_hash(init_name)
+
+
+def remove_extension(name: str) -> str:
+    """
+    Remove the extension from the file name
+    """
+    return name.split(".")[0]
+
+
+def clean_cuts(folder: str) -> None:
+    """
+    Clean the cuts folder
+    """
+    path = os.path.join(TMP_PATH, folder, CUT)
+    if os.path.exists(path):
+        shutil.rmtree(path)
+    os.makedirs(path)