DEV Community: victor_dalet

Rust and python bindings

victor_dalet — Fri, 03 Jan 2025 17:21:09 +0000

I - Install pip package

pip install maturin

II - Create project

maturin new --bindings pyo3 --mixed name_project

III - Create rust lib

Create in src matrix.rs with your code like this

use rand::Rng;
pub fn generate_matrix() -> Vec<Vec<i32>> {
    let mut matrix = vec![];
    for _ in 0..10 {
        let mut row = vec![];
        for _ in 0..10 {
            row.push(rand::thread_rng().gen_range(0..100));
        }
        matrix.push(row);
    }
    matrix
}

In src/lib.rs add your code like this :

mod matrix;
#[pyfunction]
fn random_matrix() -> Vec<Vec<i32>> {
    matrix::generate_matrix()
}

And functions in the python module like this :

#[pymodule]
fn my_project(m: &Bound<'_, PyModule>) -> PyResult<()> {
    m.add_function(wrap_pyfunction!(random_matrix, m)?)?;
    Ok(())
}

IV - Compile

maturin build && pip install .

V - Test

In python/name_project/test/ for exemple, create one file with this and run it.

import my_project

matrix = my_project.random_matrix()
print(matrix)

Où est Charlie - AI

victor_dalet — Wed, 20 Nov 2024 17:24:32 +0000

Hello, in this article we're going to find a technique to solve Où est Charlie with AI.

I - Dataset

I found a dataset on the following github:
https://github.com/kiim29/Ou_est_charlie

The first step is to convert this data to Yolo v5 format.
To do this, I create the following directories:

yolov5/
├── train
│   ├── images
│   └── labels
└── val
    ├── images
    └── labels

And put all images in the good folder.

For the annotation, I create a script to read the csv in the github and create the various label files with the position type transformation.
Beacause the format of yolov5 labels file is :
index_item (xmin + xmax) / 2 / width (ymin + ymax) / 2 / height (xmax - xmin) / width (ymax - ymin) / height

import pandas as pd

class Main:


    def __init__(self):
        csv = pd.read_csv("../annotations/annotations.csv")
        for i in range(len(csv)):
            filename = csv["filename"][i].split(".")[0]
            width = csv["width"][i]
            height = csv["height"][i]
            xmin = csv["xmin"][i]
            ymin = csv["ymin"][i]
            xmax = csv["xmax"][i]
            ymax = csv["ymax"][i]

            x_center = (xmin + xmax) / 2 / width
            y_center = (ymin + ymax) / 2 / height
            bbox_width = (xmax - xmin) / width
            bbox_height = (ymax - ymin) / height


            with open(f"../dataset/train/labels/{filename}.txt", "a") as f:
                f.write(f"0 {x_center} {y_center} {bbox_width} {bbox_height}\n")






if __name__ == "__main__":
    Main()

II - Train

For training I use ultralytics

pip install ultralytics

And I launch the training with the following arguments.

from ultralytics import YOLO

model = YOLO('yolov8n.pt')

model.train(data='dataset.yaml', epochs=50, patience=10000)

model.export()

III - Test

For testing purposes, we can run a random image with ultralytics and the new model :)

import sys

from ultralytics import YOLO


model = YOLO('../last.pt')

image_path = f'../dataset/train/images/{sys.argv[1]}.jpg'

results = model(image_path,conf=0.2)

Simple Cheat Sheet OVH AI

victor_dalet — Mon, 07 Oct 2024 17:22:12 +0000

Hi,

I post my ovh ai cheat sheet for lanching a simple Job.

Display job

ovhai job list

Display logs

ovhai job logs {id}

Run Job With env var

ovhai job run  {images}

Run Job With env var

ovhai job run -e {my_env_var} -e {my_second_en_var} {images}

Run Job With cmd

ovhai job run {images} -- echo "Hello ! "

Run Job With multiples cmd

ovhai job run {images} -- bash -c 'echo "Hello" && echo "World ! "'

Run Job With one GPU

ovhai job run --gpu 1 {images}

Run Job With monitoring port

ovhai job run -p {port} {images}

Creating an LLM for testing with tensorflow in Python

victor_dalet — Mon, 07 Oct 2024 17:12:13 +0000

Hi,

I want to test a small LLM program and I decided to do it with tensorflow .

My source code is available in https://github.com/victordalet/first_llm

I - Requirements

You need to install tensorflow and numpy

pip install 'numpy<2'
pip install tensorflow

II - Create Dataset

You need to make a data string array to countain a small dataset, for example I create :

data = [
    "Salut comment ca va",
    "Je suis en train de coder",
    "Le machine learning est une branche de l'intelligence artificielle",
    "Le deep learning est une branche du machine learning",
]

You can find a dataset on kaggle if you're not inspired.

III - Build model and train it

To do this, I create a small LLM class with the various methods.

