first_push

jddb-agents/jddb/extractor/__init__.py

+from .base_extractor import BaseExtractor
\ No newline at end of file

jddb-agents/jddb/extractor/base_extractor.py

+from abc import ABC, abstractmethod
+from typing import List
+import json
+from ..processor import Step,Pipeline
+
+class BaseExtractor(ABC):
+    def __init__(self, config_file_path: str):
+        """
+        Initializes the BaseExtractor with a configuration file path.
+
+        Args:
+            config_file_path (str): Path to the configuration file.
+        """
+        self.config_file_path = config_file_path
+        self.config = self.load_config()
+
+    def load_config(self):
+        """
+        Loads the configuration file from the specified path.
+
+        Returns:
+            dict: The loaded configuration as a dictionary.
+        """
+        with open(self.config_file_path, 'r') as f:
+            return json.load(f)
+
+    @abstractmethod
+    def extract_steps(self) -> List['Step']:
+        """
+        Abstract method that must be implemented by subclasses to define the extraction steps.
+
+        Returns:
+            List[Step]: A list of Step objects representing the extraction steps.
+        """
+        pass
+
+    @abstractmethod
+    def make_pipeline(self)->Pipeline:
+        """
+        Abstract method that must be implemented by subclasses to create a specific pipeline.
+
+        Returns:
+            Pipeline: The created pipeline object.
+        """
+        pass
+

jddb-agents/jddb/extractor/readme.md

+# extarctor
+
+`extarctor` BaseExtractor is a toolkit designed for extracting features in a pipeline. The toolkit supports HDF5 files stored in a standard format and depends on `processor`.
+
+## `base_extarctor.py`
+
+`BaseExtractor` is the smallest unit of data processing, corresponding to a single signal in the data
+
+    extarctor.BaseExtractor(
+        config_file_path: str
+    )
+
+Arguments:
+
+* `config_file_path`：Path to the configuration file, which contains parameters required for feature extraction.
+
+Properties:
+
+* `config_file_path`：Stores the path of the configuration file,`str`.
+* `config`：A dictionary containing the extracted configuration settings, `dict`.
+
+Methods:
+
+* `load_config`
+    
+      load_config(self) -> dict
+
+`load_config` Loads the configuration file from the specified path.
+
+  Returns:
+  * `dict`：The loaded configuration as a dictionary.
+
+* `extract_steps`
+
+      extract_steps(self) -> List[Step]
+
+    Returns:
+    * `List[Step]`：A list of Step objects representing the extraction steps.
+
+* `make_pipeline`
+
+      make_pipeline(self) -> Pipeline
+
+     Returns:
+    * `jddb.processor.Pipeline`：This method must be implemented by subclasses to create a specific processing pipeline.
+
+Example:
+    
+    >>> from jddb.processor import extractors
+    >>>
+    >>> # make a custom extractor
+    >>> class CustomExtractor(BaseExtractor):
+    >>> def __init__(self, config_file_path: str):
+    >>>     self.config_file_path = config_file_path
+    >>>     self.config = self.load_config()
+    >>> def extract_steps(self):
+    >>>     return [Step(processor=Resample(1000), input_tags=["\\ip"], output_tags=["\\ip"])]]
+    >>> def make_pipeline(self):
+    >>>     return Pipeline(self.extract_steps())
+    >>>
+    >>> # Initialize an extractor with a config file
+    >>> extractor = CustomExtractor("config.json")
+    >>> # Loaded configuration
+    >>> extractor_config = extractor.config  
+    >>> extractor_steps = extractor.extract_steps() 
+    >>> # Created pipeline
+    >>> extractor_pipeline = extractor.make_pipeline()  

jddb-agents/jddb/file_repo/__init__.py

+from .file_repo import FileRepo

jddb-agents/jddb/file_repo/readme.md

+# How to use FileRepo 
+
+**`file_repo`** is a subpackage used to process HDF5 files.
+
+It has a class named **`FileRepo`**.
+
+## Definition of the file format
+
+Shot file should *always* named as **'shot_no.hdf5'** (such as 1050000.hdf5).
+
+Each file contain two groups, **'data'** and **'meta'**.
+
+- 'Data' group:
+
+   - Each *tag* (diagnostic raw data or processed data) should save in the **'data'** group as a dataset.
+
+   - **Attributes** belong to each dataset, and by default, they should include the sampling rate named as **'SampleRate'** and start time named as **'StartTime'** of the data for reconstructing the time axis.
+ 
+- 'Meta' group:
+
+  - Each *label* should save in the **'meta'** group as a dataset.
+  - No **attribute** belongs to the dataset in **'meta'** group. 
+  - The labels can be synced from MongoDB service to the hdf5 file using MetaDB. 
+  
+## How file is organized
+
+All the files should be under a base folder.
+
+The base folder should use $shot_x$ for template.
+
+## FileRepo
+
+- **`FileRepo(base_path)`**
+
+  base path contains template same as BM design `\\data\\jtext\\$shot_2$XX\\$shot_1$XX\\`
+  
+  Example:
+  
+  ```python
+  from jddb.file_repo import FileRepo
+  base_path = "\\data\\jtext\\$shot_2$XX\\$shot_1$XX\\"
+  file_repo = FileRepo(base_path)
+  ```
+
+### Operate files
+
+- **`get_file(shot_no, ignore_none=False)`**
+
+  Get file path for one shot, if not exist return empty string.
+  - `ignore_none` controls the return value, if `ignore_none=True`, even the shot file does exist, still return the file path, not the empty string.
+    Example:
+
+  ```python
+  shot_no = 1050000
+  file_path = file_repo.get_file(shot_no)
+  ```
+
+- **`get_all_shots()`**
+  
+  Find all shots in the base path, return a shot list.
+  
+  Example:
+  ```python
+  all_shot_list = file_repo.get_all_shot()
+  ```
+
+- **`create_shot(shot_no)`**
+
+  Create the a shot file, return the file path.
+  
+  Example:
+  
+  ```python
+  shot_no = 1050001
+  file_path = file_repo.create_shot(shot_no)
+  ```
+  
+
+- **`get_files(shot_list=None,create_empty=False)`**
+
+  Get files path for a shot list, return a dictionary `dict{'shot_no': file_path}` 
+  - `shot_list=None` If no value is assigned to `shot_list`, return all the shot list in the root path. `shot_list = file_repo.get_all_shot()`.
+  - `create_empty` controls if you want to create new file for the shot with no file exist. If `create_empty=True`, the return dictionary will contains no empty string, and an empty shot file will be created.
+
+  Example:
+  ```python
+  shot_list = [1050000, 1050001]
+  files_path = file_repo.get_files(shot_list)
+  ```
+### Read
+- **`get_tag_list(shot_no)`**
+
+  Get all the tag list of the data group in one shot file, return a tag list.
+  
+  Example:
+  ```python
+  shot_no = 1050002
+  tag_list = file_repo.get_tag_list(shot_no)
+  ```
+- **`read_data_file(file_path, tag_list=None)`**
+
+  Read data dict from the data group in one shot file with a file path as input, return a dictionary `dict{'tag': data}`.
+  - `tag_list=None` If no list is assigned to `tag_list`, return the dictionary with all the tag list in this shot file. `tag_list = file_repo.get_tag_list(shot_no)`.
+  
+  Example:
+
+  ```python
+  file_path = "\\data\\jtext\\10500XX\\105000X\\1050000.hdf5"
+  tag_list = ['\\ip', '\\bt']
+  data_dict = file_repo.read_data_file(file_path, tag_list)
+  ```
+- **`read_data(shot_no, tag_list=None)`**
+
+  Read data dict from the data group in one shot file with a shot number as input, return a dictionary `dict{'tag': data}`.
+  - `tag_list=None` If no list is assigned to `tag_list`, return the dictionary with all the tags in this data group. 
+  
+  Example:
+
+  ```python
+  shot_no = 1050000
+  tag_list = ['\\ip', '\\bt']
+  data_dict = file_repo.read_data(shot_no, tag_list)
+  ```
+- **`read_attributes(shot_no, tag, attribute_list=None)`**
+
+  Read attribute dict of one tag in one shot file, return a dictionary `dict{'SampleRate': 1000}`.
+  
+  - `attribute_list=None` If no list is assigned to `attribute_list`, return the dictionary with all the attributes in this dataset.
+  
+  Example:
+  
+  ```python
+  shot_no = 1050000
+  tag = '\\ip'
+  attribute_list = ['SampleRate', 'StartTime']
+  attribute_list = file_repo.read_attributes(shot_no, tag, attribute_list)
+  ```
+  
+- **`read_labels_file(file_path, label_list=None)`**
+  
+  Read label dict from the meta group in one shot file with a file path as input, return a dictionary `dict{'DownTime': 0.6}`.
+  
+  - `label_list=None` If no list is assigned to `label_list`, return the dictionary with all the labels in this meta group.
+  
+  Example:
+  ```python
+  file_path = "\\data\\jtext\\10500XX\\105000X\\1050000.hdf5"
+  label_list = ['DownTime', 'IsDisrupt']
+  label_dict = file_repo.read_labels_file(file_path, label_list)
+  ```
+
+- **`read_labels(shot_no, label_list=None)`**
+
+  Read label dict from the meta group in one shot file with a shot number as input, return a dictionary `dict{'DownTime': 0.6}`.
+  
+  - `label_list=None` If no list is assigned to `label_list`, return the dictionary with all the labels in this meta group.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  label_list = ['DownTime', 'IsDisrupt']
+  label_dict = file_repo.read_labels(shot_no, label_list)
+  ```
+### Remove
+
+- **`remove_data_file(file_path, tag_list)`**
+
+  Remove the datasets from the data group in one shot file with fa ile path as input, return None.
+  
+  Example:
+  ```python
+  file_path = "\\data\\jtext\\10500XX\\105000X\\1050000.hdf5"
+  tag_list = ['\\ip', '\\bt']
+  file_repo.remove_data_file(file_path, tag_list)
+  ```
+  
+- **`remove_data(shot_no, tag_list)`**
+
+  Remove the datasets from the data group in one shot file with a shot number as input, return None.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  tag_list = ['\\ip', '\\bt']
+  file_repo.remove_data(shot_no, tag_list)
+  ```
+  
+- **`remove_attribute(shot_no, tag, attribute_list)`**
+
+  Remove the attribute of one tag in one shot file, return None.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  tag = '\\ip'
+  attribute_list = ['StartTime', 'SampleRate']
+  file_repo.remove_attributes(shot_no, tag, attribute_list)
+  ```  
+  
+- **`remove_labels_file(file_path, label_list)`**
+  
+  Remove labels from the meta group in one shot file with a file path as input, return None.
+  
+  Example:
+  ```python
+  file_path = "\\data\\jtext\\10500XX\\105000X\\1050000.hdf5"
+  label_list = ['DownTime', 'IsDisrupt']
+  file_repo.remove_labels_file(file_path, label_list)
+  ```
+  
+- **`remove_labels(shot_no, label_list)`**
+
+  Remove labels from the meta group in one shot file with a shot number as input, return None.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  label_list = ['DownTime', 'IsDisrupt']
+  file_repo.remove_labels(shot_no, label_list)
+  ```
+### Write
+- **`write_data_file(file_path, data_dict, overwrite=False)`**
+
+  Write a data dictionary in the data group in one shot file with a file path as input, return None.
+  
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  Example:
+  ```python
+  file_path = "\\data\\jtext\\10500XX\\105000X\\1050000.hdf5"
+  data_dict = {'\\ip': np.random.randn(1000), '\\bt': np.random.randn(1000)}
+  file_repo.write_data_file(file_path, data_dict)
+  ```
+- **`write_data(shot_no, data_dict, overwrite=False, create_empty=True)`**
+
+  Write a data dictionary in the data group in one shot file with a shot number as input, return None.
+  
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  - `create_empty` controls if you would like to create a new shot file.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  data_dict = {'\\ip': np.random.randn(1000), '\\bt': np.random.randn(1000)}
+  file_repo.write_data(shot_no, data_dict)
+  ```  
+
+- **`write_attribute(shot_no, tag, attribute_dict, overwrite=False)`**
+
+  Write a attribute dictionary in one tag in one shot file, return None.
+  
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  tag = '\\ip'
+  attribute_dict = {'SampleRate': 1000, 'StartTime': 0.1}
+  file_repo.write_attribute(shot_no, tag, attribute_dict)
+  ```
+  
+- **`write_label_files(file_path, label_dict, overwrite=False)`**
+  
+  Write a label dictionary in the meta group in one shot file with a file path as input, return None.
+  
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  Example:
+  ```python
+  file_path = "\\data\\jtext\\10500XX\\105000X\\1050000.hdf5"
+  label_dict = {'DownTime': 0.6, 'IsDisrupt': 1} 
+  file_repo.write_label_file(file_path, label_dict)
+  ```
+  
+- **`write_label(shot_no, label_dict, overwrite=False)`**
+  
+  Write a label dictionary in the meta group in one shot file with a shot number as input, return None.
+  
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  Example:
+  ```python
+  shot_no = 1050000
+  label_dict = {'DownTime': 0.6, 'IsDisrupt': 1} 
+  file_repo.write_label_file(shot_no, label_dict)
+  ```
+### Sync Meta
+
+- **`sync_meta(meta_db: MetaDB, shot_list=None, overwrite=False)`**
+
+  Sync labels to the meta group of the shot file from MetaDB.
+  
+  - `MetaDB` a class from the module `meta_db.py`
+  - `shot_list=None` If no value is assigned to `shot_list`, return all the shot list in the root path. `shot_list = file_repo.get_all_shot()`.
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  Example:
+  ```python
+  from jddb.meta_db import MetaDB
+  shot_list = [1050000, 1050001]
+  connection_str = {
+    "host": "1.1.1.1",
+    "port": 24011,
+    "username": "User",
+    "password": "******",
+    "database": "JDDB"
+  }
+  collection = "label"
+  meta_db = MetaDB(connection_str, collection)
+  file_repo.sync_meta(meta_db, shot_list)
+  meta_db.disconnect()
+  ```  

