Upload 4 files

FcgEngine/engine.py

@@ -0,0 +1,95 @@
+import os
+import json
+import onnxruntime
+import numpy as np
+import cv2
+from transformers import AutoModelForImageClassification, AutoConfig
+import torch
+
+
+class PredictorClsOnnx(object):
+    def __init__(self, model_dir, device='cpu', gpu_allocate=1):
+        model_bin = os.path.join(model_dir, 'model.bin')
+        with open(os.path.join(model_dir, 'configs.json')) as f:
+            configs = json.load(f)
+        self.ids = list(configs["ids"])
+        self.signal_size = configs["signal_len"]
+        if device == 'cuda':
+            providers = [
+                ('CUDAExecutionProvider', {
+                    'device_id': 0,
+                    'arena_extend_strategy': 'kNextPowerOfTwo',
+                    'gpu_mem_limit': gpu_allocate * 1024 * 1024 * 1024,
+                    'cudnn_conv_algo_search': 'EXHAUSTIVE',
+                    'do_copy_in_default_stream': True,
+                })
+            ]
+            session = onnxruntime.InferenceSession(model_bin, None, providers=providers)
+        else:
+            session = onnxruntime.InferenceSession(model_bin, providers=['CPUExecutionProvider'])
+        session.get_modelmeta()
+        self.input_name = session.get_inputs()[0].name
+        self.output_name = session.get_outputs()[0].name
+        self.model = session
+        self.device = device
+
+    def __call__(self, signal):
+        signal = np.expand_dims(signal, axis=0)
+        max = np.max(signal, axis=1)
+        min = np.min(signal, axis=1)
+        signal = ((signal.astype(np.float32) - min) / (max - min))
+        signal = cv2.resize(signal, (self.signal_size, 1), interpolation=cv2.INTER_CUBIC)
+        input_blob = np.expand_dims(signal, axis=0).astype(np.float32)
+        outputs = self.model.run([self.output_name], {self.input_name: input_blob})[0][0]
+        return outputs
+
+    def decode(self, result, th=0.5):
+        predict_cls = list(np.where(result >= th))[0]
+        description = "Signal contains"
+        for ids in predict_cls:
+            description += ' '
+            cls_name = self.ids[ids]
+            prob = result[ids]
+            description += '{} ({:.4f});'.format(cls_name.capitalize(), prob)
+        return description
+
+
+class PredictorCls(object):
+    def __init__(self, model_path='lycaoduong/FcgFormer', device='cpu'):
+        self.model = AutoModelForImageClassification.from_pretrained(model_path, trust_remote_code=True)
+        self.model.to(device)
+        config = AutoConfig.from_pretrained(model_path, trust_remote_code=True)
+        self.ids = list(config.cls_name.keys())
+        self.device = device
+
+    def __call__(self, spectra):
+        tensor = self.model.to_pt_tensor(spectra).to(self.device)
+        with torch.no_grad():
+            o = self.model(tensor)['logits']
+        outputs = torch.sigmoid(o).cpu().numpy()
+        return outputs[0]
+
+    def get_result(self, result, th=0.5, pos_only=False):
+        if pos_only:
+            predict_cls = list(np.where(result >= th))[0]
+        else:
+            result[result < th] = 0.0
+            predict_cls = list(np.where(result >= 0))[0]
+        fcn_groups = []
+        probabilities = []
+        for ids in predict_cls:
+            cls_name = self.ids[ids]
+            prob = result[ids]
+            fcn_groups.append(cls_name)
+            probabilities.append(prob)
+        return fcn_groups, probabilities
+
+    def decode(self, result, th=0.5):
+        predict_cls = list(np.where(result >= th))[0]
+        description = "Signal contains"
+        for ids in predict_cls:
+            description += ' '
+            cls_name = self.ids[ids]
+            prob = result[ids]
+            description += '{} ({:.4f});'.format(cls_name.capitalize(), prob)
+        return description

FcgEngine/utils.py