class LLM:

    def __init__(self):
        self.model = None
        self.max_sequence_length = None
        self.input_sequences = None
        self.total_words = None
        self.tokenizer = None
        self.tokenize()
        self.create_input_sequences()
        self.create_model()
        self.train()
        test_sentence = "Pour moi le machine learning est"
        print(self.test(test_sentence, 10))

    def tokenize(self):
        self.tokenizer = Tokenizer()
        self.tokenizer.fit_on_texts(data)
        self.total_words = len(self.tokenizer.word_index) + 1

    def create_input_sequences(self):
        self.input_sequences = []
        for line in data:
            token_list = self.tokenizer.texts_to_sequences([line])[0]
            for i in range(1, len(token_list)):
                n_gram_sequence = token_list[:i + 1]
                self.input_sequences.append(n_gram_sequence)

        self.max_sequence_length = max([len(x) for x in self.input_sequences])
        self.input_sequences = pad_sequences(self.input_sequences, maxlen=self.max_sequence_length, padding='pre')

    def create_model(self):
        self.model = Sequential()
        self.model.add(Embedding(self.total_words, 100, input_length=self.max_sequence_length - 1))
        self.model.add(LSTM(150, return_sequences=True))
        self.model.add(Dropout(0.2))
        self.model.add(LSTM(100))
        self.model.add(Dense(self.total_words, activation='softmax'))

    def train(self):
        self.model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])

        X, y = self.input_sequences[:, :-1], self.input_sequences[:, -1]
        y = tf.keras.utils.to_categorical(y, num_classes=self.total_words)

        self.model.fit(X, y, epochs=200, verbose=1)

IV - Test

Finally, I test the model, with a test method called in the constructor of my classes.

Warning: I block generation in this test function if the word generated is identical to the previous one.

    def test(self, sentence: str, nb_word_to_generate: int):
        last_word = ""
        for _ in range(nb_word_to_generate):

            token_list = self.tokenizer.texts_to_sequences([sentence])[0]
            token_list = pad_sequences([token_list], maxlen=self.max_sequence_length - 1, padding='pre')
            predicted = np.argmax(self.model.predict(token_list), axis=-1)
            output_word = ""
            for word, index in self.tokenizer.word_index.items():
                if index == predicted:
                    output_word = word
                    break

            if last_word == output_word:
                return sentence

            sentence += " " + output_word
            last_word = output_word

        return sentence

3 example of computer vison technical test - Python / c++

victor_dalet — Sun, 08 Sep 2024 18:17:52 +0000

To install opencv in python

pip install opencv-python

To install opencv in c++

git clone https://github.com/opencv/opencv.git
mkdir -p build && cd build
cmake ../opencv
make -j4
sudo make install

CmakeLists.txt

cmake_minimum_required(VERSION 3.0)
project(opencv_c__)

find_package(OpenCV REQUIRED)
include_directories(${OpenCV_INCLUDE_DIRS})
set(CMAKE_CXX_STANDARD 17)

add_executable(opencv_c__ main.cpp)
target_link_libraries(opencv_c__ ${OpenCV_LIBS})

I - Mouvement detection

def ex1():
    cap = cv2.VideoCapture(0)

    object_detector = cv2.createBackgroundSubtractorMOG2()

    while True:
        ret, frame = cap.read()
        mask = object_detector.apply(frame)
        cv2.imshow('Video', mask)
        if cv2.waitKey(30) & 0xFF == 27:
            break

    cap.release()
    cv2.destroyAllWindows()

#include <iostream>

#include <opencv2/opencv.hpp>
#include <opencv2/videoio.hpp>
#include <opencv2/video.hpp>
#include <opencv2/imgproc.hpp>
#include <opencv2/highgui.hpp>

int main() {
    cv::VideoCapture cap(0);
    cv::Ptr<cv::BackgroundSubtractor> object_detector = cv::createBackgroundSubtractorMOG2();

    while (true) {
        cv::Mat frame;
        cap >> frame;
        cv::Mat mask;
        object_detector->apply(frame, mask);
        cv::imshow("Video", mask);
        if (cv::waitKey(30) == 27) {
            break;
        }
    }

    cap.release();
    cv::destroyAllWindows();
    return 0;
}

II - Blur faces

def ex2():
    cap = cv2.VideoCapture(0)

    while True:
        ret, frame = cap.read()
        face_cascade = cv2.CascadeClassifier('haarcascade_frontalface_default.xml')
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        faces = face_cascade.detectMultiScale(gray, 1.1, 4)
        for (x, y, w, h) in faces:
            roi = frame[y:y + h, x:x + w]
            roi = cv2.GaussianBlur(roi, (23, 23), 30)
            frame[y:y + h, x:x + w] = roi

        cv2.imshow("gray", gray)
        if cv2.waitKey(30) & 0xFF == 27:
            break

    cap.release()
    cv2.destroyAllWindows()

#include <iostream>

#include <opencv2/opencv.hpp>
#include <opencv2/videoio.hpp>
#include <opencv2/video.hpp>
#include <opencv2/imgproc.hpp>
#include <opencv2/highgui.hpp>