jddb-agents/jddb/mds_dumper/__init__.py

+from .mds_dumper import MDSDumper

jddb-agents/jddb/mds_dumper/mds_dumper.py

+import warnings
+import numpy as np
+from MDSplus import connection
+from typing import List
+from ..file_repo import FileRepo
+import time as delay_time
+
+
+class MDSDumper:
+    def __init__(self, host_name: str, tree_name: str):
+        self.host_name = host_name
+        self.tree_name = tree_name
+        self.conn = None
+
+    def connect(self):
+        self.conn = connection.Connection(self.host_name)
+        return self.conn
+
+    def disconnect(self):
+        if self.conn is not None:
+            self.conn.disconnect()
+
+    def dumper(self, file_repo: FileRepo, shot_list: List[int], tag_list: List[str], overwrite=False):
+        """
+
+        Dump data from MDSPlus into the hdf5 shot file
+        Args:
+            file_repo: initialized object of FileRepo
+            shot_list: shot list
+            tag_list: tag list
+            overwrite: True -> remove the existed tag, then write the new one
+
+        Returns: None
+
+        """
+        sample_rate_attr = 'SampleRate'
+        start_time_attr = 'StartTime'
+        i = int(0)
+        while True:
+            if i > len(shot_list) - 1:
+                break
+            shot = int(shot_list[i])
+            print(r"Shot: {}".format(shot))
+            try:
+                self.conn = self.connect()
+            except ConnectionError:
+                warnings.warn("Connect Error", category=UserWarning)
+                delay_time.sleep(5)
+                warnings.warn("Delay 5s and reconnect", category=UserWarning)
+                continue
+            i += 1
+            try:
+                self.conn.openTree(self.tree_name, shot=shot)
+            except ConnectionError:
+                warnings.warn("Could not open the tree of shot {}".format(shot), category=UserWarning)
+                delay_time.sleep(5)
+                warnings.warn("Delay 5s and reconnect", category=UserWarning)
+                continue
+            file_repo.create_shot(shot)
+            exist_tag_list = file_repo.get_tag_list(shot)
+            for tag in tag_list:
+                if tag not in exist_tag_list or overwrite:
+                    try:
+                        data_raw = np.array(self.conn.get(tag))
+                        time_raw = np.array(self.conn.get(r'DIM_OF({})'.format(tag)))
+                        fs = len(time_raw) / (time_raw[-1] - time_raw[0]) if len(time_raw) > 1 else 0
+                        st = time_raw[0] if len(time_raw) > 1 else 0
+                        data_dict_temp = dict()
+                        data_dict_temp[tag] = data_raw
+                        attribute_dict = dict()
+                        attribute_dict[sample_rate_attr] = fs
+                        attribute_dict[start_time_attr] = st
+                        file_repo.write_data(shot, data_dict_temp, overwrite, create_empty=False)
+                        file_repo.write_attributes(shot, tag, attribute_dict, overwrite)
+                        file_repo.write_label(shot, {tag: True}, True)
+                        del attribute_dict
+                        del data_dict_temp
+                    except Exception:
+                        warnings.warn("Could not read data from {}".format(tag), category=UserWarning)
+                        file_repo.write_label(shot, {tag: False}, True)
+                        pass
+
+                else:
+                    continue
+            self.conn.closeAllTrees()
+            self.conn.disconnect()
+
+    def dump_tag(self, file_repo: FileRepo, shot_list: List[int], tag: str, time_tag=False, overwrite=False):
+        """
+
+            Dump data from MDSPlus into the hdf5 shot file
+                Args:
+                    file_repo: initialized object of FileRepo
+                    shot_list: shot list
+                    tag: tag
+                    time_tag: the tag of time axis
+                    overwrite: True -> remove the existed tag, then write the new one
+
+                Returns: None
+
+        """
+        if not time_tag:
+            time_tag = r'DIM_OF({})'.format(tag)
+        sample_rate_attr = 'SampleRate'
+        start_time_attr = 'StartTime'
+        i = int(0)
+        while True:
+            if i > len(shot_list) - 1:
+                break
+            shot = int(shot_list[i])
+            print(r"Shot: {}".format(shot))
+            try:
+                self.conn = self.connect()
+            except ConnectionError:
+                warnings.warn("Connect Error", category=UserWarning)
+                delay_time.sleep(5)
+                warnings.warn("Delay 5s and reconnect", category=UserWarning)
+                continue
+            i += 1
+            try:
+                self.conn.openTree(self.tree_name, shot=shot)
+            except ConnectionError:
+                warnings.warn("Could not open the tree of shot {}".format(shot), category=UserWarning)
+                delay_time.sleep(5)
+                warnings.warn("Delay 5s and reconnect", category=UserWarning)
+                continue
+            file_repo.create_shot(shot)
+            exist_tag_list = file_repo.get_tag_list(shot)
+            if tag not in exist_tag_list or overwrite:
+                try:
+                    data_raw = np.array(self.conn.get(tag))
+                    time_raw = np.array(self.conn.get(time_tag))
+                    fs = len(time_raw) / (time_raw[-1] - time_raw[0]) if len(time_raw) > 1 else 0
+                    st = time_raw[0] if len(time_raw) > 1 else 0
+                    data_dict_temp = dict()
+                    data_dict_temp[tag] = data_raw
+                    attribute_dict = dict()
+                    attribute_dict[sample_rate_attr] = fs
+                    attribute_dict[start_time_attr] = st
+                    file_repo.write_data(shot, data_dict_temp, overwrite, create_empty=False)
+                    file_repo.write_attributes(shot, tag, attribute_dict, overwrite)
+                    file_repo.write_label(shot, {tag: True}, True)
+                    del attribute_dict
+                    del data_dict_temp
+                except Exception:
+                    warnings.warn("Could not read data from {}".format(tag), category=UserWarning)
+                    file_repo.write_label(shot, {tag: False}, True)
+                    pass
+
+            else:
+                continue

jddb-agents/jddb/mds_dumper/readme.md

+# How to use MDSDupmer 
+
+**`mds_dumper`** is a subpackage used to dump data from MDSplus (only support MDSpuls now).
+
+It has a class named **`MDSDumper`**.
+
+YOU MUST install **`MDSplus`** package FIRST
+
+## MDSDumper
+
+- **`MDSDumper(host_name, tree_name)`**
+  
+  Example:
+  ```python
+  from jddb.file_repo import MDSDumper
+  dump = MDSDumper('1.1.1.1', 'jtext')
+  ```
+### Functions
+- **`connect`**
+  
+  Connect to the host
+
+- **`disconnect`**
+
+  Disconnect to the host
+
+- **`dumper(file_repo: FileRepo, shot_list, tag_list, overwrite=False)`**
+
+  Dump data from MDSPlus into the hdf5 shot file.
+  
+  - `FileRepo` a class from the module `file_repo.py`
+  - `overwrite` controls if the input data will overwrite the original data.
+  
+  Example:
+  ```python
+  from jddb.file_repo import FileRepo
+  from jddb.mds_dumper import MDSDumper
+  base_path = "\\data\\jtext\\$shot_2$XX\\$shot_1$XX\\"
+  file_repo = FileRepo(base_path)
+  shot_list = [1050000, 1050001]
+  tag_list = ['\\ip', '\\bt']
+  dump = MDSDumper('1.1.1.1', 'jtext')
+  dump.dumper(file_repo, shot_list, tag_list) 
+  ```

jddb-agents/jddb/meta_db/__init__.py

+from .meta_db import MetaDB

jddb-agents/jddb/meta_db/readme.md

+# How to use MetaDB  
+## **MetaDB**
+  Get meta or query eligible information from MetaDB.
+  ```python
+    from jddb import meta_db
+    connection_str = {
+              "host" : "localhost",
+              "port" : 27017,
+              "username" : "DDBUser",
+              "password" : "*******",
+              "database": "JDDB"
+            }
+    collection = "Labels"
+    db = meta_db.MetaDB(connection_str, collection)                  
+  ```
+- ### **db.connect(connection_str, collection)**
+    **Description:**  
+      Connect to MetaDB befor any other action.  
+
+    **Parameters:**  
+      connection_string : Dictionary. The connections string to a mongodb server. Such as "host", "port", "username", "password" and so on.  
+      collection : String. Collection name for the MetaDB.  
+    **Return:**  
+      Collection of the MetaDB
+    
+- ### **db.disconnect()**
+    **Description:**  
+      Disconnect from MetaDB after any other action.
+
+    **Example :**
+    ```python
+    # Connect to the MetaDB
+    from jddb import meta_db
+    connection_str = {
+              "host" : "localhost",
+              "port" : 27017,
+              "username" : "DDBUser",
+              "password" : "*******",
+              "database": "JDDB"
+            }
+    collection = "Labels"
+    db = meta_db.MetaDB()
+    labels = db.connect(connection_str,collection)
+
+    # Disconnect from MetaDB
+    db.disconnect()
+    ```
+- ### **db.get_labels(shot_no)**  
+  **Description:** 
+    Get all meta of the input shot.   
+  **Parameters:**  
+    shot_no : int or string. The shot number whose meta you want to get.  
+  **Return:**  
+    Dictionary. The meta you want.
+
+  **Example :**
+    ```python
+    db.get_labels(1066648)
+
+    -Return:
+    {'shot': 1066648, 'ip': True, 'IsDisrupt': False, 'DownTime': 0.5923408076837159, 'bt': True, ... 'MA_TOR1_R09': True}
+    ```
+- ### **db.update_labels(shot_no, labels)**
+  **Description:**  
+    Update or modify the meta of a shot in MetaDB.  
+  **Parameters:**  
+    shot_no : int or string. The shot number whose meta you want to update or modify.  
+    labels : Dictionary. The meta contents you want to update or modify. 
+
+  **Example :**
+    ```python
+    new_meta = {'ip': True, 'IsDisrupt': False, 'DownTime': 0.59234, ...}
+    db.update_labels(1066648, new_meta)
+    ```
+
+- ### **db.query(shot_list=None, filter=None)**
+  **Description:**  
+    Query the shots that meet the filter conditions within the given shot number range.  
+  **Parameters:**  
+    shot_list : List. The queried range of shot numbers. If shot_list=None, query all shots in the MetaDB.  
+    filter : Dictionary. The filter condition for the query. The description format of the condition must comply with Mongodb's specifications, and specific details can be found on the official website of Mongodb. If filter=None, Return all shot number in MetaDB.  
+  **Return:**  
+    List. Shot numbers that meet the filter condition.  
+  **Example :**
+    ```python
+    my_query = {'IsDisrupt': True, 'IpFlat':{'$gt':50}}
+    db.query(my_query)
+
+    -Return:
+    [1046770, 1046794, 1046795, 1046806, 1046800 1046858, . . . , 1049184, 1050467, 1052286, 1050560, 1052295]
+    ```
+
+
+- ### **db.query_valid(shot_list=None, label_true=None, label_false=None)**
+  **Description:**  
+    For labels whose information stored in the database is True or False, return shot numbers that meet the filter condition.  
+  **Parameters:**  
+    shot_list : List. The range of shot numbers queried. If shot_list=None, query all shots in the MetaDB.  
+    label_true : List of label names. Filter condition. The returned shots must satisfy that all labels in the label_true are True.  
+    label_false : List of label names. Filter condition. The returned shots must satisfy that all labels in the label_false are False.  
+  **Return:**  
+    List. Shot number that meets the filter condition.  
+  **Example :**  
+    Get non-disruption shots with [" ip", " bt"] diagnostics available in the shot number range of [1064000, 1066649]
+    ```python
+    shot_list = [shot for shot in range(1064000, 1066649+1)]
+    db.query_valid(shot_list=shot_list, label_true=["ip", "bt"], label_false=["IsDisrupt"])
+
+    -Return:
+    [1064000, 1064001, 1064002, 1064003, 1064004,  .... 1066642, 1066643, 1066644, 1066646, 1066648]
+    ```
+
+- ### **db.query_range(label_list, lower_limit=None, upper_limit=None, shot_list=None)**  
+  **Description:**  
+    For labels with numeric values stored in the database, return shot numbers that meet the filter condition.  
+  **Parameters:**  
+    label_list : A list of labels that store information as numeric values.   
+    lower_limit : List of lower limit values. ">=". If there is only an upper limit and no lower limit in the filter condition, the lower limit value in the corresponding position in the lower limit list is set to None.   
+    upper_limit : List of upper limit values. "<=". If there is only an lower limit and no upper limit in the filter condition, the upper limit value in the corresponding position in the upper limit list is set to None.    
+    shot_list : List. The range of shot numbers queried. If shot_list=None, query all shots in the MetaDB.    
+  **Return:**  
+    List. Shot number that meets the filter condition.  
+  **Example :**  
+    Get shots with DownTime in [0.2, 0.8]s and average flat-top value of IP in [150, inf]KA within the shot number range of [1064000, 1066648].
+    ```python
+    shot_list = [shot for shot in range(1064000, 10666489)]
+    db.query_range(["DownTime","IpFlat"], lower_limit=[0.2,150], upper_limit=[0.8, None], shot_list=shot_list)
+
+    -Return:
+    [1064000, 1064001, 1064002, 1064003, 1064004,  .... 1066642, 1066643, 1066644, 1066646, 1066648]
+    ```
+
+- ### **db.count_label(shot_list, label_list, need_nd=False, show=True)**
+  **Description:**   
+    Count and display the number of shots available in the given shot number list for each given diagnosis; Finally, count and display the number of shots available for all given diagnostic signals in the given shot number list, as well as the number of non-disruption and disruption shots. Returns the list of available shots for diagnosis.  
+  **Parameters:**  
+    shot_list : List. The queried range of shot numbers.  
+    label_list : List. The queried label names.  
+    need_nd : Bool. Whether to divide the returned shots into disruption shots and non-disruption shots, and return the disruption shots and non-disruption shots. The default is False, and only the list of shots available for all diagnostics is returned; When need_nd=True, the list of shots available for all diagnostics, as well as a list of non-disruption shots and a list of disruption shots are returned.  
+    show : Bool. Whether to display statistical results. The default is True, which displays the number of shots available for each given diagnostic in the given shot number list, the number of shots available for all given diagnostic signals, and the number of non-disruption and disruption shots. If show=False, don't displayed.    
+  **Return:**  
+    When need_nd=False, only return one list, which is the list of shots available for all given diagnostic signals.  
+    When need_nd=True, return three lists, which are the list of shots available for all given diagnostic signals, list of non-disruption shots, and list of disruption shots.  
+  **Example :**  
+    ```python
+    shot_list = [shot for shot in range(1064000, 10666489)]
+    labels = ["ip", "bt", "polaris_den_v09"]
+    db.count_label(shot_list, labels)
+    ```
+    Show:
+    ![2](./plot/1.png)  
+    Returns a list of shots available for all given diagnosis, a list of non-disruption shots, and a list of disruption shots.  
+    ```python
+    CompleteShots, NormalShots, DisruptShots = db.count_label(shot_list, labels, need_nd=True, show=False)
+    ```
\ No newline at end of file

