levraistagiaire/Stage_IJL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Started implementation of GNG

Browse Source 
chore: Updated requirements.txt
This commit is contained in:
Le Stagiaire
2024-05-31 14:44:48 +02:00
parent 29c0474bd8
commit 83fb326cb3
6 changed files with 471 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
3D_app/README.md
View File

@ -1,3 +1,8 @@
# Installation
1. Install Python requirements: `pip install -r requirements.txt`
2. Launch program: `py main.py`
# Changelog
## V1

109
3D_app/gng2.py Normal file
View File

@ -0,0 +1,109 @@
import numpy as np
import plotly.graph_objects as plt
class Neuron:
def __init__(self, position):
self.position = position
self.error = 0.0
class Edge:
def __init__(self, node1, node2):
self.nodes = (node1, node2)
self.age = 0
class GrowingNeuralGas:
def __init__(self, input_dim, max_nodes=100, max_age=100, epsilon_b=0.05, epsilon_n=0.006, alpha=0.5, delta=0.995, lambda_=100):
self.input_dim = input_dim
self.max_nodes = max_nodes
self.max_age = max_age
self.epsilon_b = epsilon_b
self.epsilon_n = epsilon_n
self.alpha = alpha
self.delta = delta
self.lambda_ = lambda_
self.nodes = []
self.edges = []
# Initialize with two random neurons
self.nodes.append(Neuron(np.random.rand(input_dim)))
self.nodes.append(Neuron(np.random.rand(input_dim)))
self.edges.append(Edge(self.nodes[0], self.nodes[1]))
def fit(self, X, num_iterations=1000):
for iteration in range(num_iterations):
# Step 1: Select a random input sample
x = X[np.random.randint(len(X))]
# Step 2: Find the nearest and second nearest neurons
dists = np.array([np.linalg.norm(node.position - x) for node in self.nodes])
winner_idx = np.argmin(dists)
winner = self.nodes[winner_idx]
dists[winner_idx] = np.inf
second_winner_idx = np.argmin(dists)
second_winner = self.nodes[second_winner_idx]
# Step 3: Increment age of edges connected to the winner
for edge in self.edges:
if winner in edge.nodes:
edge.age += 1
# Step 4: Add error to the winner
winner.error += np.linalg.norm(winner.position - x) ** 2
# Step 5: Move the winner and its topological neighbors
winner.position += self.epsilon_b * (x - winner.position)
for edge in self.edges:
if winner in edge.nodes:
other = edge.nodes[0] if edge.nodes[1] is winner else edge.nodes[1]
other.position += self.epsilon_n * (x - other.position)
# Step 6: Connect the winner with the second winner
edge = self.find_edge(winner, second_winner)
if edge:
edge.age = 0
else:
self.edges.append(Edge(winner, second_winner))
# Step 7: Remove old edges
self.edges = [edge for edge in self.edges if edge.age <= self.max_age]
# Step 8: Remove isolated nodes
self.nodes = [node for node in self.nodes if any(node in edge.nodes for edge in self.edges)]
# Step 9: Insert new nodes
if iteration % self.lambda_ == 0 and len(self.nodes) < self.max_nodes:
self.insert_node()
# Step 10: Decrease all errors
for node in self.nodes:
node.error *= self.delta
def find_edge(self, node1, node2):
for edge in self.edges:
if node1 in edge.nodes and node2 in edge.nodes:
return edge
return None
def insert_node(self):
# Find the node with the largest error
q = max(self.nodes, key=lambda node: node.error)
# Find the neighbor with the largest error
connected_edges = [edge for edge in self.edges if q in edge.nodes]
f = max((node for edge in connected_edges for node in edge.nodes if node is not q), key=lambda node: node.error)
# Insert a new neuron halfway between q and f
r_position = (q.position + f.position) / 2
r = Neuron(r_position)
self.nodes.append(r)
# Remove the edge between q and f and add edges q-r and r-f
self.edges = [edge for edge in self.edges if not (q in edge.nodes and f in edge.nodes)]
self.edges.append(Edge(q, r))
self.edges.append(Edge(r, f))
# Decrease the error of q and f
q.error *= self.alpha
f.error *= self.alpha
r.error = q.error