int main() {
    cv::VideoCapture cap(0);

    while(true) {
        cv::Mat frame;
        cap >> frame;

        cv::CascadeClassifier face_cascade;
        face_cascade.load("haarcascade_frontalface_default.xml");

        std::vector<cv::Rect> faces;
        face_cascade.detectMultiScale(frame, faces, 1.1, 3, 0, cv::Size(30, 30));

        for(int i = 0; i < faces.size(); i++) {
            cv::Rect face = faces[i];
            cv::Mat faceROI = frame(face);
            cv::blur(faceROI, faceROI, cv::Size(30, 30));
        }

        cv::imshow("frame", frame);

        if(cv::waitKey(1) == 27) {
            break;
        }
    }

}

III - Tracing a movement

def ex3():
    cap = cv2.VideoCapture(0)
    object_detector = cv2.createBackgroundSubtractorMOG2()

    last_coordinates = []

    while True:
        ret, frame = cap.read()
        mask = object_detector.apply(frame)
        contours, _ = cv2.findContours(mask, cv2.RETR_TREE, cv2.CHAIN_APPROX_SIMPLE)
        for contour in contours:
            if cv2.contourArea(contour) < 5000:
                continue
            (x, y, w, h) = cv2.boundingRect(contour)
            last_coordinates.append((x, y, w, h))
        for i in range(1, len(last_coordinates)):
            cv2.line(frame, (last_coordinates[i - 1][0], last_coordinates[i - 1][1]),
                     (last_coordinates[i][0], last_coordinates[i][1]), (0, 0, 255), 5)
        cv2.imshow('Video', frame)
        if cv2.waitKey(30) & 0xFF == 27:
            break

    cap.release()
    cv2.destroyAllWindows()

#include <iostream>

#include <opencv2/opencv.hpp>
#include <opencv2/videoio.hpp>
#include <opencv2/video.hpp>
#include <opencv2/imgproc.hpp>
#include <opencv2/highgui.hpp>


int main() {
    cv::VideoCapture cap(0);
    cv::Ptr<cv::BackgroundSubtractor> object_detector = cv::createBackgroundSubtractorMOG2();

    std::vector<cv::Rect> last_coordinates;

    while (true) {
        cv::Mat frame;
        cap >> frame;
        cv::Mat mask;
        object_detector->apply(frame, mask);
        std::vector<std::vector<cv::Point>> contours;
        std::vector<cv::Vec4i> hierarchy;
        cv::findContours(mask, contours, hierarchy, cv::RETR_TREE, cv::CHAIN_APPROX_SIMPLE);
        for (auto &contour : contours) {
            if (cv::contourArea(contour) < 5000) {
                continue;
            }
            cv::Rect rect = cv::boundingRect(contour);
            last_coordinates.push_back(rect);
        }
        for (int i = 1; i < last_coordinates.size(); i++) {
            cv::line(frame, cv::Point(last_coordinates[i - 1].x, last_coordinates[i - 1].y),
                     cv::Point(last_coordinates[i].x, last_coordinates[i].y), cv::Scalar(0, 0, 255), 5);
        }
        cv::imshow("Video", frame);
        if (cv::waitKey(30) & 0xFF == 27) {
            break;
        }
    }

    cap.release();
    cv::destroyAllWindows();
    return 0;
}

Emmision CO2 of a cat and a dog with python

victor_dalet — Mon, 02 Sep 2024 19:16:04 +0000

Hi,

I've found a small article that talks about pet emissions, so I've decided to display the CO2 emissions if they don't exist.

Code :
https://github.com/victordalet/Kaggle_analysis/tree/feat/dog_co2

Sources:

I - Get data

First, I obtain data estimating the world's CO2 consumption, the average emission of a dog and a cat, and the number of these pets.

import plotly.express as px


class Main:
    def __init__(self):
        self.estimation = {
            "2005": 750,
            "2010": 900,
            "2020": 1300,
            "2030": 1800,
            "2040": 2700,
            "2050": 4000,
        }

        self.estimation_no_cat = {
            "2005": 750,
            "2010": 900,
            "2020": 1300,
            "2030": 1800,
            "2040": 2700,
            "2050": 4000,
        }

        self.estimation_no_dog = {
            "2005": 750,
            "2010": 900,
            "2020": 1300,
            "2030": 1800,
            "2040": 2700,
            "2050": 4000,
        }

        self.estimation_no_cat_and_dog = {
            "2005": 750,
            "2010": 900,
            "2020": 1300,
            "2030": 1800,
            "2040": 2700,
            "2050": 4000,
        }

        self.cat_emission = 240
        self.dog_emission = 358
        self.nb_cats = 600000000
        self.nb_dogs = 900000000

II - Transformation

The total emission is in millions of tons, so I've created a method for converting the animal data, which are in kg.

    @staticmethod
    def transform_to_million_of_tonnes(value):
        return value / (1000000 * 1000)

III - Calcule