jddb-agents/jddb/performance/__init__.py

+from .result import Result
+from .report import Report

jddb-agents/jddb/performance/report.py

+from typing import List
+import pandas as pd
+import os
+import numpy as np
+from matplotlib import pyplot as plt
+from sklearn.metrics import auc
+import matplotlib.font_manager as fm
+
+class Report:
+    """Assign a str value to the header"""
+    MODEL_NAME = 'model_name'
+    ACCURACY = 'accuracy'
+    PRECISION = 'precision'
+    RECALL = 'recall'
+    FPR = 'fpr'
+    TPR = 'tpr'
+    TP = 'tp'
+    FN = 'fn'
+    FP = 'fp'
+    TN = 'tn'
+
+    def __init__(self, report_csv_path: str):
+
+        self.report_csv_path = report_csv_path
+        """        
+            check if file exist
+            if not create df,set threshold nan
+            if exists, read it populate self
+        """
+
+        self.__header = [self.MODEL_NAME, self.ACCURACY, self.PRECISION, self.RECALL, self.FPR, self.TPR, self.TP,
+                         self.FN, self.FP, self.TN]
+        self.report = None
+        if os.path.exists(self.report_csv_path):
+            self.read()
+        else:
+            self.report = pd.DataFrame(columns=self.__header)
+
+    def read(self):
+        """
+            check file format
+            read in all module
+        """
+
+        self.report = pd.read_csv(self.report_csv_path)
+        if self.report.columns is None:
+            self.report = pd.DataFrame(columns=self.__header)
+        elif len(set(self.report.columns) & set(self.__header)) != len(self.report.columns):
+            raise ValueError("The file from csv_path:{} contains unknown information ".format(self.report_csv_path))
+
+    def save(self):
+        """
+            save report in disk
+        """
+        self.report.to_csv(self.report_csv_path, index=False)
+
+    def add(self, result, model_name: str):
+        """
+            add new model
+        Args:
+            result: a csv file that is instantiated
+            model_name: a name of str
+
+        """
+
+        result.calc_metrics()
+        tpr = result.tpr
+        fpr = result.fpr
+        accuracy = result.accuracy
+        precision = result.precision
+        recall = result.recall
+        matrix = result.confusion_matrix
+        tn = int(matrix[0][0])
+        fp = int(matrix[0][1])
+        fn = int(matrix[1][0])
+        tp = int(matrix[1][1])
+
+        if model_name in self.report[self.MODEL_NAME].values:
+            # update the row with matching key
+            self.report.loc[self.report[self.MODEL_NAME] == model_name, [self.MODEL_NAME, self.ACCURACY, self.PRECISION,
+                                                                         self.RECALL,
+                                                                         self.FPR, self.TPR, self.TP, self.FN, self.FP,
+                                                                         self.TN]] = \
+                [model_name, accuracy, precision, recall, fpr, tpr, tp, fn, fp, tn]
+        else:
+            # insert new row
+            new_row = {self.MODEL_NAME: model_name, self.ACCURACY: accuracy, self.PRECISION: precision,
+                       self.RECALL: recall,
+                       self.FPR: fpr, self.TPR: tpr, self.TP: tp, self.FN: fn, self.FP: fp, self.TN: tn}
+            self.report = self.report.append(new_row, ignore_index=True)
+
+    def remove(self, model_name: List[str]):
+        """
+            giving model_name to remove the specified row
+        Args:
+            model_name: a list of model
+        """
+        if model_name in self.report[self.MODEL_NAME].values:
+            for i in range(len(model_name)):
+                self.report = self.report.drop(self.report[self.report[self.MODEL_NAME] == model_name[i]].index)
+        else:
+            pass
+
+    def plot_roc(self, roc_file_path=None):
+        """
+            draw roc curve
+        Args:
+            roc_file_path: a path to save picture
+
+        """
+
+        fpr = self.report[self.FPR].tolist()
+        tpr = self.report[self.TPR].tolist()
+        tpr_ordered = []
+        index = np.array(fpr).argsort()
+        for i in index:
+            tpr_ordered.append(tpr[i])
+        fpr.sort()
+        roc_auc = auc(fpr, tpr_ordered)
+        lw = 2
+        fig = plt.figure()
+        fig.tight_layout()
+        plt.rcParams["figure.figsize"] = (10, 10)
+        plt.xlabel('False Positive Rate', fontsize='xx-large')
+        plt.ylabel('True Positive Rate', fontsize='xx-large')
+        plt.plot(np.array(fpr), np.array(tpr_ordered), color='darkorange',
+                 lw=lw, label='ROC curve (area = %0.2f)' % roc_auc)
+        plt.plot([0, 1], [0, 1], color='navy', lw=lw, linestyle='--')
+        plt.xlim([0.0, 1.0])
+        plt.ylim([0.0, 1.0])
+        plt.title('Receiver operating characteristic', fontsize='xx-large')
+        font_props = fm.FontProperties(size='x-large', weight='bold')
+        plt.legend(loc="lower right", prop=font_props)
+        if roc_file_path:
+            plt.savefig(os.path.join(roc_file_path, 'Receiver_operating_characteristic.png'), dpi=300)
+        plt.show(block=True)

jddb-agents/jddb/processor/__init__.py

+from .signal import Signal
+from .shot import Shot
+from .shot_set import ShotSet
+from .base_processor import BaseProcessor
+from .pipeline import Step, Pipeline
\ No newline at end of file

jddb-agents/jddb/processor/base_processor.py

+from abc import ABCMeta, abstractmethod
+from .signal import Signal
+from typing import Tuple, Union, Dict
+
+
+class BaseProcessor(metaclass=ABCMeta):
+    """Process signal(s).
+
+    Process signals according to the ``transform()`` method.
+
+    """
+
+    @abstractmethod
+    def __init__(self, **kwargs):
+        self._init_args = kwargs
+        self.params = dict()
+
+    @abstractmethod
+    def transform(self, *signal: Signal) -> Union[Signal, Tuple[Signal, ...]]:
+        """Process signal(s).
+
+        This is used for subclassed processors.
+        To use it, override the method.
+
+        Args:
+            signal: signal(s) to be processed.
+
+        Returns:
+            Signal: new signal(s) processed.
+        """
+        raise NotImplementedError()
+
+    def to_config(self) -> Dict:
+        """Save instance information to a dictionary.
+
+        Returns:
+            A dictionary that contains type, init arguments and params of the instance.
+        """
+        return {
+            'type': self.__class__.__name__,
+            'init_args': self._init_args,
+            'param': self.params
+        }
+
+    @classmethod
+    def from_config(cls, processor_config: Dict):
+        """Instantiate a processor from given config.
+
+        Args:
+            processor_config: A dictionary that contains init_args (necessary), type and param of the processor instance.
+        Raises:
+            KeyError: if 'init_args' is not in processor_config keys.
+        Returns:
+            Instantiated processor.
+        """
+        if 'init_args' not in processor_config.keys():
+            raise KeyError('Config loaded does not contain \'init_args\'.')
+        else:
+            init_args = processor_config.get('init_args')
+            processor = cls(**init_args)
+
+        return processor
\ No newline at end of file

jddb-agents/jddb/processor/basic_processors/__init__.py

+from .resampling_processor import ResamplingProcessor
+from .normalization_processor import NormalizationProcessor
+from .trim_processor import TrimProcessor
+from .clip_processor import ClipProcessor

jddb-agents/jddb/processor/basic_processors/clip_processor.py

+from .. import BaseProcessor, Signal
+from copy import deepcopy
+import numpy as np
+
+
+class ClipProcessor(BaseProcessor):
+    def __init__(self, start_time: float, end_time: float = None, end_time_label: str = None):
+        super().__init__(start_time=start_time, end_time=end_time, end_time_label=end_time_label)
+        self._start_time = start_time
+        self._end_time_label = end_time_label
+        self._end_time = end_time
+
+    def transform(self, signal: Signal) -> Signal:
+        """Clip the given signal to given start time and end time given by the processor initialization.
+
+        Args:
+            signal: the signal to be clipped.
+
+        Returns: Signal: the signal clipped.
+
+        """
+        if self._end_time_label:
+            # self._end_time = self.params[self._end_time_label]
+            end_time = signal.parent.labels[self._end_time_label]
+        elif self._end_time:
+            # self._end_time = signal.time[-1]
+            end_time = self._end_time
+        else:
+            end_time = signal.time[-1]
+        if self._start_time > end_time:
+            raise ValueError('Down time is earlier than start time.')
+        clipped_data = signal.data[(self._start_time <= signal.time) & (signal.time <= end_time)]
+        clipped_attributes = deepcopy(signal.attributes)
+        start_time_idx = np.argmax(signal.time >= self._start_time)
+        clipped_attributes['StartTime'] = signal.time[start_time_idx]
+
+        return Signal(data=clipped_data, attributes=clipped_attributes, parent=signal.parent)

jddb-agents/jddb/processor/basic_processors/normalization_processor.py

+from .. import BaseProcessor, Signal
+import numpy as np
+
+
+class NormalizationProcessor(BaseProcessor):
+    def __init__(self, std: float, mean: float):
+        super().__init__(std=std, mean=mean)
+        self._std = std
+        self._mean = mean
+
+    def transform(self, signal: Signal) -> Signal:
+        """Compute the z-score.
+        Compute the z-score of the given signal according to the param_dict given by the processor initialization.
+
+        Note:
+            The result of the normalized to signal will be clipped to [-10, 10] if beyond the range.
+
+        Args:
+            signal: The signal to be normalized.
+
+        Returns: Signal: The normalized signal.
+        """
+        normalized_data = (signal.data - self._mean) / self._std
+        normalized_data = np.clip(normalized_data, -10, 10)
+
+        return Signal(data=normalized_data, attributes=signal.attributes, parent=signal.parent)