483
3D_app/pages/ascan.py
View File

@ -10,18 +10,20 @@ from filtrage import *
from selection_filtre import *
dash.register_page(__name__, path="/ascan", title='A-Scan filters', name='A-Scan filters')
dash.register_page(
__name__, path="/ascan", title="A-Scan filters", name="A-Scan filters"
)
# on définit le dossier et les fichiers à lire
dossier = "Dataset/Shear_Wave_Rot00_CSV_Data"
fichiers_selectionnes = [
"Shear_x001-x101_y{:03d}_Rot00.csv".format(i) for i in range(10,13)
"Shear_x001-x101_y{:03d}_Rot00.csv".format(i) for i in range(10, 13)
]
# on lit les fichiers et on les met dans un tableau
pre_volume = np.array(lire_fichier_csv(dossier, fichiers_selectionnes))
volume = pre_volume[:, :, :]
data_traits=volume
data_traits = volume
dim_x, dim_y, dim_z = volume.shape
X, Y, Z = np.mgrid[0:dim_x, 0:dim_y, 0:dim_z]
@ -64,8 +66,14 @@ layout = html.Div(
{"label": "filtre passe bas ", "value": "3"},
{"label": "filtre de moyenne mobile", "value": "4"},
{"label": "filtre adaptatif (wiener)", "value": "5"},
{"label": "filtre à réponse impulsionnelle infinie", "value": "6"},
{"label": "filtre à réponse impulsionnelle finie", "value": "7"},
{
"label": "filtre à réponse impulsionnelle infinie",
"value": "6",
},
{
"label": "filtre à réponse impulsionnelle finie",
"value": "7",
},
],
value=2,
style={"margin-bottom": "15px"},
@ -82,153 +90,158 @@ layout = html.Div(
{"label": "filtre passe bas ", "value": "3"},
{"label": "filtre de moyenne mobile", "value": "4"},
{"label": "filtre adaptatif (wiener)", "value": "5"},
{"label": "filtre à réponse impulsionnelle infinie", "value": "6"},
{"label": "filtre à réponse impulsionnelle finie", "value": "7"},
{
"label": "filtre à réponse impulsionnelle infinie",
"value": "6",
},
{
"label": "filtre à réponse impulsionnelle finie",
"value": "7",
},
],
value=2,
style={"margin-bottom": "15px"},
),
],
width=3,
),
dbc.Col(
[
dbc.Label("applique les filtres selections sur tous les data"),
dbc.Label(
"applique les filtres selections sur tous les data",
style={"marginRight": "5px"},
),
dbc.Button(
id="button-validate-filter",
children=dbc.Spinner(html.Div(id="loading"), show_initially=False),
children=dbc.Spinner(
html.Div("Valider", id="loading"), show_initially=False
),
color="primary",
style={"marginBottom": "15px"},
),
],
width=3,
),
]
),
dbc.Row(
[
dbc.Label(" paramètre du 1er filtre ", html_for="Fs "),
dbc.Col(
[
dbc.Label("Fs ", html_for="Fs "),
dbc.Input(
id="input-ascan-solo-fs",
type="number",
placeholder="Fs",
value=1,
step=0.1,
style={"marginTop": "15px"},
dbc.Col(
[html.Br(), html.B(" paramètre du 1er filtre ")],
width=2,
style={"textAlign": "center"},
),
],
width=3,
dbc.Col(
[
dbc.Label("Fs ", html_for="Fs "),
dbc.Input(
id="input-ascan-solo-fs",
type="number",
placeholder="Fs",
value=1,
step=0.1,
),
],
width=1,
),
dbc.Col(
[
dbc.Label("cut off ", html_for="cut off"),
dbc.Input(
id="input-ascan-solo-cutoff",
type="number",
placeholder="cut_off",
value=1,
step=0.1,
),
],
width=1,
),
dbc.Col(
[
dbc.Label("order ", html_for="order"),
dbc.Input(
id="input-ascan-solo-order",
type="number",
placeholder="order",
value=1,
step=1,
),
],
width=1,
),
dbc.Col(
[
dbc.Label("window size ", html_for="window size"),
dbc.Input(
id="input-ascan-solo-windowsize",
type="number",
placeholder="window_size",
value=1,
step=1,
),
],
width=1,
),
dbc.Col(
[html.Br(), html.B(" paramètre du 2e filtre ")],
width=2,
style={"textAlign": "center"},
),
dbc.Col(
[
dbc.Label("Fs ", html_for="Fs "),
dbc.Input(
id="input-ascan-solo-fs-2",
type="number",
placeholder="Fs",
value=1,
step=0.1,
),
],
width=1,
),
dbc.Col(
[
dbc.Label("cut off ", html_for="cut off"),
dbc.Input(
id="input-ascan-solo-cutoff-2",
type="number",
placeholder="cut_off",
value=1,
step=0.1,
),
],
width=1,
),
dbc.Col(
[
dbc.Label("order ", html_for="order"),
dbc.Input(
id="input-ascan-solo-order-2",
type="number",
placeholder="order",
value=1,
step=1,
),
],
width=1,
),
dbc.Col(
[
dbc.Label("window size ", html_for="window size"),
dbc.Input(
id="input-ascan-solo-windowsize-2",
type="number",
placeholder="window_size",
value=1,
step=1,
),
],
width=1,
),
]
),
dbc.Col(
[
dbc.Label("cut off ", html_for="cut off"),
dbc.Input(
id="input-ascan-solo-cutoff",
type="number",
placeholder="cut_off",
value=1,
step=0.1,
style={"marginTop": "15px"},
),
],
width=3,
),
dbc.Col(
[
dbc.Label("order ", html_for="order"),
dbc.Input(
id="input-ascan-solo-order",
type="number",
placeholder="order",
value=1,
step=1,
style={"marginTop": "15px"},
),
],
width=3,
),
dbc.Col(
[
dbc.Label("window size ", html_for="window size"),
dbc.Input(
id="input-ascan-solo-windowsize",
type="number",
placeholder="window_size",
value=1,
step=1,
style={"marginTop": "15px"},
),
],
width=3,
),]),
dbc.Row(
[
dbc.Label("paramètre de 2eme filtre ", html_for="Fs "),
dbc.Col(
[
dbc.Label("Fs ", html_for="Fs "),
dbc.Input(
id="input-ascan-solo-fs-2",
type="number",
placeholder="Fs",
value=1,
step=0.1,
style={"marginTop": "15px"},
),
],
width=3,
),
dbc.Col(
[
dbc.Label("cut off ", html_for="cut off"),
dbc.Input(
id="input-ascan-solo-cutoff-2",
type="number",
placeholder="cut_off",
value=1,
step=0.1,
style={"marginTop": "15px"},
),
],
width=3,
),
dbc.Col(
[
dbc.Label("order ", html_for="order"),
dbc.Input(
id="input-ascan-solo-order-2",
type="number",
placeholder="order",
value=1,
step=1,
style={"marginTop": "15px"},
),
],
width=3,
),
dbc.Col(
[
dbc.Label("window size ", html_for="window size"),
dbc.Input(
id="input-ascan-solo-windowsize-2",
type="number",
placeholder="window_size",
value=1,
step=1,
style={"marginTop": "15px"},
),
],
width=3,
),
]),
dbc.Row(
[
dbc.Col(
@ -265,7 +278,7 @@ layout = html.Div(
value=1,
step=1,
marks={
str(i): str(i) for i in range(1, dim_z+1,max(1, int(dim_z / 20)))
str(i): str(i) for i in range(1, dim_z + 1, max(1, int(dim_z / 20)))
},
),
dbc.Label("y"),
@ -276,25 +289,25 @@ layout = html.Div(
value=1,
step=1,
marks={
str(i): str(i) for i in range(1, dim_x+1,max(1, int(dim_x / 20)))
},
str(i): str(i) for i in range(1, dim_x + 1, max(1, int(dim_x / 20)))
},
),
dbc.Label("z"),
dcc.RangeSlider(
id="layer-slider-ascan-solo-z",
min=1,
max=dim_y,
value=[dim_y/dim_y,dim_y],
value=[dim_y / dim_y, dim_y],
step=1,
marks={
str(i): str(i) for i in range(1, dim_y+1,max(1, int(dim_y / 20)))
},
str(i): str(i) for i in range(1, dim_y + 1, max(1, int(dim_y / 20)))
},
),
],
style={"padding": "20px"},
)
# callback to update filter values
@callback(
[
@ -313,24 +326,46 @@ layout = html.Div(
],
)
def update_filter_values(select_filtre_1, select_filtre_2):
fs_1=True;cutoff_1=True;ordre_1=True;windowsize_1=True
fs_2=True;cutoff_2=True;ordre_2=True;windowsize_2=True
if (int(select_filtre_1)==3):
fs_1=False;cutoff_1=False;ordre_1=False
if(int(select_filtre_2)==3):
fs_2=False;cutoff_2=False;ordre_2=False
if(int(select_filtre_1) in (4, 5, 6, 7)):
windowsize_1=False
if(int(select_filtre_2) in (4, 5, 6, 7)):
windowsize_2=False
return [fs_1, cutoff_1, ordre_1, windowsize_1, fs_2, cutoff_2, ordre_2, windowsize_2]
fs_1 = True
cutoff_1 = True
ordre_1 = True
windowsize_1 = True
fs_2 = True
cutoff_2 = True
ordre_2 = True
windowsize_2 = True
if int(select_filtre_1) == 3:
fs_1 = False
cutoff_1 = False
ordre_1 = False
if int(select_filtre_2) == 3:
fs_2 = False
cutoff_2 = False
ordre_2 = False
if int(select_filtre_1) in (4, 5, 6, 7):
windowsize_1 = False
if int(select_filtre_2) in (4, 5, 6, 7):
windowsize_2 = False
return [
fs_1,
cutoff_1,
ordre_1,
windowsize_1,
fs_2,
cutoff_2,
ordre_2,
windowsize_2,
]
# callback to update the heatmap
@callback(
[Output("heatmap-ascan-solo", "figure"), Output("heatmap-bscan-solo", "figure"),Output("heatmap-fft-solo", "figure"),Output("loading", "children"),],
[
Output("heatmap-ascan-solo", "figure"),
Output("heatmap-bscan-solo", "figure"),
Output("heatmap-fft-solo", "figure"),
Output("loading", "children"),
],
[
Input("select-ascan-filter1", "value"),
Input("select-ascan-filter2", "value"),
@ -349,38 +384,128 @@ def update_filter_values(select_filtre_1, select_filtre_2):
Input("input-ascan-solo-windowsize-2", "value"),
],
)
def update_heatmap_ascan(selec_transforme_hilbert,select_filtre_1,select_filtre_2,select_ascan_x,select_ascan_y,select_ascan_z,n_clicks,fs_filtre_1,cutoff_filtre_1,order_filtre_1,windowsize_filtre_1,fs_filtre_2,cutoff_filtre_2,order_filtre_2,windowsize_filtre_2):
def update_heatmap_ascan(
selec_transforme_hilbert,
select_filtre_1,
select_filtre_2,
select_ascan_x,
select_ascan_y,
select_ascan_z,
n_clicks,
fs_filtre_1,
cutoff_filtre_1,
order_filtre_1,
windowsize_filtre_1,
fs_filtre_2,
cutoff_filtre_2,
order_filtre_2,
windowsize_filtre_2,
):
# TODO: implement the filter
print("debut du traitement")
data_avec_traitement=volume[int(select_ascan_y)-1,select_ascan_z[0]:select_ascan_z[1],int(select_ascan_x)-1]
data_sans_traitement=volume[int(select_ascan_y)-1,select_ascan_z[0]:select_ascan_z[1],int(select_ascan_x)-1]
data_avec_traitement=switch_case(data_avec_traitement,int(selec_transforme_hilbert))
data_sans_traitement=switch_case(data_sans_traitement,int(selec_transforme_hilbert))
data_avec_traitement=switch_case(data_avec_traitement,int(select_filtre_1),float(fs_filtre_1),float(cutoff_filtre_1),int(order_filtre_1),int(windowsize_filtre_1))
data_avec_traitement=switch_case(data_avec_traitement,int(select_filtre_2),float(fs_filtre_2),float(cutoff_filtre_2),int(order_filtre_2),int(windowsize_filtre_2))
data_avec_traitement = volume[
int(select_ascan_y) - 1,
select_ascan_z[0] : select_ascan_z[1],
int(select_ascan_x) - 1,
]
data_sans_traitement = volume[
int(select_ascan_y) - 1,
select_ascan_z[0] : select_ascan_z[1],
int(select_ascan_x) - 1,
]
data_avec_traitement = switch_case(
data_avec_traitement, int(selec_transforme_hilbert)
)
data_sans_traitement = switch_case(
data_sans_traitement, int(selec_transforme_hilbert)
)
data_avec_traitement = switch_case(
data_avec_traitement,
int(select_filtre_1),
float(fs_filtre_1),
float(cutoff_filtre_1),
int(order_filtre_1),
int(windowsize_filtre_1),
)
data_avec_traitement = switch_case(
data_avec_traitement,
int(select_filtre_2),
float(fs_filtre_2),
float(cutoff_filtre_2),
int(order_filtre_2),
int(windowsize_filtre_2),
)
print("fin du traitement")
if(n_clicks!=None):
data_traits=switch_case(data_traits,int(selec_transforme_hilbert))
data_traits=switch_case(data_traits,int(select_filtre_1),float(fs_filtre_1),float(cutoff_filtre_1),int(order_filtre_1),int(windowsize_filtre_1))
data_traits=switch_case(data_traits,int(select_filtre_2),float(fs_filtre_2),float(cutoff_filtre_2),int(order_filtre_2),int(windowsize_filtre_2))
fig = px.line( title="A-scan")
new_trace = go.Scatter(y=data_avec_traitement, mode='lines', name=' Ascan trait ')
bouton = "Valider"
if n_clicks != None:
data_traits = switch_case(data_traits, int(selec_transforme_hilbert))
data_traits = switch_case(
data_traits,
int(select_filtre_1),
float(fs_filtre_1),
float(cutoff_filtre_1),
int(order_filtre_1),
int(windowsize_filtre_1),
)
data_traits = switch_case(
data_traits,
int(select_filtre_2),
float(fs_filtre_2),
float(cutoff_filtre_2),
int(order_filtre_2),
int(windowsize_filtre_2),
)
bouton = "Valider"
fig = px.line(title="A-scan")
new_trace = go.Scatter(y=data_avec_traitement, mode="lines", name=" Ascan trait ")
fig.add_trace(new_trace)
new_trace = go.Scatter(y=data_sans_traitement, mode='lines', name=' Ascan (hilbert) ')
new_trace = go.Scatter(
y=data_sans_traitement, mode="lines", name=" Ascan (hilbert) "
)
fig.add_trace(new_trace)
fig.update_layout(xaxis_title="indix",yaxis_title="amplitude")
data_bscan=switch_case(volume[select_ascan_y - 1, select_ascan_z[0]:select_ascan_z[1], :],int(selec_transforme_hilbert))
data_bscan=switch_case(data_bscan,int(select_filtre_1),float(fs_filtre_1),float(cutoff_filtre_1),int(order_filtre_1),int(windowsize_filtre_1))
data_bscan=switch_case(data_bscan,int(select_filtre_2),float(fs_filtre_2),float(cutoff_filtre_2),int(order_filtre_2),int(windowsize_filtre_2))
fig2 = px.imshow(data_bscan,color_continuous_scale="Jet",aspect="auto",title="B-scan ZX",)
fig2.update_layout(xaxis_title="X",yaxis_title="Z ")
data_sans_traitement_fft=np.fft.fft(volume[int(select_ascan_y)-1,select_ascan_z[0]:select_ascan_z[1],int(select_ascan_x)-1])
fig3 = px.line( title="FFT")
trace3=go.Scatter(y=np.abs(data_sans_traitement_fft),mode='lines',name=' FFT ')
fig.update_layout(xaxis_title="indix", yaxis_title="amplitude")
data_bscan = switch_case(
volume[select_ascan_y - 1, select_ascan_z[0] : select_ascan_z[1], :],
int(selec_transforme_hilbert),
)
data_bscan = switch_case(
data_bscan,
int(select_filtre_1),
float(fs_filtre_1),
float(cutoff_filtre_1),
int(order_filtre_1),
int(windowsize_filtre_1),
)
data_bscan = switch_case(
data_bscan,
int(select_filtre_2),
float(fs_filtre_2),
float(cutoff_filtre_2),
int(order_filtre_2),
int(windowsize_filtre_2),
)
fig2 = px.imshow(
data_bscan,
color_continuous_scale="Jet",
aspect="auto",
title="B-scan ZX",
)
fig2.update_layout(xaxis_title="X", yaxis_title="Z ")
data_sans_traitement_fft = np.fft.fft(
volume[
int(select_ascan_y) - 1,
select_ascan_z[0] : select_ascan_z[1],
int(select_ascan_x) - 1,
]
)
fig3 = px.line(title="FFT")
trace3 = go.Scatter(y=np.abs(data_sans_traitement_fft), mode="lines", name=" FFT ")
fig3.add_trace(trace3)
fig3.update_layout(xaxis_title="FFT indix",yaxis_title="FFT of signal (Mangnitude)")
return [fig, fig2,fig3,"Valider"]
fig3.update_layout(
xaxis_title="FFT indix", yaxis_title="FFT of signal (Mangnitude)"
)
return [fig, fig2, fig3, bouton]

43
3D_app/pages/gng.py Normal file
View File

@ -0,0 +1,43 @@
import dash
import plotly.graph_objects as go
from dash import html, dcc
from gng2 import GrowingNeuralGas
from sklearn import datasets as sk
dash.register_page(__name__, path="/gng", title="GNG", name="GNG")
# Generate synthetic data
X, _ = sk.make_moons(n_samples=200, noise=0.1)
# Create and fit the GNG model
gng = GrowingNeuralGas(input_dim=2)
gng.fit(X, num_iterations=2000)
fig = go.Figure()
for edge in gng.edges:
fig.add_trace(
go.Scatter(
x=[edge.nodes[0].position[0], edge.nodes[1].position[0]],
y=[edge.nodes[0].position[1], edge.nodes[1].position[1]],
mode="lines",
line=dict(width=2, color="black"),
)
)
for node in gng.nodes:
fig.add_trace(
go.Scatter(
x=[node.position[0]],
y=[node.position[1]],
mode="markers",
marker=dict(size=10, color="red"),
)
)
fig.update_layout(
showlegend=False,
margin=dict(l=0, r=0, t=0, b=0),
xaxis=dict(visible=False),
yaxis=dict(visible=False),
)
layout = html.Div(dcc.Graph(figure=fig))

4
3D_app/pages/home.py
View File

@ -805,10 +805,10 @@ def update_settings(
Output("layer-slider-bscan-zx", "marks"),
Output("layer-slider-bscan-xy", "max"),
Output("layer-slider-bscan-xy", "marks"),
Output("settings-spinner", "children")
Output("settings-spinner", "children"),
],
Input("store-settings", "data"),
prevent_initial_call=True
prevent_initial_call=True,
)
def redef_data(data):
global volume, dim_x, dim_y, dim_z, X, Y, Z

8
3D_app/requirements.txt Normal file
View File

@ -0,0 +1,8 @@
dash==2.17.0
dash_bootstrap_components==1.6.0
matplotlib==3.8.4
numpy==1.26.4
pandas==2.2.2
plotly==5.22.0
scikit_learn==1.5.0
scipy==1.13.1