To modify the estimate without cat or dog, go through the first estimate and replace the values of the other dictionaries with the values found in the first step.

    def calculate(self):
        for year, value in self.estimation.items():
            self.estimation_no_cat[year] = value - self.transform_to_million_of_tonnes(
                self.cat_emission * self.nb_cats
            )
            self.estimation_no_dog[year] = value - self.transform_to_million_of_tonnes(
                self.dog_emission * self.nb_dogs
            )
            self.estimation_no_cat_and_dog[year] = (
                value
                - self.transform_to_million_of_tonnes(self.cat_emission * self.nb_cats)
                - self.transform_to_million_of_tonnes(self.dog_emission * self.nb_dogs)
            )

IV - Display result

To display a graph with all the data, I use the plotly library.

Code for pip installation :

pip install plotly

Code to display the three estimates :

    def display(self):
        fig = px.line(
            x=list(self.estimation.keys()),
            y=[
                list(self.estimation.values()),
                list(self.estimation_no_cat.values()),
                list(self.estimation_no_dog.values()),
                list(self.estimation_no_cat_and_dog.values()),
            ],
            labels={
                "x": "Year",
                "y": "CO2 Emission (in million of tonnes)",
                "color": "Legend",
            },
            title="CO2 Emission with and without cats and dogs",
            color_discrete_map={
                "CO2 Emission": "blue",
                "CO2 Emission without cats": "green",
                "CO2 Emission without dogs": "red",
                "CO2 Emission without cats and dogs": "orange",
            },
        )
        fig.show()

So now we have the chart with our results.

Website Time dataset

victor_dalet — Sun, 01 Sep 2024 13:33:19 +0000

Hello, I found a dataset on kaggle in the time of use of a website, so I want to find a ratio between the number of pages visited and the total time in the website.

You can find the dataset and the code in my github : https://github.com/victordalet/Kaggle_analysis/tree/feat/website_traffic

I - Installation

To do this, I use sqlalchemy in python to convert my csv into a database and plotly to display my results.

pip install plotly
pip install sqlalchemy

II - Code

I create a Main class, in which I retrieve my csv and put it in a database, using the get_data method.
The result is a list of tuples, so I create the transform_data method to obtain a double list.
Finally, I can display a simple graph between the number of pages viewed and the total time.

import pandas as pd
from sqlalchemy import create_engine, text
import plotly.express as px


class Main:
    def __init__(self):
        self.result = None
        self.connection = None

        self.engine = create_engine("sqlite:///my_database.db", echo=False)
        self.df = pd.read_csv("website_wata.csv")
        self.df.to_sql("website_data", self.engine, index=False, if_exists="append")
        self.get_data()
        self.transform_data()
        self.display_graph()


    def get_data(self):
        self.connection = self.engine.connect()
        query = text("SELECT Page_Views, Time_on_Page FROM website_data")
        self.result = self.connection.execute(query).fetchall()

    def transform_data(self):
        for i in range(len(self.result)):
            self.result[i] = list(self.result[i])


    def display_graph(self):
        fig = px.scatter(
            self.result, x=0, y=1, title=""
        )
        fig.show()


Main()

III - Result

The x-axis indicates the number of pages visited by the user, while the y-axis shows the time spent on the website in minutes.

We can see that the users who stay the longest visit between 4 and 6 pages, and that between 11 and 15 pages all users stay at least a few minutes.

Python - Jira ticket management

victor_dalet — Tue, 27 Aug 2024 21:53:13 +0000

Hello, I've just discovered Jira Lib, and I've decided to test it in order to find the people who make the most tickets.

Code in my github : https://github.com/victordalet/Jira_python_test

I - Installation

You juste need to python and install jira lib.

pip install jira

II - Log in

Declare the 3 variables with your information and go to https://id.atlassian.com/manage-profile/profile-and-visibility in security to generate your token (password).

JIRA_URL = "" # https://name.alassian.net
JIRA_USER = "" # me@name.fr
JIRA_PASSWORD = "" # token

III - Lib fonctionnement

I'm creating a class to get JIRA information, which can create a kind of mysql query like the get_tickets method to find resources or tickets, projects ...

class TicketManager:
    def __init__(self):
        self.jira = JIRA(JIRA_URL, basic_auth=(JIRA_USER, JIRA_PASSWORD))

    def get_projects(self):
        return self.jira.projects()

    def get_tickets(self, project_key: str):
        return self.jira.search_issues(f'project="{project_key}"')

    @staticmethod
    def ticket_status(ticket_):
        return ticket_.fields.status

    @staticmethod
    def ticket_reporter(ticket_):
        try:
            return ticket_.fields.reporter
        except AttributeError:
            return "Unknown"

    @staticmethod
    def ticket_assignee(ticket_):
        try:
            return ticket_.fields.assignee
        except AttributeError:
            return "Unknown"

IV - Display user activity

I browse the project to find all the tickets and add the right statues to a user dictionary.

if __name__ == '__main__':
    ticket_manager = TicketManager()
    projects = ticket_manager.get_projects()
    user = {}
    nb_total_tickets = 0
    for project in projects:
        tickets = ticket_manager.get_tickets(project.key)
        nb_total_tickets += len(tickets)
        for ticket in tickets:
            reporter = ticket_manager.ticket_reporter(ticket)
            assignee = ticket_manager.ticket_assignee(ticket)
            if assignee not in user:
                user[assignee] = {'ticket_to_do': 0, 'ticket_reported': 0}
            if reporter not in user:
                user[reporter] = {'ticket_to_do': 0, 'ticket_reported': 0}
            user[assignee]['ticket_to_do'] += 1
            user[reporter]['ticket_reported'] += 1

    print(f'There are {nb_total_tickets} tickets in total')