jddb-agents/jddb/processor/basic_processors/resampling_processor.py

+from .. import BaseProcessor, Signal
+from scipy.interpolate import interp1d
+from copy import deepcopy
+import numpy as np
+
+
+class ResamplingProcessor(BaseProcessor):
+    def __init__(self, sample_rate: float):
+        super().__init__(sample_rate=sample_rate)
+        self._sample_rate = sample_rate
+
+    def transform(self, signal: Signal) -> Signal:
+        """Resample signals according to the new sample rate given by the processor initialization.
+
+        Args:
+            signal: The signal to be resampled.
+
+        Returns: Signal: The resampled signal.
+
+        """
+        start_time = signal.time[0]
+        end_time = signal.time[-1]
+        new_end_time = start_time + int((end_time - start_time) * self._sample_rate) / self._sample_rate
+        # new time axis should include start time point
+        new_time = np.linspace(start_time, new_end_time,
+                               int((new_end_time - start_time) * self._sample_rate) + 1)
+        f = interp1d(signal.time, signal.data)
+        resampled_attributes = deepcopy(signal.attributes)
+        resampled_attributes['SampleRate'] = self._sample_rate
+
+        return Signal(data=f(new_time), attributes=resampled_attributes, parent=signal.parent)

jddb-agents/jddb/processor/basic_processors/trim_processor.py

+from .. import BaseProcessor, Signal
+from typing import Tuple
+
+
+class TrimProcessor(BaseProcessor):
+    def __init__(self):
+        super().__init__()
+
+    def transform(self, *signal: Signal) -> Tuple[Signal, ...]:
+        """Trim all signals to the same length with the shortest one.
+
+        Args:
+            *signal: The signals to be trimmed.
+
+        Returns: Tuple[Signal, ...]: The signals trimmed.
+        """
+        lengths = [len(each_signal.data) for each_signal in signal]
+        min_length = min(lengths)
+        for each_signal in signal:
+            each_signal.data = each_signal.data[:min_length]
+        return signal

jddb-agents/jddb/processor/pipeline.py

+from datetime import datetime
+from ..processor import Shot, ShotSet, BaseProcessor, Signal
+from ..file_repo import FileRepo
+import yaml
+from typing import List, Union, Dict
+import logging
+from functools import partial
+import platform
+import multiprocessing as mp
+from multiprocessing import Queue
+from logging.handlers import QueueHandler, QueueListener
+from logging import FileHandler
+
+class Step:
+    def __init__(self, processor: BaseProcessor, input_tags: List[Union[str, List[str]]],
+                 output_tags: List[Union[str, List[str]]]):
+        assert len(input_tags) == len(output_tags), "Lengths of input tags and output tags do not match."
+        self.processor = processor
+        self.input_tags = input_tags
+        self.output_tags = output_tags
+
+    def execute(self, shot: Shot):
+        self.processor.params.update(shot.labels)
+        self.processor.params.update({"Shot": shot.shot_no})
+
+        for i_tag, o_tag in zip(self.input_tags, self.output_tags):
+            if isinstance(i_tag, str):
+                new_signal = self.processor.transform(shot.get_signal(i_tag))
+            else:
+                new_signal = self.processor.transform(*[shot.get_signal(each_tag) for each_tag in i_tag])
+
+            if isinstance(o_tag, str) and isinstance(new_signal, Signal):
+                shot.update_signal(o_tag, new_signal)
+
+            elif isinstance(o_tag, list) and isinstance(new_signal, tuple):
+                if len(o_tag) != len(new_signal):
+                    raise ValueError("Lengths of output tags and signals do not match!")
+                for idx, each_signal in enumerate(new_signal):
+                    shot.update_signal(o_tag[idx], each_signal)
+            else:
+                raise ValueError("Lengths of output tags and signals do not match!")
+
+        return shot
+
+    def to_config(self) -> Dict:
+        step_config = {
+            'processor_type': type(self.processor).__name__,
+            'input_tags': self.input_tags,
+            'output_tags': self.output_tags,
+            'params': self.processor.params,
+            'init_args': self.processor.to_config().get('init_args')
+        }
+        return step_config
+
+    def to_yaml(self, filepath: str = 'step_config.yaml'):
+        config = self.to_config()
+        with open(filepath, 'w') as file:
+            yaml.dump(config, file)
+
+    @classmethod
+    def from_config(cls, step_config: Dict, processor_registry: Dict):
+        processor_type = step_config['processor_type']
+        if processor_type not in processor_registry:
+            raise ValueError(f"Processor type '{processor_type}' not found in the registry.")
+
+        # Dynamically initialize processor using saved arguments
+        processor_cls = processor_registry[processor_type]
+        init_args = step_config.get('init_args')
+        print(processor_cls)
+        print(init_args)
+        processor = processor_cls(**init_args)
+        processor.params = step_config.get('params')
+
+        return cls(processor, step_config['input_tags'], step_config['output_tags'])
+
+    @classmethod
+    def from_yaml(cls, filepath: str, processor_registry: Dict):
+        with open(filepath, 'r') as file:
+            step_config = yaml.safe_load(file)
+
+        return cls.from_config(step_config=step_config, processor_registry=processor_registry)
+
+class Pipeline:
+    def __init__(self, steps: List[Step] = None):
+        self.steps = steps if steps else list()
+
+    def add_step(self, step: Step):
+        self.steps.append(step)
+
+    def to_config(self) -> Dict:
+        pipeline_config = dict()
+        for i, step in enumerate(self.steps):
+            pipeline_config[f'Step_{i + 1}'] = step.to_config()
+        return pipeline_config
+
+    def to_yaml(self, filepath: str = 'pipeline_config.yaml'):
+        pipeline_config = self.to_config()
+        with open(filepath, 'w') as file:
+            yaml.dump(pipeline_config, file)
+
+    @classmethod
+    def from_config(cls, pipeline_config: Dict, processor_registry: Dict):
+        steps = list()
+        for i in range(len(pipeline_config)):
+            steps.append(Step.from_config(pipeline_config[f'Step_{i + 1}'], processor_registry))
+        return cls(steps=steps)
+
+    @classmethod
+    def from_yaml(cls, filepath: str, processor_registry: Dict):
+        with open(filepath, 'r') as file:
+            pipeline_config = yaml.safe_load(file)
+
+        return cls.from_config(pipeline_config=pipeline_config, processor_registry=processor_registry)
+
+    def process_by_shot(self, shot: Shot, save_repo: FileRepo = None, save_updated_only: bool = False):
+        logger = logging.getLogger('single_process')
+        logger.setLevel(logging.ERROR)
+        if not logger.hasHandlers():
+            file_handler = logging.FileHandler('./process_exceptions.log', mode="a")
+            file_handler.setFormatter(logging.Formatter("%(asctime)s - %(levelname)s - %(message)s"))
+            logger.addHandler(file_handler)
+
+        for i, step in enumerate(self.steps):
+            try:
+                step.execute(shot)
+            except Exception as e:
+                error_message = (
+                    f"Shot No: {shot.shot_no}\n"
+                    f"Pipeline Step: {i + 1}\n"
+                    f"Processor Type: {type(step.processor).__name__}\n"
+                    f"Input Signals: {step.input_tags}\n"
+                    f"Output Signals: {step.output_tags}\n"
+                    f"Error Message: {e}\n"
+                )
+                logger.error(msg=error_message, exc_info=None)
+
+        shot.save(save_repo, save_updated_only=save_updated_only)
+
+    def process_by_shotset(self, shotset: ShotSet, processes: int = mp.cpu_count() - 1,
+                           shot_filter: List[int] = None, save_repo: FileRepo = None, save_updated_only: bool = False):
+        """Process all shots in a ShotSet."""
+        if shot_filter is None:
+            shot_filter = shotset.shot_list
+        if processes:
+            if platform.system() == 'Linux':
+                pool = mp.get_context('spawn').Pool(processes=processes)
+            else:
+                pool = mp.Pool(processes=processes)
+            logger = logging.getLogger('multi_process')
+            logger.setLevel(logging.ERROR)
+            queue = mp.Manager().Queue(-1)
+            queue_handler = QueueHandler(queue)
+            logger.addHandler(queue_handler)
+            listener = QueueListener(
+                queue,
+                FileHandler('./process_exceptions.log', mode="a")
+            )
+            listener.start()
+            pool.map(partial(self._subprocess_task_in_shotset, queue=queue, shot_set=shotset, save_repo=save_repo,
+                             save_updated_only=save_updated_only), shot_filter)
+            pool.close()
+            pool.join()
+            listener.stop()
+        else:
+            for shot_no in shot_filter:
+                shot = shotset.get_shot(shot_no)
+                self.process_by_shot(shot, save_repo=save_repo)
+                print("Shot No: ", shot_no, " is processed.")
+
+    def _subprocess_task_in_shotset(self, shot_no: int, queue: Queue, shot_set: ShotSet, save_repo: FileRepo = None,
+                                    save_updated_only: bool = False):
+        single_shot = shot_set.get_shot(shot_no)
+        time_info = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        for i, step in enumerate(self.steps):
+            try:
+                step.execute(single_shot)
+            except Exception as e:
+                error_message = (
+                    f"Time: {time_info}\n"
+                    f"Shot No: {shot_no}\n"
+                    f"Pipeline Step: {i + 1}\n"
+                    f"Processor Type: {type(step.processor).__name__}\n"
+                    f"Input Signals: {step.input_tags}\n"
+                    f"Output Signals: {step.output_tags}\n"
+                    f"Error Message: {e}\n"
+                )
+                queue.put(
+                    logging.LogRecord('multi_process', logging.ERROR, '', 0, error_message, exc_info=None, args=None))
+
+        single_shot.save(save_repo=save_repo, save_updated_only=save_updated_only)
+        print("Shot No: ", shot_no, " is processed.")
+

jddb-agents/jddb/processor/shot.py