@@ -0,0 +1,38 @@
+import matplotlib.pyplot as plt
+import numpy as np
+
+
+def plot_self_attention_map(spectra, att_map, offset=400):
+    if att_map is not None:
+        fig, ax1 = plt.subplots(figsize=(12, 12))
+        ax2 = plt.twinx().twiny()
+        ax2.set_xlim(offset, offset+len(spectra))
+        ax1.set_xlabel('Patch index', fontsize=18)
+        ax1.set_ylabel('Patch index', fontsize=18)
+        ax1.imshow(att_map[1:, 1:], cmap='inferno', interpolation='nearest', aspect="auto")
+        # tem_x = np.zeros_like(spectra)
+        x = np.linspace(offset, offset + len(spectra), len(spectra))
+        # for i, s in enumerate(spectra):
+        #     tem_x[i] = i + 400
+        ax2.set_xlabel('Wavelength', fontsize=18)
+        ax2.set_ylabel('Intensity (a.u.)', fontsize=18)
+        ax2.plot(x, spectra)
+        return fig
+
+def plot_spectra(spectra):
+    fig = plt.figure()
+    # fg.title("Spectra Signal")
+    plt.plot(spectra)
+    return fig
+
+def plot_result(name, score):
+    fig, ax = plt.subplots(figsize=(10, 10))
+    y_pos = np.arange(len(name))
+    ax.barh(y_pos, score, align='center')
+    ax.set_yticks(y_pos, labels=name)
+    for value, pos in zip(score, y_pos):
+        ax.text(value-0.2, pos, value, ha='right')
+    ax.invert_yaxis()  # labels read top-to-bottom
+    ax.set_xlabel('Confidence')
+    ax.set_title('Prediction')
+    return fig

app.py

@@ -0,0 +1,72 @@
+import numpy as np
+import gradio as gr
+import io
+from FcgEngine.engine import PredictorCls
+from FcgEngine.utils import plot_result, plot_spectra, plot_self_attention_map
+import os
+from huggingface_hub import login
+
+
+model_path = os.getenv("API_MODEL")
+auth_token = os.getenv("API_TOKEN")
+login(token=auth_token)
+engine = PredictorCls(model_path=model_path)
+
+
+def process(array_binary, th, option):
+    spectra = np.load(io.BytesIO(array_binary[0]))
+    outputs = engine(spectra)
+    if option == 'Positive only':
+        fcn_groups, probabilities = engine.get_result(outputs, th=th, pos_only=True)
+    else:
+        fcn_groups, probabilities = engine.get_result(outputs, th=th)
+    prediction_fg = plot_result(fcn_groups, probabilities)
+
+    spectra_fg = plot_spectra(spectra)
+
+    att = engine.model.get_self_attention(layer_value=1)
+    att = np.sum(att[0], axis=0)
+    attention_fg = plot_self_attention_map(spectra, att)
+
+    return spectra_fg, prediction_fg, attention_fg
+
+
+def clear():
+    return None, None, None
+
+
+block = gr.Blocks(title="FcgFormer APP - Ohlabs")
+with block:
+    with gr.Row():
+        gr.Markdown("## Transformer based for Spectra Characterization")
+    with gr.Row():
+        with gr.Column():
+            with gr.Row():
+                spectra_input = gr.File(file_count="multiple",
+                                        label="Select File (*.npy)-signal with shape [1, n], n: signal length",
+                                        type="binary", file_types=['.npy'])
+            with gr.Row():
+                with gr.Accordion("Advanced options", open=False):
+                    threshold = gr.Slider(label="Control Confidence", minimum=0.05, maximum=0.95, value=0.5, step=0.05)
+                    draw_option = gr.Radio(["Positive only", "All classes"], value="Positive only", label="Draw Prediction option")
+            with gr.Row():
+                with gr.Column():
+                    run_button = gr.Button(value="Run", variant="primary")
+                with gr.Column():
+                    clear_button = gr.Button(value="Clear")
+
+        with gr.Column():
+            with gr.Tab("Prediction"):
+                predicted_plot = gr.Plot(label="Result")
+            with gr.Tab("Attention map"):
+                attention_plot = gr.Plot(label="Attention Map")
+            with gr.Tab("Input Spectra"):
+                signal_plot = gr.Plot(label="Spectra Signal")
+
+    ips = [spectra_input, threshold, draw_option]
+    ops = [signal_plot, predicted_plot, attention_plot]
+    run_button.click(fn=process, inputs=ips, outputs=ops)
+    clear_button.click(fn=clear, inputs=[], outputs=ops)
+
+block.launch(server_name='0.0.0.0', share=False)
+

pre-requirements.txt

@@ -0,0 +1,4 @@
+onnxruntime-gpu==1.14.1
+transformers==4.37.1
+opencv-python==4.9.0.80
+torch==2.1.2