V - Sorting and display result

Now I'm sorting users by the number of tickets created and used.

    user = dict(sorted(user.items(), key=lambda x: (x[1]['ticket_to_do'], x[1]['ticket_reported']), reverse=True))

    for name, value in user.items():
        print(f'{name} : {value["ticket_to_do"]} tickets to do, {value["ticket_reported"]} tickets reported')

Result :

J.M. : 90 tickets to do, 60 tickets reported
L.M : 75 tickets to do, 21 tickets reported
J.M : 57 tickets to do, 76 tickets reported
V.M : 50 tickets to do, 0 tickets reported

Traefik with docker compose

victor_dalet — Sun, 25 Aug 2024 17:40:27 +0000

In this short post, I show how to use the traefik image in docker compose to publish your site with your domain name and generate an ssl certificate.

I - Create the traefik.toml file

Create the traefik.tom file
Copy the following code into it
Add your email address at the end of the file

[api]
  dashboard = true
  insecure = true

[entryPoints]
  [entryPoints.web]
    address = ":80"
    [entryPoints.web.http]
      [entryPoints.web.http.redirections]
        [entryPoints.web.http.redirections.entryPoint]
          to = "websecure"
          scheme = "https"
          permanent = true

  [entryPoints.websecure]
    address = ":443"
      [entryPoints.websecure.http.tls]
        certResolver = "default"

[providers]
  [providers.docker]
    watch = true
    exposedByDefault = false
    network = "web"

[certificatesResolvers]
  [certificatesResolvers.default]
    [certificatesResolvers.default.acme]
      email = ""
      storage = "acme.json"
      caServer = "https://acme-v02.api.letsencrypt.org/directory"
    [certificatesResolvers.default.acme.tlsChallenge]

II - Create the acme.json file

This file is used to store https certificates.
You must assign the correct right as in the following example.

touch acme.json
chmod 600 acme.json

III - Create network

Type the following command in your terminal

docker network create web

Add this code to the end of your docker-compose file

networks:
  web:
    external: true

IV - Traefik docker image

Here's the docker-compose block to call traefik, share the http and https ports and the many configuration files.
D'ont fortget to add the network we've just created inside.
Port 8080 is the traefik interface, so you don't have to add it.
The different labels are used to accept www subdomains.

  reverse-proxy:
    image: traefik:v2.4
    container_name: traefik
    ports:
      - "80:80"
      - "8080:8080"
      - "443:443"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock
      - traefik.toml:/etc/traefik/traefik.toml
      - acme.json:/acme.json
    labels:
      - "traefik.http.middlewares.strip-www.redirectregex.regex=^https?://(www\\.)(.+)"
      - "traefik.http.middlewares.strip-www.redirectregex.replacement=https://$${2}"
      - "traefik.http.middlewares.strip-www.redirectregex.permanent=true"
    restart: always
    networks:
      - web

V - Configure traefik for your docker compose service

In your other block, you need to add the network and the various labels.
The first label is to activate traefik, the second to add your domain or sub-domain (don't forget to change the name of your router --> in this example it's api and front), the last is to redirect to https.

  api:
    build: api/
    labels:
      - "traefik.enable=true"
      - "traefik.http.routers.api.rule=Host(`api.monsite.fr`)"
      - "traefik.http.routers.api.entrypoints=websecure"
    networks:
      - web
    restart: always

  front:
    build:
      context: front/
      dockerfile: Dockerfile.prod
    labels:
      - "traefik.enable=true"
      - "traefik.http.routers.front.rule=Host(`monsite.fr`)"
      - "traefik.http.routers.front.entrypoints=websecure"
    networks:
      - web

Create chat bot - JO PARIS 2024

victor_dalet — Sun, 25 Aug 2024 17:39:49 +0000

In this article, I show how to create a simple chat bot with tensorflow.

For the data, I use a kaggle dataset from PARIS JO JO 2024 to get sentences at the training stage.

You can obtain the finish code in my github : https://github.com/victordalet/Kaggle_analysis/tree/feat/paris_2024_olympics

I - Default chat bot dataset

A tensorflow dataset on chat bots looks like this.
We can find a tag, a pattern and the various responses.
Our goal will be to add the different sequences from the JO betting dataset and add them to a file like this.