+import os
+
+from ..file_repo import FileRepo
+from .signal import Signal
+from typing import Union, List
+from .base_processor import BaseProcessor
+
+
+class Shot(object):
+    """Shot object.
+
+    An HDF5 file stored in the file repo.
+
+    Args:
+        shot_no (int): the number of the shot.
+        file_repo (FileRepo): the file repo the shot belongs to.
+
+    Attributes:
+        labels (dict): the label dictionary of the shot.
+    """
+    def __init__(self, shot_no: int, file_repo: FileRepo):
+        self._file_repo = file_repo
+        self._shot_no = shot_no
+        self.__new_signals = dict()
+        try:
+            self.__original_tags = file_repo.get_tag_list(self.shot_no)
+        except ValueError:
+            self.__original_tags = list()
+        try:
+            self.labels = file_repo.read_labels(self.shot_no)
+        except ValueError:
+            self.labels = dict()
+        except KeyError:
+            self.labels = dict()
+
+    @property
+    def tags(self):
+        """set: tags of ALL signals in the shot, including newly added and modified. Readonly."""
+        return set().union(self.__original_tags, self.__new_signals.keys())
+
+    @property
+    def shot_no(self):
+        """int: the number of the shot. Readonly."""
+        return self._shot_no
+
+    @property
+    def file_repo(self):
+        """FileRepo: the file repo the shot belongs to. Readonly."""
+        return self._file_repo
+
+    def update_signal(self, tag: str, signal: Signal):
+        """Add a new signal or modify an existing signal to the shot.
+
+        The method DOES NOT change any file until save() is called.
+
+        Args:
+            tag (str): name of the signal to be added or modified.
+            signal (Signal): the signal to be added or modified.
+        """
+        signal.parent = self
+        self.__new_signals[tag] = signal
+
+    def remove_signal(self, tags: Union[str, List[str]], keep: bool = False):
+        """Remove (or keep) existing signals from the shot.
+        The method DOES NOT change any file until save() is called.
+
+        Args:
+            tags (Union[str, List[str]]): name(s) of the signal(s) to be removed (or kept)
+            keep (bool): whether to remove the signals or not. Default False.
+        Raises:
+            ValueError: if any of the signal(s) is not found in the shot.
+        """
+        if isinstance(tags, str):
+            tags = [tags]
+        tags = set(tags)
+        for tag in tags:
+            if tag not in self.tags:
+                raise ValueError("{} is not found in data.".format(tag))
+        if keep:
+            drop_tags = self.tags.difference(tags)
+        else:
+            drop_tags = tags
+        for tag in drop_tags:
+            if tag in self.__original_tags:
+                self.__original_tags.remove(tag)
+            if tag in self.__new_signals.keys():
+                del self.__new_signals[tag]
+
+    def get_signal(self, tag: str) -> Signal:
+        """Get an existing signal from the shot.
+
+        Args:
+            tag (str): name of the signal to be got.
+        Returns:
+            Signal: the signal got.
+        Raises:
+            ValueError: if the signal is not found in the shot.
+        """
+        if tag in self.__new_signals.keys():
+            return self.__new_signals[tag]
+        elif tag in self.__original_tags:
+            return Signal(data=self.file_repo.read_data(self.shot_no, [tag])[tag],
+                          attributes=self.file_repo.read_attributes(self.shot_no, tag),
+                          parent=self)
+        else:
+            raise ValueError("{} is not found in data.".format(tag))
+
+    def process(self, processor: BaseProcessor, input_tags: List[Union[str, List[str]]],
+                output_tags: List[Union[str, List[str]]]):
+        """Process one (or multiple) signals of the shot.
+
+        Apply transformation to the signal(s) according to the processor.
+        The element of input/output tags can be a string or a list of strings.
+        The method DOES NOT change any file until save() is called.
+
+        Note: Each element of the input and output MUST correspond to each other.
+
+        Args:
+            processor (BaseProcessor): an instance of a subclassed BaseProcessor. The calculation is overrided in transform().
+            input_tags (List[Union[str, List[str]]]): input tag(s) to be processed.
+            output_tags (List[Union[str, List[str]]]): output tag(s) to be processed.
+        Raises:
+            ValueError: if lengths of input tags and output tags do not match.
+            ValueError: if lengths of output signals and output tags do not match.
+        """
+        if len(input_tags) != len(output_tags):
+            raise ValueError("Lengths of input tags and output tags do not match.")
+        processor.params.update(self.labels)
+        processor.params.update({"Shot": self.shot_no})
+
+        for i_tag, o_tag in zip(input_tags, output_tags):
+            if isinstance(i_tag, str):
+                new_signal = processor.transform(self.get_signal(i_tag))
+            else:
+                new_signal = processor.transform(*[self.get_signal(each_tag) for each_tag in i_tag])
+
+            if isinstance(o_tag, str) and isinstance(new_signal, Signal):
+                self.update_signal(o_tag, new_signal)
+
+            elif isinstance(o_tag, list) and isinstance(new_signal, tuple):
+                if len(o_tag) != len(new_signal):
+                    raise ValueError("Lengths of output tags and signals do not match!")
+                for idx, each_signal in enumerate(new_signal):
+                    self.update_signal(o_tag[idx], each_signal)
+            else:
+                raise ValueError("Lengths of output tags and signals do not match!")
+
+    def save(self, save_repo: FileRepo = None, data_type=None, save_updated_only: bool = False):
+        """Save the shot to specified file repo.
+
+        Save all changes done before to disk space.
+        Note: if the file repo give is None or shares the same base path with the origin file repo,
+        changes will COVER the origin file. Please CHECK the new file repo to save.
+
+        Args:
+            save_repo (FileRepo): file repo specified to save the shot. Default None.
+            data_type: control the type of writen data
+            save_updated_only (bool): whether to save updated signals only. Default False.
+        """
+        if save_updated_only:
+            tags_to_save = self.__new_signals.keys()
+        else:
+            tags_to_save = self.tags
+        if save_repo is not None and (save_repo.base_path != self.file_repo.base_path):
+            output_path = save_repo.get_file(self.shot_no)
+            if output_path == "":
+                output_path = save_repo.create_shot(self.shot_no)
+            else:
+                os.remove(output_path)
+                output_path = save_repo.create_shot(self.shot_no)
+
+            data_dict = dict()
+            for tag in tags_to_save:
+                signal = self.get_signal(tag)
+                data_dict[tag] = signal.data
+            save_repo.write_data_file(output_path, data_dict, overwrite=True, data_type=data_type)
+            for tag in tags_to_save:
+                save_repo.write_attributes(self.shot_no, tag, self.get_signal(tag).attributes, overwrite=True)
+            save_repo.write_label_file(output_path, self.labels, overwrite=True)
+
+        else:
+            existing_tags = self.file_repo.get_tag_list(self.shot_no)
+            if save_updated_only:
+                tags_to_remove = [r_tag for r_tag in existing_tags if r_tag not in self.__new_signals.keys()]
+            else:
+                tags_to_remove = [r_tag for r_tag in existing_tags if r_tag not in self.tags]
+
+            if len(tags_to_remove):
+                self.file_repo.remove_data(self.shot_no, tags_to_remove)
+            data_dict = dict()
+
+            for tag, signal in self.__new_signals.items():
+                data_dict[tag] = signal.data
+            self.file_repo.write_data(self.shot_no, data_dict, overwrite=True, data_type=data_type)
+            for tag, signal in self.__new_signals.items():
+                self.file_repo.write_attributes(self.shot_no, tag, signal.attributes, overwrite=True)
+            self.file_repo.write_label(self.shot_no, self.labels, overwrite=True)

jddb-agents/jddb/processor/shot_set.py

+from __future__ import annotations
+from ..file_repo import FileRepo
+from .base_processor import BaseProcessor
+from .shot import Shot
+from typing import Union, List
+import multiprocessing as mp
+from itertools import repeat
+import platform
+import logging
+from logging.handlers import QueueHandler, QueueListener
+from multiprocessing import Queue
+
+
+class ShotSet(object):
+    """ShotSet object.
+    A subset of file repo. Support pipeline of data processing.
+
+    Args:
+        file_repo (FileRepo): the file repo the shot set belongs to.
+        shot_list (List[int]): shot list within the shot set.
+    """
+    def __init__(self, file_repo: FileRepo, shot_list: List[int] = None):
+        self._file_repo = file_repo
+        if shot_list is None:
+            self._shot_list = self._file_repo.get_all_shots()
+        else:
+            self._shot_list = shot_list
+
+    @property
+    def file_repo(self):
+        """FileRepo: the file repo the shot set belongs to. Readonly."""
+        return self._file_repo
+
+    @property
+    def shot_list(self):
+        """list (int): shot list within the shot set. Readonly."""
+        return self._shot_list
+
+    def get_shot(self, shot_no: int) -> Shot:
+        """Get an instance of Shot of given shot_no.
+
+        Args:
+            shot_no (int): the shot to be got.
+        Returns:
+            Shot: the instance got.
+        """
+        return Shot(shot_no, self.file_repo)
+
+    def remove_signal(self, tags: List[str], shot_filter: List[int] = None, keep: bool = False, save_repo: FileRepo = None) -> ShotSet:
+        """Remove (or keep) existing signal(s) from the shots within the shot filter.
+
+        The changes removed WILL be saved immediately by calling this method.
+
+        Args:
+            tags (List[str]): tags of signals to be removed (or kept).
+            shot_filter (List[int]): shot files to be processed within the shot set. If None, process all shots within the shot set. Default None.
+            keep (bool): whether to keep the tags or not. Default False.
+            save_repo (FileRepo): file repo specified to save the shots. Default None.
+        Returns:
+            ShotSet: a new instance (or the previous instance itself) according to the base path of save_repo.
+        """
+        if shot_filter is None:
+            shot_filter = self.shot_list
+        for each_shot in shot_filter:
+            shot = Shot(each_shot, self.file_repo)
+            shot.remove_signal(tags, keep)
+            shot.save(save_repo)
+        if save_repo is None and shot_filter == self.shot_list:
+            return self
+        else:
+            return ShotSet(save_repo, shot_filter)
+
+    def process(self, processor: BaseProcessor, input_tags: List[Union[str, List[str]]], output_tags: List[Union[str, List[str]]],
+                shot_filter: List[int] = None, save_repo: FileRepo = None, processes: int = mp.cpu_count()) -> ShotSet:
+        """Process one (or several) signal(s) of the shots within the shot filter.
+
+        The changes WILL be saved immediately by calling this method.
+
+        Args:
+            processor (BaseProcessor): an instance of a subclassed BaseProcessor.
+            input_tags (List[Union[str, List[str]]]): input tag(s) to be processed.
+            output_tags (List[Union[str, List[str]]]): output tag(s) to be processed.
+            shot_filter (List[int]): shot files to be processed within the shot set. If None, process all shots within the shot set. Default None.
+            save_repo (FileRepo): file repo specified to save the shots. Default None.
+            processes (int): the number of processes. Default logical CPU counts. If processes is 0, multiprocessing is not used.
+        Returns:
+            ShotSet: a new instance (or the previous instance itself) according to the base path of save_repo.
+        """
+        if len(input_tags) != len(output_tags):
+            raise ValueError("Lengths of input tags and output tags do not match.")
+        if shot_filter is None:
+            shot_filter = self.shot_list
+
+        if processes:
+            if platform.system() == 'Linux':
+                pool = mp.get_context('spawn').Pool(processes)
+            else:
+                pool = mp.Pool(processes)
+
+            queue = mp.Manager().Queue(-1)
+            handler = QueueHandler(queue)
+            handler.setLevel(logging.ERROR)
+            listener = QueueListener(queue, logging.FileHandler('process_exceptions.log'))
+            listener.start()
+
+            pool.starmap(
+                self._parallel_task,
+                zip(repeat(queue), shot_filter, repeat(processor), repeat(input_tags), repeat(output_tags), repeat(save_repo))
+            )
+            pool.close()
+            pool.join()
+            listener.stop()
+
+        else:
+            for each_shot in shot_filter:
+                try:
+                    shot = self.get_shot(each_shot)
+                    shot.process(processor, input_tags, output_tags)
+                    shot.save(save_repo)
+                except Exception as e:
+                    error_message = f'Shot No: {each_shot} \nProcessor Type: {type(processor)} \nError Message: {e}'
+                    logging.basicConfig(filename='process_exceptions.log',
+                                        filemode='a',
+                                        format=error_message,
+                                        level=logging.ERROR)
+
+        if save_repo is None and shot_filter == self.shot_list:
+            return self
+        else:
+            return ShotSet(save_repo, shot_filter)
+
+    def _parallel_task(self, queue: Queue, shot_no: int, processor: BaseProcessor, input_tags: list,
+                       output_tags: list, save_repo: FileRepo):
+        try:
+            shot = self.get_shot(shot_no)
+            shot.process(processor, input_tags, output_tags)
+            shot.save(save_repo)
+        except Exception as e:
+            error_message = f'Shot No: {shot_no} \nProcessor Type: {type(processor)} \nError Message: {e}'
+            queue.put(logging.LogRecord('process', logging.ERROR, '', 0, error_message, None, None))
+

jddb-agents/jddb/processor/signal.py

+import numpy as np
+import warnings
+
+
+class Signal(object):
+    """Signal object.
+
+    Stores data and necessary parameters of a signal.
+
+    Attributes:
+        data (numpy.ndarray): raw data of the Signal instance.
+        attributes (dict): necessary attributes of the Signal instance. Note that SampleRate and StartTime will be set default if not given.
+        tag (str): name of the signal. Default None
+        parent (Shot): the shot which the signal belongs to. Default None.
+    """
+
+    def __init__(self, data: np.ndarray, attributes: dict, tag: str = None, parent = None, dim:np.ndarray = None):
+        self.data = data
+        self.attributes = attributes
+        self.tag = tag
+        self.parent = parent
+
+        if 'SampleRate' not in self.attributes.keys():
+            self.attributes['SampleRate'] = 1
+            warnings.warn("SampleRate not in attributes. Set to 1.")
+
+        if 'StartTime' not in self.attributes.keys():
+            self.attributes['StartTime'] = 0
+            warnings.warn("StartTime not in attributes. Set to 0.")
+
+        if dim is not None:
+            self.attributes['Dim'] = dim
+
+    @property
+    def time(self):
+        """
+        Returns:
+            numpy.ndarray: time axis of the Signal instance according to SampleRate and StartTime given in the attributes.
+        """
+        if 'Dim' in self.attributes.keys():
+            warnings.warn('This signal may be unevenly sampled.')
+            return self.attributes['Dim']
+        else:
+            start_time = self.attributes['StartTime']
+            sample_rate = self.attributes['SampleRate']
+            down_time = start_time + (len(self.data) - 1) / sample_rate
+            return np.linspace(start_time, down_time, len(self.data))

jddb-agents/jddb/utils.py

+import re
+import os
+
+
+def replace_pattern(directory: str, shot: int, include_filename: bool = False) -> str:
+    """Replace the '$shot_*$' pattern to normal directory to save shots.
+
+    Args:
+        directory (str): origin input directory that may cover the pattern.
+        shot (int): the shot number to save.
+        include_filename: use file name as the base_path
+    Returns:
+        str: replaced directory to save shots.
+    """
+    pattern = r'\$shot_\d+\$'
+    match = re.findall(pattern, directory)
+    if match:
+        for each in match:
+            number = 10 ** int(re.findall(r'\d+', each)[0])
+            directory = directory.replace(each, '{}'.format(int(shot) // number))
+    if include_filename:
+        directory = directory.replace(r'$shot$', '{}'.format(shot))
+        file_path = directory
+    else:
+        file_path = os.path.join(directory, '{}.hdf5'.format(shot))
+    return file_path

jddb-agents/jddb_skills/processor_generator/SKILL.md

+# Processor Generator Skill
+
+## Description
+
+Wrap user-provided code, algorithms, or basic computational logic into a Processor class following `jddb.processor` conventions, enabling seamless integration with the jddb framework.
+
+---
+
+## Basic Processor Structure
+
+Every Processor must inherit from `BaseProcessor` and implement the `transform` method:
+
+```python
+from jddb.processor import BaseProcessor, Signal
+from copy import deepcopy
+import numpy as np
+
+class YourProcessor(BaseProcessor):
+    def __init__(self, param1, param2, ...):
+        """
+        Initialize the Processor with required parameters.
+        
+        Args:
+            param1: Description of param1
+            param2: Description of param2
+        """
+        super().__init__(param1=param1, param2=param2, ...)
+        # Optional: store parameters in self.params for later use
+        self.params.update({"Param1": param1, "Param2": param2})
+
+    def transform(self, signal: Signal) -> Signal:
+        """
+        Process the input signal and return a new signal.
+        
+        Args:
+            signal: Input signal containing data (numpy.ndarray), attributes (dict), and time (1D array)
+        
+        Returns:
+            Signal: The processed new signal
+        """
+        new_signal = deepcopy(signal)
+        # Implement your algorithm/logic here
+        new_signal.data = ...  # Processed data
+        new_signal.attributes['SampleRate'] = ...  # Update if changed
+        new_signal.attributes['StartTime'] = ...   # Update if changed
+        return new_signal
+```
+
+---
+
+## Key Points
+
+1. **Inherit from `BaseProcessor`**: All Processors must inherit from `jddb.processor.BaseProcessor`.
+
+2. **`__init__` method**:
+   - Must call `super().__init__(**kwargs)` and pass all initialization parameters to the parent class.
+   - Optionally store parameters in `self.params` dict for configuration saving and loading.
+
+3. **`transform` method**:
+   - Input: One or more `Signal` objects.
+   - Output: One or more `Signal` objects.
+   - Use `deepcopy(signal)` to create a copy of the signal, avoiding modification of the original.
+   - If sample rate or start time changes, update `SampleRate` and `StartTime` in `attributes`.
+
+4. **Signal structure**:
+   - `signal.data`: numpy.ndarray, the raw data.
+   - `signal.attributes`: dict, must contain `StartTime` (seconds) and `SampleRate` (Hz).
+   - `signal.time`: 1D numpy.ndarray, time axis automatically calculated from data length, StartTime, and SampleRate.
+
+---
+
+## Prompt Template for Generating Processors
+
+When users provide an algorithm/code/logic, use the following prompt to have the LLM generate a jddb-compliant Processor:
+
+```
+Please wrap the following algorithm/code/logic into a Processor class that conforms to jddb.processor conventions.
+
+Requirements:
+1. Inherit from `jddb.processor.BaseProcessor`
+2. Call `super().__init__(**kwargs)` in `__init__` and pass all initialization parameters
+3. Implement `transform(self, signal: Signal) -> Signal` method
+4. Use `deepcopy(signal)` to create a copy of the signal
+5. If sample rate or start time changes, update `new_signal.attributes['SampleRate']` and `new_signal.attributes['StartTime']`
+6. Add complete docstrings
+
+User-provided algorithm/code/logic:
+{user_input}
+
+Please output complete Python code including necessary import statements.
+```
+
+---
+
+## Examples
+
+### Example 1: Multiply Processor
+
+User requirement: Multiply signal data by a constant.
+
+```python
+from jddb.processor import BaseProcessor, Signal
+from copy import deepcopy
+
+class MultiplyProcessor(BaseProcessor):
+    """Multiply signal data by a specified factor."""
+    
+    def __init__(self, multiplier: float):
+        """
+        Args:
+            multiplier: The multiplication factor
+        """
+        super().__init__(multiplier=multiplier)
+        self.multiplier = multiplier
+
+    def transform(self, signal: Signal) -> Signal:
+        """
+        Args:
+            signal: Input signal
+        Returns:
+            Signal: New signal with data multiplied by multiplier
+        """
+        new_signal = deepcopy(signal)
+        new_signal.data = new_signal.data * self.multiplier
+        return new_signal
+```
+
+### Example 2: Sliding Window Slice Processor
+
+User requirement: Slice signal by specified window length and overlap rate.
+
+```python
+from jddb.processor import BaseProcessor, Signal
+from copy import deepcopy
+import numpy as np
+
+class SliceProcessor(BaseProcessor):
+    """Slice signal into windows, returning a sequence of windows."""
+    
+    def __init__(self, window_length: int, overlap: float):
+        """
+        Args:
+            window_length: Window length (number of points)
+            overlap: Overlap rate, between 0 and 1
+        """
+        super().__init__(window_length=window_length, overlap=overlap)
+        assert 0 <= overlap <= 1, "Overlap must be between 0 and 1."
+        self.params.update({"WindowLength": window_length, "Overlap": overlap})
+
+    def transform(self, signal: Signal) -> Signal:
+        window_length = self.params["WindowLength"]
+        overlap = self.params["Overlap"]
+        new_signal = deepcopy(signal)
+        raw_sample_rate = new_signal.attributes["SampleRate"]
+        step = int(window_length * (1 - overlap))
+
+        idx = len(signal.data)
+        windows = []
+        while (idx - window_length) >= 0:
+            windows.append(new_signal.data[idx - window_length:idx])
+            idx -= step
+        windows.reverse()
+
+        new_signal.data = np.array(windows)
+        new_signal.attributes['SampleRate'] = raw_sample_rate / step
+        new_signal.attributes['StartTime'] = signal.time[-1] - (len(windows) - 1) / new_signal.attributes['SampleRate']
+        new_signal.attributes['OriginalSampleRate'] = raw_sample_rate
+        return new_signal
+```
+
+### Example 3: Multi-Input Multi-Output Processor
+
+User requirement: Receive multiple signals and trim them to the same length.
+
+```python
+from jddb.processor import BaseProcessor, Signal
+from typing import Tuple
+from copy import deepcopy
+
+class TrimProcessor(BaseProcessor):
+    """Trim multiple signals to the same length."""
+    
+    def __init__(self):
+        super().__init__()
+
+    def transform(self, *signals: Signal) -> Tuple[Signal, ...]:
+        """
+        Args:
+            signals: Multiple input signals
+        Returns:
+            Tuple[Signal, ...]: Tuple of signals trimmed to the same length
+        """
+        min_len = min(len(s.data) for s in signals)
+        result = []
+        for signal in signals:
+            new_signal = deepcopy(signal)
+            new_signal.data = new_signal.data[:min_len]
+            result.append(new_signal)
+        return tuple(result)
+```
+
+---
+
+## Using a Processor
+
+```python
+from jddb.processor import Shot, ShotSet
+from jddb.file_repo import FileRepo
+
+# Single Shot processing
+file_repo = FileRepo('path/to/hdf5')
+shot = Shot(1000000, file_repo)
+shot.process(YourProcessor(param1, param2), input_tags=['\\signal1'], output_tags=['\\signal1_processed'])
+shot.save()
+
+# ShotSet batch processing
+shot_set = ShotSet(file_repo)
+shot_set.process(YourProcessor(param1, param2), input_tags=['\\signal1'], output_tags=['\\signal1_processed'])
+```
+
+---
+
+## Applicable Domains
+
+- Magnetic confinement fusion data processing
+- Plasma physics signal analysis
+- Time series data batch processing

jddb-agents/tm_extractor/.gitignore

+__pycache__/
+.idea/
+final_plt_path/
\ No newline at end of file

jddb-agents/tm_extractor/README.md

+# tm_extractor
+
+`tm_extractor` is a command-line tool for extracting tearing mode (TM) features from diagnostic signals. It supports configuration through JSON or YAML and can process input data to extract the presence, frequency, amplitude, and coupling characteristics of the tearing mode.
+
+## Installation
+
+Make sure all necessary dependencies are installed. If you are using tm_extractor as a Python package, follow these steps:
+
+First, navigate to the tm_extractor directory, where the `setup.py` file is located, using the cd command in your terminal:
+
+```bash
+cd /path/to/tm_extractor
+```
+
+Then, run the following command to install the package:
+
+```bash
+pip install .
+```
+This will install tm_extractor and its dependencies in the current Python environment. After installation, you can use tm_extractor as a command-line tool.
+
+## Usage
+
+Run the following command in the terminal (replace the path with your actual data location):
+
+```bash
+    tm_extractor --from_json_or_yaml "from json" \
+                 --input_file_path "example\raw_data" \
+                 --output_file_path "example\tearing_mode_data" \
+                 --processes 1
+```
+
+### Command-Line Arguments
+
+| Argument                 | Type   | Required | Default Value            | Description                                      |
+|--------------------------|--------|----------|--------------------------|--------------------------------------------------|
+| `--from_json_or_yaml`    | `str`  | ✅ Yes   | `"from json"`            | Specify the configuration format: `"from json"` or `"from yaml"` |
+| `--from_json_to_yaml`    | `bool` | ❌ No    | `True`                   | Whether to convert JSON configuration to YAML    |
+| `--json_file_path`       | `str`  | ❌ No    | `"default_path"`         | Path to the JSON configuration file              |
+| `--to_yaml_file_path`    | `str`  | ❌ No    | `"pipeline_config.yaml"` | Path to save the converted YAML configuration   |
+| `--input_file_path`      | `str`  | ✅ Yes   | `"example_shotset"`      | Path to the input data                          |
+| `--output_file_path`     | `str`  | ✅ Yes   | `"save_shotset"`         | Path to save the processed output data           |
+| `--final_plt_path`       | `str`  | ❌ No    | `"mode_amp_plt"`         | Path to save processed plots                     |
+| `--processes`            | `int`  | ❌ No    | `0`                      | Number of processes for parallel execution       |
+| `--save_updated_only`    | `bool` | ❌ No    | `False`                  | Save only updated results                        |
+| `--shot_filter_config`   | `str`  | ❌ No    | `"all_shot"`             | Configuration to filter shots                    |
+| `--ext_name`             | `str`  | ❌ No    | `"tm_extractor"`         | Name of the extractor                            |
+
+## Workflow
+
+1. **Load Configuration**  
+   - If `--from_json_or_yaml "from json"`, load configuration from `default_config_json.json` or the specified JSON file.
+   - If `--from_json_or_yaml "from yaml"`, load configuration from `pipeline_config.yaml`.
+
+2. **Initialize Extractor and Processing Pipeline**  
+   - Create a `TMExtractor` instance from the configuration file.
+   - Generate the processing pipeline.
+
+3. **Process Shotset Data**  
+   - Load the input data files.
+   - Run the processing pipeline to extract tearing mode information and save the results.
+
+4. **Save Processing Results**  
+   - If `--from_json_to_yaml` is enabled, convert and save the JSON configuration as YAML.
+   - The processed data is saved in the directory specified by `--output_file_path`.
+   - The path for saving the tearing mode extraction plot, specified by `--final_plt_path`. If not specified, the plot will be saved in the current directory under the default name `mode_amp_plt`.
+   - the exceptions are saved in the cuurent directory named `process_exceptions.log`.

jddb-agents/tm_extractor/custom_processor/.gitignore

+*.pyc
+__pycache__/

jddb-agents/tm_extractor/custom_processor/__init__.py

+from.amp_judge_processor import AmpJudgeProcessor
+from .amp_is_tearing_processor import AmpIsTearingProcessor
+from .couple_mode_processor import CoupleModeProcessor
+from .divertor_safety_factor_processor import DivertorSafetyFactorProcessor
+from .double_judge_processor import DoubleJudgeProcessor
+from .four_b_fft_processor import FourBFFTProcessor
+from .inte_processor import InteProcessor
+from .is_tm_processor import IsTMProcessor
+from .lowpass_processor import LowPassProcessor
+from .m_mode_csd_phase import M_mode_csd_phase
+from .m_mode_th_processor import M_mode_th
+from .mn_mode_union import MNModeUnionProcessor
+from .mode_fre_check_processor import ModeFreCheck
+from .mode_fre_noisy_check_processor import ModeFreNoisyCheck
+from .n_mode_csd_phase_processor import N_mode_csd_phase
+from .n_phase_mode_union_processor import NPhaseModeUnionProcessor
+from .odd_even_couple_judge_processor import OddEvenCoupleJudgeProcessor
+from .odd_even_processor import EvenOddProcessor
+from .plot_mode_amp_processor import PlotModeAmpProcessor
+# from .plot_shot_processor import PlotShotProcessor
+from .reorder_processor import ReOrderProcessor
+from .slice_processor import SliceProcessor
+from .slice_trim_processor import SliceTrimProcessor
+from .small_mode_processor import SmallModeProcessor
+from .specific_mode_processor import SpecificModeExtractProcessor
+from .split_amp_fre_processor import SplitAmpFreProcessor

jddb-agents/tm_extractor/custom_processor/amp_is_tearing_processor.py

+# -*-  coding:utf-8  -*-
+import numpy as np
+from jddb.processor import Signal, BaseProcessor
+from copy import deepcopy
+
+
+class AmpIsTearingProcessor(BaseProcessor):
+    def __init__(self, amp_threshold_ratio=0.5, f_upper_threshold=800):
+        super().__init__(amp_threshold_ratio=amp_threshold_ratio, f_upper_threshold=f_upper_threshold)
+        self.threshold_ratio =  amp_threshold_ratio
+        self.f_upper_threshold= f_upper_threshold
+    def transform(self, signal: Signal):
+        # signal[1]: m_most_max_all
+        B_LFS_Inte_slice_signal = deepcopy(signal)
+        is_tearing_signal = deepcopy(signal)
+        recover, is_tearing = self.is_tearing_threshold(B_LFS_Inte_slice_signal.data)
+        # recover_data,is_tearing_near_time_data = self.is_tearing_near_time( recover_data, f_upper=800)
+        B_LFS_Inte_slice_signal.data=recover
+        is_tearing_signal.data=is_tearing
+        return B_LFS_Inte_slice_signal,is_tearing_signal
+
+    def is_tearing_threshold(self, B_LFS_Inte_slice_data):
+        """
+        判断撕裂是否发生
+        :param B_LFS_Inte_slice: B_LFS_Inte_slice
+        :return: is_tearing: 是否发生撕裂
+        """
+        is_tearing = np.zeros(len(B_LFS_Inte_slice_data))
+        recover = deepcopy(B_LFS_Inte_slice_data)
+        for i in range(len(B_LFS_Inte_slice_data)):
+            if B_LFS_Inte_slice_data[i][1] >= self.threshold_ratio and B_LFS_Inte_slice_data[i][2] > self.f_upper_threshold :
+                is_tearing[i] = int(1)
+            elif B_LFS_Inte_slice_data[i][1] ==0 and B_LFS_Inte_slice_data[i][2] == 0:
+                is_tearing[i] = -1
+            else:
+                is_tearing[i] = int(0)
+                recover[i] = np.array([0,0,0])
+
+        return recover,is_tearing
\ No newline at end of file

jddb-agents/tm_extractor/custom_processor/amp_judge_processor.py

+# -*-  coding:utf-8  -*-
+import numpy as np
+from scipy.fftpack import fft
+import math
+from typing import Tuple, Union
+from jddb.processor import Signal, BaseProcessor
+from typing import Optional, List
+from scipy.fftpack import fft
+from copy import deepcopy
+
+class AmpJudgeProcessor(BaseProcessor):
+    """
+       给定频率,判断B_tehta是否大于阈值,如果大于阈值,则认为是撕裂模,如果不是撕裂模，其频率标注为-1
+    """
+
+    def __init__(self, threshold=2):  # pad 表示f精度，精度为 1/(fs*pad),f 的长度为 pad*fs/2
+        super().__init__(threshold=threshold)
+        self.threshold=threshold
+    def transform(self, *signal: Signal):
+        # #signal[0]:B_LFS_Inte_slice
+        # #signal[1]:m_most_max_all
+        # #signal[2]:m_sec_max_all
+        # #signal[3]:m_third_max_all
+        # #signal[4]:m_forth_max_all
+        mode_fre_signal = deepcopy((signal.__getitem__(0)))
+        m_most_all_signal = deepcopy((signal.__getitem__(0)))
+        m_sec_all_signal = deepcopy((signal.__getitem__(1)))
+        m_third_all_signal = deepcopy((signal.__getitem__(2)))
+        m_forth_all_signal = deepcopy((signal.__getitem__(3)))
+        # n_m_th_signal = [deepcopy((signal.__getitem__(i)) for i in range(len(signal)))]
+        mode_fre = np.zeros(shape=(len(mode_fre_signal.data), 4), dtype=float)
+        mode_fre[:,0], m_most_all_signal.data,  = self.amp_judge(m_most_all_signal.data)
+        mode_fre[:,1], m_sec_all_signal.data = self.amp_judge(m_sec_all_signal.data)
+        mode_fre[:,2], m_third_all_signal.data = self.amp_judge(m_third_all_signal.data)
+        mode_fre[:,3], m_forth_all_signal.data = self.amp_judge(m_forth_all_signal.data)
+        mode_fre_signal.data=mode_fre
+        return mode_fre_signal, m_most_all_signal, m_sec_all_signal, m_third_all_signal, m_forth_all_signal
+
+    def amp_judge(self, m_data):#根据幅值判断是否计入撕裂模
+        lfs_amp_fre_cover = np.zeros(shape=(len(m_data), 8), dtype=float)
+        mode_fre = np.zeros(shape=(len(m_data)), dtype=float)
+        for i in range(len(m_data)):
+            if m_data[i][1]>=self.threshold:
+                lfs_amp_fre_cover[i] = np.concatenate([np.array([m_data[i][0]]),np.array([1]),m_data[i][1:]])
+                mode_fre[i]=m_data[i][2]
+            elif 0<m_data[i][1]<self.threshold:#这个地方其实不会到0，到0.5这个阈值0.5<=data<2Gs
+                lfs_amp_fre_cover[i] = np.concatenate([np.array([m_data[i][0]]),np.array([0]),m_data[i][1:]] )
+                # mode_fre[i]=0
+            else:
+                mode_fre[i]=-1#
+                lfs_amp_fre_cover[i][1] = -1
+        return mode_fre, lfs_amp_fre_cover
\ No newline at end of file

jddb-agents/tm_extractor/custom_processor/divertor_safety_factor_processor.py

+#  -*-  coding: utf-8  -*-
+from jddb.processor import Signal, Shot, ShotSet, BaseProcessor
+import numpy as np
+from copy import deepcopy
+
+class DivertorSafetyFactorProcessor(BaseProcessor):
+    "适用于Jtext的绘制撕裂模幅值,在23秋季与24春季minor_radius_a=0.22, major_radius_R=1.05"
+    def __init__(self,minor_radius_a=0.22, major_radius_R=1.05):
+        super().__init__(minor_radius_a=minor_radius_a, major_radius_a=major_radius_R)
+        self.MU_0 = 4 * np.pi * 10 ** (-7)
+        self.minor_radius_a=minor_radius_a
+        self.major_radius_R =major_radius_R
+    def transform(self,  *signal: Signal):
+        # #signal[0]:Ip
+        # #signal[1]:qa#在这一步计算一次qa_raw
+        ip_signal = deepcopy(signal.__getitem__(0))
+        bt_signal = deepcopy(signal.__getitem__(1))
+        qa_signal = deepcopy(signal.__getitem__(1))
+        K = 2 * np.pi * self.minor_radius_a ** 2 / (self.MU_0 * 1000 * self.major_radius_R)
+        qa_signal.data=K * bt_signal.data / ip_signal.data
+        # if mir_raw_signal.parent.labels["ERTUsed"]==1:
+        return qa_signal
+

jddb-agents/tm_extractor/custom_processor/double_judge_processor.py

+# -*-  coding:utf-8  -*-
+import numpy as np
+from jddb.processor import Signal, BaseProcessor
+from copy import deepcopy
+
+class DoubleJudgeProcessor(BaseProcessor):
+    """
+       判断是否是倍频/非倍频模式
+           # 判断是否为倍频模式 # 那么只要被放入了 mode_double_fre就不可以再跟其他比较了
+    # A B C D四种频率
+    # A B为倍频，则只取A mode_undouble_fre(1, 0, *, *)  mode_double_fre(0, 1, *, *)
+    # A B不为倍频 则取 A B 到mode_undouble_fre(1, 1, *, *) mode_double_fre(0, 0, *, *)
+    # B为倍频
+    # A C为倍频，则只取A mode_undouble_fre(1, 0, 0, *) B mode_double_fre(0, 1, 1, *)
+    # A C不为倍频 则取 A C 到mode_undouble_fre(1, 0, 1, *) mode_double_fre(0, 1, 0, *)
+    # B 不为倍频
+    # A C为倍频，则只取A mode_undouble_fre(1, 0, 0, *) B mode_double_fre(0, 1, 1, *)
+    # A C不为倍频 则取 A C 到mode_undouble_fre(1, 0, 1, *) mode_double_fre(0, 1, 0, *)
+    """
+
+    def __init__(self, fre_diff_threshold=1000, dpi=100):  # （fre +-400/fre_base） =有一个是整数就认为是倍频
+        super().__init__(fre_diff_threshold=fre_diff_threshold, dpi=dpi)
+        self.fre_diff_threshold=fre_diff_threshold
+        self.dpi = dpi
+    def transform(self, signal: Signal):
+        mode_fre_signal = deepcopy(signal)
+        double_fre_data = np.zeros(shape=(len(mode_fre_signal.data), 4), dtype=float)
+        undouble_fre_data = np.zeros(shape=(len(mode_fre_signal.data), 4), dtype=float)
+        index=0
+        for slice_index in  range(len(mode_fre_signal.data)):#412,426
+            mode_fre_signal.data[slice_index][mode_fre_signal.data[slice_index] <=0] = np.inf
+            double_fre_data[index], undouble_fre_data[index] = self.judge_fre_double(mode_fre_signal.data[slice_index])
+            index+=1
+        mode_fre_signal.data = undouble_fre_data
+        return mode_fre_signal
+
+    def sort_with_indices(self, arr):
+        # 对数组按从小到大排序，同时返回排序后的元素和它们的原始索引
+        return sorted(enumerate(arr), key=lambda x: x[1])
+
+    # 对数组进行排序并返回排序元素及原始索引
+    def double_judge(self, m_amp, m_amp_later):  # 根据幅值判断是否计入撕裂模
+        is_double = False
+        # if m_amp_later==np.inf:
+        m_amp_arr = self.generate_sequence(m_amp=m_amp, step=self.dpi,
+                                      count=int(self.fre_diff_threshold / self.dpi) *2 + 1)
+        m_amp_later_arr = self.generate_sequence(m_amp=m_amp_later, step=self.dpi,
+                                            count=int(self.fre_diff_threshold / self.dpi) * 2 + 1)
+        if self.check_divisible(arr1=m_amp_arr, arr2=m_amp_later_arr):
+            is_double = True
+
+        return is_double
+
+    def check_divisible(self, arr1, arr2):
+        # 遍历 arr1 和 arr2 中的每个元素，检查是否存在相除为整数的情况
+        for a in arr1:
+            for b in arr2:
+                if b != 0 and b % a == 0:  # 避免除以 0
+                    return True  # 一旦找到，立即返回 True
+        return False  # 如果遍历结束仍未找到，返回 False
+
+    def generate_sequence(self, m_amp, step, count):
+        # 生成以 m_amp 为中心，间隔为 step 的等差数列
+        start = m_amp - step * (count // 2)  # 计算起始值
+        return [start + i * step for i in range(count)]
+
+    def process_array(self, undouble_fre,double_fre,mode_fre_i):
+        # 找到值为 1 的元素的索引
+        undouble_index = np.argmax(undouble_fre)  # 适合仅一个 1 的情况
+
+        # 如果索引不为 0，修改数组
+        if undouble_index != 0 and mode_fre_i[0]!=0 and mode_fre_i[0]!=np.inf:
+            undouble_fre[undouble_index] = 0
+            undouble_fre[0] = 1
+            double_fre[undouble_index] = 1
+            double_fre[0] = 0
+
+        # 返回索引和修改后的数组
+        return undouble_fre,double_fre
+
+    def judge_fre_double(self, mode_fre_data_slice):
+        processed_double_fre = np.zeros(4, dtype=float)
+        processed_undouble_fre = np.zeros(4, dtype=float)
+        inf_fre = np.zeros(4, dtype=float)
+        inf_fre[mode_fre_data_slice == np.inf] = 1#mode fre为-1的即<2Gs的情况 为mode_fre_data_slice == np.inf
+        processed_undouble_fre[0] = 1
+        sorted_with_indices = self.sort_with_indices(mode_fre_data_slice)
+        sorted_indices = [index for index, value in sorted_with_indices]
+        sorted_mode_fre = np.array([mode_fre_data_slice[index] for index in sorted_indices])
+        sorted_inf_fre = np.array([inf_fre[index] for index in sorted_indices])
+        for i in range(4):
+            if processed_double_fre[i] != 1:
+                for j in range(4):
+                    if processed_double_fre[j] != 1 and i != j and sorted_inf_fre[i]!=1 and sorted_inf_fre[j]!=1:#mode fre为-1的即<2Gs的情况 为1
+                        is_double = self.double_judge(sorted_mode_fre[i], sorted_mode_fre[j])
+                        if is_double:
+                            processed_double_fre[j] = 1
+                            processed_undouble_fre[j] = 0
+                        else:
+                            processed_double_fre[j] = 0
+                            processed_undouble_fre[j] = 1
+                        #如果只有一个频率，但是基频不是the_most,则将the_most的processed_undouble_fre置为[1,0,0,0]processed_double_fre=[0,1,1,1],-1都还是保留
+        processed_undouble_fre[ sorted_inf_fre == 1] = -1#mode fre为-1的即<2Gs的情况 ，等同于sorted_inf_fre == 1 ，在最高后的处理时processed_undouble_fre取-1
+        # 恢复原始顺序
+        # restore_indices = np.argsort(sorted_indices)
+        double_fre = processed_double_fre[np.argsort(sorted_indices)]
+        undouble_fre =processed_undouble_fre[np.argsort(sorted_indices)]
+        if np.sum(processed_undouble_fre == 1) == 1:#如果只有一个频率，但是基频不是the_most,则将the_most的processed_undouble_fre置为[1,0,0,0]processed_double_fre=[0,1,1,1],-1都还是保留
+            undouble_fre, double_fre = self.process_array(undouble_fre, double_fre,mode_fre_data_slice)
+        return double_fre, undouble_fre
\ No newline at end of file

jddb-agents/tm_extractor/custom_processor/inte_processor.py

+#  -*-  coding: utf-8  -*-
+import copy
+import math
+import os
+from jddb.processor import Signal, Shot, ShotSet, BaseProcessor
+from typing import Optional, List
+from scipy.fftpack import fft
+from copy import deepcopy
+import numpy as np
+from copy import deepcopy
+
+class InteProcessor(BaseProcessor):
+    """
+    给定积分范围，返回积分数值以及求B的数值，（250k的采样）
+    降采样，做时间切片
+    傅里叶分解
+
+    """
+    def __init__(self,NS=0.0689):# cm^2
+        super().__init__(NS=NS)
+
+        self.NS = NS
+
+    def transform(self, *signal: Signal):
+        ##signal[0]:LFS MA_POLA_P06
+        # signal[1]:HFS MA_POLB_P06
+        hfs_signal = deepcopy(signal.__getitem__(0))
+        lfs_signal = deepcopy(signal.__getitem__(1))
+        NS=self.NS# cm^2
+
+        # dt=1/lfs_signal.attributes["OriginalSampleRate"]
+        dt = 1 / lfs_signal.attributes["SampleRate"]
+        lfs_signal.data = 1e4 * dt * np.cumsum(lfs_signal.data) / NS#Gs
+        hfs_signal.data = 1e4 * dt * np.cumsum(hfs_signal.data) / NS#Gs
+
+        return hfs_signal, lfs_signal

jddb-agents/tm_extractor/custom_processor/is_tm_processor.py

+# -*-  coding:utf-8  -*-
+import numpy as np
+from typing import Tuple
+from jddb.processor import Signal, BaseProcessor
+from copy import deepcopy
+class IsTMProcessor(BaseProcessor):
+    def __init__(self):
+        super().__init__()
+
+    def transform(self, *signal: Signal) -> Signal:
+        most_signal = deepcopy(signal.__getitem__(0))
+        is_tm_signal = deepcopy(signal.__getitem__(0))
+        is_tm_data = np.zeros((len(most_signal.data)),dtype=int)
+        for i in range(len(most_signal.data)):
+            data_th_array =np.array([signal.__getitem__(th).data[i] for th in range(len(signal))])
+            for index_th in range(len(data_th_array)):
+                if  data_th_array[index_th, 2]>0:
+                    is_tm_data[i] = 1
+                    break
+        is_tm_signal.data = is_tm_data
+        return is_tm_signal

jddb-agents/tm_extractor/custom_processor/lowpass_processor.py

+#  -*-  coding: utf-8  -*-
+import math
+from typing import Tuple
+from jddb.processor import Signal, Shot, ShotSet, BaseProcessor
+from typing import Optional, List
+from scipy.fftpack import fft
+from copy import deepcopy
+from scipy.interpolate import interp1d
+import pandas as pd
+import numpy as np
+from scipy import signal as sig
+from copy import deepcopy
+
+import matplotlib.pyplot as plt
+
+class LowPassProcessor(BaseProcessor):
+    def __init__(self, cutoff_freq: float):
+        super().__init__(cutoff_freq=cutoff_freq)
+        self.cutoff_freq = cutoff_freq # 截止频率
+
+    def transform(self, signal: Signal) -> Signal:
+
+        fs = signal.attributes["SampleRate"]
+        # 设计低通滤波器
+
+        nyquist_freq = 0.5 * fs
+        normal_cutoff = self.cutoff_freq / nyquist_freq
+
+        b, a = sig.butter(4, normal_cutoff, btype='low', analog=False)
+        # 设计倒相滤波器
+
+        # 应用滤波器
+
+        filtered_signal_data = sig.lfilter(b, a, signal.data)
+        # filtered_signal = signal_input.lfilter(b, a, smoothed_signal)
+        filtered_signal_inverse_data = sig.lfilter(b, a, np.flip(filtered_signal_data))
+        # 对倒相滤波器的结果进行倒序，以抵消相位延迟
+        filtered_signal_phase_compensated_data= np.flip(filtered_signal_inverse_data)
+        filtered_signal_phase_compensated_data = np.round(filtered_signal_phase_compensated_data, 6)
+
+
+        return Signal(data=filtered_signal_phase_compensated_data, attributes=signal.attributes)

jddb-agents/tm_extractor/custom_processor/mn_mode_union.py

+import os
+from collections import Counter
+import math
+from typing import Tuple
+from copy import deepcopy
+import numpy as np
+from scipy import signal as sig
+from jddb.processor import *
+class MNModeUnionProcessor(BaseProcessor):
+    def __init__(self,pol_samperate: int,real_angle:float,pad=1):
+        super().__init__(pol_samperate=pol_samperate,real_angle=real_angle,pad=pad)
+        self.pad = pad
+        self.pol_samperate = pol_samperate
+        self.real_angle = real_angle
+        self.n_pad=250
+
+    def transform(self, *signal: Signal) -> Tuple[Signal, ...]:
+        n_most_signal = deepcopy(signal.__getitem__(0))
+        n_sec_signal = deepcopy(signal.__getitem__(1))
+        n_third_signal = deepcopy(signal.__getitem__(2))
+        n_forth_signal = deepcopy(signal.__getitem__(3))
+        m_four_signal_list = [deepcopy(signal.__getitem__(i)) for i in range(4, 8)]
+        # m_most_signal = deepcopy(signal.__getitem__(4))
+        # m_sec_signal = deepcopy(signal.__getitem__(5))
+        # m_third_signal = deepcopy(signal.__getitem__(6))
+        # m_forth_signal = deepcopy(signal.__getitem__(7))
+        m_phase_four_signal = deepcopy(signal.__getitem__(8))
+        pol03_signal = deepcopy(signal.__getitem__(9))
+        pol04_signal = deepcopy(signal.__getitem__(10))
+        n_data = np.array([n_most_signal.data, n_sec_signal.data, n_third_signal.data, n_forth_signal.data])
+        n_data_new = np.zeros( (n_data.shape[0], n_data.shape[1], n_data.shape[2] + 4))
+        #new_B_LFS_n_most_mode_number
+        self.n_pad=len(pol03_signal.data[0])*self.pad
+        self.pol_samperate = pol03_signal.attributes["OriginalSampleRate"]
+        # self.f=np.fft.fftfreq(self.n_pad, 1 / pol03_signal.attributes["OriginalSampleRate"]) [:int(self.n_pad / 2)]
+        for i in range(len(n_most_signal.data)):
+
+            m_4fre_i = [m_four_signal.data[i,3] for m_four_signal in m_four_signal_list]
+            m_4mode_i =[m_four_signal.data[i,0] for m_four_signal in m_four_signal_list]
+            for j in range(len(n_data)):
+                if not np.all(n_data[j][i] == 0):
+                    use_ampd = 0
+                    #找到对应的m_four_signal_list的模数和频率
+                    m_fre_closest_index,m_fre_closest_value=self.find_fre_mode(m_4fre_i,n_data[j][i][2])
+                    if m_fre_closest_index!=-1:
+                        m_mode_number=m_4mode_i[m_fre_closest_index]
+                        use_ampd=1
+                    else:
+                        #重新做一下csd然后找到对应的值去填充
+                        m_mode_number,m_fre_closest_value=self.csd_resolve(n_data[j][i][2],pol03_signal.data[i],pol04_signal.data[i])
+                    n_data_new[j][i] = np.hstack((n_data[j][i], use_ampd, m_fre_closest_value, m_mode_number, m_phase_four_signal.data[i, j]))
+        n_most_signal.data = n_data_new[0]
+        n_sec_signal.data = n_data_new[1]
+        n_third_signal.data = n_data_new[2]
+        n_forth_signal.data = n_data_new[3]
+        return n_most_signal,n_sec_signal,n_third_signal,n_forth_signal
+
+    def csd_resolve(self, n_j_data_i, pol03_data_i, pol04_data_i):
+        """
+        重新做一下csd然后找到对应的值去填充
+        """
+        f,csd_resolve  = sig.csd( pol03_data_i,pol04_data_i, fs=self.pol_samperate, window='hann', nperseg=self.n_pad,
+                            scaling='density')
+        # 找到 CSD 幅值最大处的索引
+        max_idx_03 = np.argmax(np.abs(csd_resolve))
+        # 获取最大幅值对应的频率
+        m_fre_closest_index = np.abs(np.array(f) - n_j_data_i).argmin()
+        # 使用索引找到最接近的数值
+        m_fre_closest_value = f[m_fre_closest_index]
+        phase_csd = (np.angle(csd_resolve) * 180 / np.pi)[m_fre_closest_index]
+        mode_number=phase_csd/self.real_angle
+        return mode_number,m_fre_closest_value
+
+    def find_fre_mode(self, m_fre_i, n_j_data_i):
+        """
+        找到对应的m_j_data_i的模数和频率
+        """
+        # 找到最接近值的索引
+        m_fre_closest_index = np.abs(np.array(m_fre_i) - n_j_data_i).argmin()
+        # 使用索引找到最接近的数值
+        m_fre_closest_value = m_fre_i[m_fre_closest_index]
+        # 检查一下误差值不能大于等于1K，如果超过，那么需要重新做一下csd然后找到对应的值去填充
+        if np.abs(m_fre_closest_value - n_j_data_i) >= 600:
+            m_fre_closest_index = -1
+            m_fre_closest_value = -1
+        return m_fre_closest_index, m_fre_closest_value
\ No newline at end of file

jddb-agents/tm_extractor/custom_processor/odd_even_processor.py

+#  -*-  coding: utf-8  -*-
+import copy
+import math
+import os
+from jddb.processor import Signal, Shot, ShotSet, BaseProcessor
+from typing import Optional, List
+from scipy.fftpack import fft
+from copy import deepcopy
+import numpy as np
+from copy import deepcopy
+
+class EvenOddProcessor(BaseProcessor):
+    """
+    给定积分范围，返回积分数值以及求B的数值，（250k的采样）
+    降采样，做时间切片
+    傅里叶分解
+
+    """
+    def __init__(self):
+        super().__init__()
+
+
+    def transform(self, *signal: Signal):
+        ##signal[0]:LFS MA_POLA_P06
+        # signal[1]:HFS MA_POLB_P06
+        hfs_signal = deepcopy(signal.__getitem__(0))
+        lfs_signal = deepcopy(signal.__getitem__(1))
+        even_signal = deepcopy(signal.__getitem__(0))
+        odd_signal = deepcopy(signal.__getitem__(0))
+        odd_signal.data =( lfs_signal.data-hfs_signal.data)/2
+        even_signal.data = (lfs_signal.data + hfs_signal.data) / 2
+        return odd_signal, even_signal

jddb-agents/tm_extractor/custom_processor/slice_processor.py

+#  -*-  coding: utf-8  -*-
+import math
+from typing import Tuple
+from jddb.processor import Signal, Shot, ShotSet, BaseProcessor
+import numpy as np
+from copy import deepcopy
+class SliceProcessor(BaseProcessor):
+    """
+            input the point number of the window  and overlap rate of the given window ,
+        then the sample rate is recalculated,  return a signal_input of time window sequence
+    """
+    def __init__(self, window_length: int, overlap: float):
+        super().__init__(window_length=window_length, overlap=overlap)
+        assert (0 <= overlap <= 1), "Overlap is not between 0 and 1."
+        self.params.update({"WindowLength": window_length,
+                            "Overlap": overlap})
+
+    def transform(self, signal: Signal) -> Signal:
+        window_length = self.params["WindowLength"]
+        overlap = self.params["Overlap"]
+        new_signal = deepcopy(signal)
+        raw_sample_rate = new_signal.attributes["SampleRate"]
+        step = int(window_length * round((1 - overlap), 3))
+
+        down_time = new_signal.time[-1]
+
+        # down_time = round(down_time, 3)
+
+        idx = len(signal.data)
+        window = list()
+        while (idx - window_length) >= 0:
+            window.append(new_signal.data[idx - window_length:idx])
+            idx -= step
+        window.reverse()
+        new_signal.attributes['SampleRate'] = raw_sample_rate / step
+        new_signal.data = np.array(window)
+        new_signal.attributes['StartTime']  = down_time - (len(window)-1) / new_signal.attributes['SampleRate']
+        # new_signal.attributes['StartTime'] = round(new_start_time, 3)
+        new_signal.attributes['OriginalSampleRate'] = raw_sample_rate
+        return new_signal
\ No newline at end of file