{
  "intents": [
    {
      "tag": "google",
      "patterns": [
        "google",
        "search",
        "internet"
      ],
      "responses": [
        "Redirecting to Google..."
      ]
    },

II - Data processing

I read a chat bot dataset in default json and JO's csv and split and processed it to add the sentence in the json

import json


class CreateDataset:
    def __init__(self):
        self.json_path = 'data.json'
        self.csv_path = '../paris-2024-faq.csv'
        with open(self.json_path) as file:
            self.dataset = json.load(file)
        f = open(self.csv_path, 'r')
        dataset_split = f.read().split(";")
        question = False
        for data in dataset_split:
            if question:
                question = False
                self.dataset["intents"][-1]["responses"].append(data)

            if "?" in data:
                question = True
                self.dataset["intents"].append({
                    "tag": "",
                    "patterns": [
                        data
                    ],
                    "responses": [
                    ]
                })
        with open(self.json_path, 'w') as f:
            json.dump(self.dataset, f)

III - Training

For training purposes, I've edited a tensorflow example.
If you take my code to run it, add in the first argument the number of epochs you want.
Create a save directory where your model will go, and add inside the classes.pkl and words.pkl files that are in the github like at the beginning of this article.

import random
import json
import pickle
import numpy as np
import sys

import nltk
from nltk.stem import WordNetLemmatizer

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout
from tensorflow.keras.optimizers import SGD


class Train:
    words: list
    classes: list
    documents: list
    ignore_letters: list
    training: list
    output_empty: list
    train_x: list
    train_y: list
    model: Sequential
    epochs: int

    def __init__(self):
        self.lemmatizer = WordNetLemmatizer()
        self.intents = json.loads(open('data.json').read())
        self.words = []
        self.classes = []
        self.documents = []
        self.training = []
        self.ignore_letters = ['?', '!']
        self.epochs = int(sys.argv[1])

    def run(self):
        self.download_nltk_data()
        self.load_training_data()
        self.prepare_training_data()
        self.build_neural_network()
        self.train()

    @staticmethod
    def download_nltk_data():
        nltk.download('punkt')
        nltk.download('wordnet')

    def load_training_data(self):
        for intent in self.intents['intents']:
            for pattern in intent['patterns']:
                word_list = nltk.word_tokenize(pattern)
                self.words.extend(word_list)
                self.documents.append((word_list, intent['tag']))
                if intent['tag'] not in self.classes:
                    self.classes.append(intent['tag'])

    def prepare_training_data(self):
        self.words = [self.lemmatizer.lemmatize(word)
                      for word in self.words
                      if word not in self.ignore_letters]

        self.words = sorted(set(self.words))
        self.classes = sorted(set(self.classes))
        pickle.dump(self.words, open('saves/words.pkl', 'wb'))
        pickle.dump(self.classes, open('saves/classes.pkl', 'wb'))

        self.output_empty = [0] * len(self.classes)
        for document in self.documents:
            bag = []
            word_patterns = document[0]
            word_patterns = [self.lemmatizer.lemmatize(word.lower())
                             for word in word_patterns]
            for word in self.words:
                bag.append(1) if word in word_patterns else bag.append(0)

            output_row = list(self.output_empty)
            output_row[self.classes.index(document[1])] = 1
            self.training.append([bag, output_row])

        random.shuffle(self.training)
        self.training = np.array(self.training)

        self.train_x = list(self.training[:, 0])
        self.train_y = list(self.training[:, 1])

    def build_neural_network(self):
        self.model = Sequential()
        self.model.add(Dense(128, input_shape=(len(self.train_x[0]),),
                             activation='relu'))
        self.model.add(Dropout(0.5))
        self.model.add(Dense(64, activation='relu'))
        self.model.add(Dropout(0.5))
        self.model.add(Dense(len(self.train_y[0]), activation='softmax'))

        sgd = SGD(lr=0.01, momentum=0.9, nesterov=True)
        self.model.compile(loss='categorical_crossentropy',
                           optimizer=sgd,
                           metrics=['accuracy'])

    def train(self):
        self.model.fit(np.array(self.train_x),
                       np.array(self.train_y),
                       epochs=self.epochs,
                       batch_size=5,
                       verbose=1)
        self.model.save('saves/chatbot_model.model')


if __name__ == "__main__":
    Train().run()

IV - Test

I create a ChatBot class, with a test method that takes a random message.
You can use the get_response method to add this chatbot to your application, for example I call it in one of my projects in a flask api to have my chatbot in a website.

import random
import json
import pickle
import numpy as np

import nltk
from nltk.stem import WordNetLemmatizer
from tensorflow.keras.models import load_model


class ChatBot:
    lemmatizer: WordNetLemmatizer
    intents: dict
    words: list
    classes: list
    model: load_model
    ERROR_THRESHOLD = 0.25

    def __init__(self):
        self.download_nltk_data()
        self.lemmatizer = WordNetLemmatizer()
        self.intents = json.loads(open('data.json').read())
        self.words = pickle.load(open('saves/words.pkl', 'rb'))
        self.classes = pickle.load(open('saves/classes.pkl', 'rb'))
        self.model = load_model('saves/chatbot_model.model')

    @staticmethod
    def download_nltk_data():
        nltk.download('punkt')
        nltk.download('wordnet')

    def clean_up_sentence(self, sentence):
        sentence_words = nltk.word_tokenize(sentence)
        sentence_words = [self.lemmatizer.lemmatize(word)
                          for word in sentence_words]
        return sentence_words

    def bag_of_words(self, sentence):
        sentence_words = self.clean_up_sentence(sentence)
        bag = [0] * len(self.words)
        for w in sentence_words:
            for i, word in enumerate(self.words):
                if word == w:
                    bag[i] = 1
        return np.array(bag)

    def predict_class(self, sentence):
        bow = self.bag_of_words(sentence)
        res = self.model.predict(np.array([bow]))[0]
        results = [[i, r]
                   for i, r in enumerate(res)
                   if r > self.ERROR_THRESHOLD]
        results.sort(key=lambda x: x[1], reverse=True)
        return_list = []
        for r in results:
            return_list.append({'intent': self.classes[r[0]],
                                'probability': str(r[1])})
        return return_list

    def get_response(self, intents_list):
        intents_json = self.intents
        tag = intents_list[0]['intent']
        list_of_intents = intents_json['intents']
        for i in list_of_intents:
            if i['tag'] == tag:
                result = random.choice(i['responses'])
                break
        return result

    def test(self):
        while True:
            message = input("")
            ints = self.predict_class(message)
            res = self.get_response(ints)
            print(res)

Tensorflow music prediction

victor_dalet — Sun, 25 Aug 2024 17:39:17 +0000

In this article, I show how to use tensorflow to predict a style of music.
In my example, I compare techno and classical music.

You can find the code on my github :
https://github.com/victordalet/sound_to_partition

I - Dataset

For the first step, you need to create one dataset forlder and inside add one folder for music style, for example i add one techno folder and classic folder in which in put my wav soung.

II - Train

I create a train file, with the arguments max_epochs to be completed.

Modify the classes in the constructor that correspond to your directory in the dataset folder.

In the loading and processing method, I retrieve the wav file from a different directory and obtain the spectogram.

For training purposes, I use the Keras convolutions and model.

import os
import sys
from typing import List

import librosa
import numpy as np
from tensorflow.keras.layers import Input, Conv2D, MaxPooling2D, Flatten, Dense
from tensorflow.keras.models import Model
from tensorflow.keras.optimizers import Adam
from sklearn.model_selection import train_test_split
from tensorflow.keras.utils import to_categorical
from tensorflow.image import resize



class Train:

    def __init__(self):
        self.X_train = None
        self.X_test = None
        self.y_train = None
        self.y_test = None
        self.data_dir: str = 'dataset'
        self.classes: List[str] = ['techno','classic']
        self.max_epochs: int = int(sys.argv[1])

    @staticmethod
    def load_and_preprocess_data(data_dir, classes, target_shape=(128, 128)):
        data = []
        labels = []

        for i, class_name in enumerate(classes):
            class_dir = os.path.join(data_dir, class_name)
            for filename in os.listdir(class_dir):
                if filename.endswith('.wav'):
                    file_path = os.path.join(class_dir, filename)
                    audio_data, sample_rate = librosa.load(file_path, sr=None)
                    mel_spectrogram = librosa.feature.melspectrogram(y=audio_data, sr=sample_rate)
                    mel_spectrogram = resize(np.expand_dims(mel_spectrogram, axis=-1), target_shape)
                    data.append(mel_spectrogram)
                    labels.append(i)

        return np.array(data), np.array(labels)

    def create_model(self):
        data, labels = self.load_and_preprocess_data(self.data_dir, self.classes)
        labels = to_categorical(labels, num_classes=len(self.classes))  # Convert labels to one-hot encoding
        self.X_train, self.X_test, self.y_train, self.y_test = train_test_split(data, labels, test_size=0.2,
                                                                                random_state=42)

        input_shape = self.X_train[0].shape
        input_layer = Input(shape=input_shape)
        x = Conv2D(32, (3, 3), activation='relu')(input_layer)
        x = MaxPooling2D((2, 2))(x)
        x = Conv2D(64, (3, 3), activation='relu')(x)
        x = MaxPooling2D((2, 2))(x)
        x = Flatten()(x)
        x = Dense(64, activation='relu')(x)
        output_layer = Dense(len(self.classes), activation='softmax')(x)
        self.model = Model(input_layer, output_layer)

        self.model.compile(optimizer=Adam(learning_rate=0.001), loss='categorical_crossentropy', metrics=['accuracy'])

    def train_model(self):
        self.model.fit(self.X_train, self.y_train, epochs=self.max_epochs, batch_size=32,
                       validation_data=(self.X_test, self.y_test))
        test_accuracy = self.model.evaluate(self.X_test, self.y_test, verbose=0)
        print(test_accuracy[1])

    def save_model(self):
        self.model.save('weight.h5')


if __name__ == '__main__':
    train = Train()
    train.create_model()
    train.train_model()
    train.save_model()

III - Test

To test and use the model, I've created this class to retrieve the weight and predict the style of the music.

Don't forget to add the right classes to the constructor.

from typing import List

import librosa
import numpy as np
from tensorflow.keras.models import load_model
from tensorflow.image import resize
import tensorflow as tf



class Test:

    def __init__(self, audio_file_path: str):
        self.model = load_model('weight.h5')
        self.target_shape = (128, 128)
        self.classes: List[str] = ['techno','classic']
        self.audio_file_path: str = audio_file_path

    def test_audio(self, file_path, model):
        audio_data, sample_rate = librosa.load(file_path, sr=None)
        mel_spectrogram = librosa.feature.melspectrogram(y=audio_data, sr=sample_rate)
        mel_spectrogram = resize(np.expand_dims(mel_spectrogram, axis=-1), self.target_shape)
        mel_spectrogram = tf.reshape(mel_spectrogram, (1,) + self.target_shape + (1,))

        predictions = model.predict(mel_spectrogram)

        class_probabilities = predictions[0]

        predicted_class_index = np.argmax(class_probabilities)

        return class_probabilities, predicted_class_index

    def test(self):
        class_probabilities, predicted_class_index = self.test_audio(self.audio_file_path, self.model)

        for i, class_label in enumerate(self.classes):
            probability = class_probabilities[i]
            print(f'Class: {class_label}, Probability: {probability:.4f}')

        predicted_class = self.classes[predicted_class_index]
        accuracy = class_probabilities[predicted_class_index]
        print(f'The audio is classified as: {predicted_class}')
        print(f'Accuracy: {accuracy:.4f}')

ROBOFLOW - train & test with python

victor_dalet — Sun, 25 Aug 2024 17:38:54 +0000

Roboflow is a platform for annotating images for use in object detection AI.

I use this platform for C2SMR c2smr.fr, my computer vision association for maritime rescue.

In this article I show you how to use this platform and train your model with python.

You can find more sample code on my github : https://github.com/C2SMR/detector

I - Dataset

To create your dataset, go to https://app.roboflow.com/ and start annotating your image as shown in the following image.

In this example, I detour all the swimmers to predict their position in future images.
To get a good result, crop all the swimmers and place the bounding box just after the object to surround it correctly.

You can already use a public roboflow dataset, for this check https://universe.roboflow.com/

II - Training

For the training stage, you can use roboflow directly, but by the third time you'll have to pay, which is why I'm showing you how to do it with your laptop.

The first step is to import your dataset. To do this, you can import the Roboflow library.

pip install roboflow

To create a model, you need to use the YOLO algorithm, which you can import with the ultralytics library.

pip install ultralytics

In my script, I use the following command :

py train.py api-key project-workspace project-name project-version nb-epoch size_model

You must obtain :

the access key
workspace
roboflow project name
project dataset version
number of epochs to train the model
neural network size

Initially, the script downloads yolov8-obb.pt, the default yolo weight with pre-workout data, to facilitate training.

import sys
import os
import random
from roboflow import Roboflow
from ultralytics import YOLO
import yaml
import time


class Main:
    rf: Roboflow
    project: object
    dataset: object
    model: object
    results: object
    model_size: str

    def __init__(self):
        self.model_size = sys.argv[6]
        self.import_dataset()
        self.train()

    def import_dataset(self):
        self.rf = Roboflow(api_key=sys.argv[1])
        self.project = self.rf.workspace(sys.argv[2]).project(sys.argv[3])
        self.dataset = self.project.version(sys.argv[4]).download("yolov8-obb")

        with open(f'{self.dataset.location}/data.yaml', 'r') as file:
            data = yaml.safe_load(file)

        data['path'] = self.dataset.location

        with open(f'{self.dataset.location}/data.yaml', 'w') as file:
            yaml.dump(data, file, sort_keys=False)

    def train(self):
        list_of_models = ["n", "s", "m", "l", "x"]
        if self.model_size != "ALL" and self.model_size in list_of_models:

            self.model = YOLO(f"yolov8{self.model_size}-obb.pt")

            self.results = self.model.train(data=f"{self.dataset.location}/"
                                                 f"yolov8-obb.yaml",
                                            epochs=int(sys.argv[5]), imgsz=640)



        elif self.model_size == "ALL":
            for model_size in list_of_models:
                self.model = YOLO(f"yolov8{model_size}.pt")

                self.results = self.model.train(data=f"{self.dataset.location}"
                                                     f"/yolov8-obb.yaml",
                                                epochs=int(sys.argv[5]),
                                                imgsz=640)



        else:
            print("Invalid model size")



if __name__ == '__main__':
    Main()

III - Display

After training the model, you get the files best.py and last.py, which correspond to the weight.

With ultralytics library, you can also import YOLO and load your weight and then your test video.
In this example, I'm using the tracking function to get an ID for each swimmer.

import cv2
from ultralytics import YOLO
import sys


def main():
    cap = cv2.VideoCapture(sys.argv[1])

    model = YOLO(sys.argv[2])

    while True:
        ret, frame = cap.read()
        results = model.track(frame, persist=True)
        res_plotted = results[0].plot()
        cv2.imshow("frame", res_plotted)

        if cv2.waitKey(1) == 27:
            break

    cap.release()
    cv2.destroyAllWindows()


if __name__ == "__main__":
    main()

To analyze the prediction, you can obtain the model json as follows.

 results = model.track(frame, persist=True)
 results_json = json.loads(results[0].tojson())