+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import json
+import yaml
+from yaml.loader import SafeLoader
+
+from .strings import USTR
+from vsensebox.utils.commontools import getAbsPathFDS, isExist
+from vsensebox.utils.logtools import add_error_log
+
+
+
+[docs]
+def isDictString(input_string):
+ """Check whether the :obj:`input_string` is a valid raw dictionary.
+
+ Parameters
+ ----------
+ input_string : str
+ An input of raw string.
+
+ Returns
+ -------
+ bool
+ Validation status.
+ """
+ res = True
+ if (
+ isinstance(input_string, str) and
+ len(input_string) > 4 and
+ input_string[0] == '[' and
+ input_string[1] == '{' and
+ input_string[-1] == ']' and
+ input_string[-2] == '}'
+ ):
+ try:
+ yaml.load(input_string, Loader=SafeLoader)
+ except ValueError as e:
+ res = False
+ else:
+ res = False
+ return res
+
+
+
+[docs]
+def getCFGDict(input):
+ """Get a configuration dictionary of a single document from the given :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document.
+ """
+ doc = {}
+ if isinstance(input, str):
+ if ".yaml" in input.lower() or ".json" in input.lower():
+ doc = loadDocument(getAbsPathFDS(input))
+ else:
+ doc = loadRawYAMLString(input)
+ elif isinstance(input, dict):
+ doc = input
+ return doc
+
+
+
+[docs]
+def getListCFGDoc(input):
+ """Get a list of configuration dictionary of document from the given :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+
+ Returns
+ -------
+ list[dict, ...]
+ A list of configuration dictionary.
+ """
+ doc_list = []
+ if isinstance(input, str):
+ if ".yaml" in input.lower():
+ doc_list = loadListDocument(getAbsPathFDS(input))
+ else:
+ doc_list = loadRawYAMLStringMT(input)
+ elif isinstance(input, dict):
+ for doc in input:
+ doc_list.append(doc)
+ return doc_list
+
+
+
+[docs]
+def loadDocument(yaml_json):
+ """Return a configuration dictionary of a single document from the given file
+ :obj:`yaml_json`.
+
+ Parameters
+ ----------
+ yaml_json : str
+ A path of a YAML/JSON file.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document.
+ """
+ document = {}
+ if isExist(yaml_json):
+ with open(yaml_json, 'rb') as cfg:
+ try:
+ if '.json' in yaml_json.lower():
+ document = json.load(cfg)
+ elif '.yaml' in yaml_json.lower():
+ document = yaml.load(cfg, Loader=SafeLoader)
+ except ValueError as e:
+ msg = 'loadDocument() -> ' + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ msg = "loadDocument() -> " + str(yaml_json) + "' does not exist!"
+ add_error_log(msg)
+ raise ValueError(msg)
+ return document
+
+
+
+[docs]
+def loadListDocument(yaml_json):
+ """Return a list of configuration dictionary from the given file :obj:`yaml_json`.
+
+ Parameters
+ ----------
+ yaml_json : str
+ A path of a YAML/JSON file.
+
+ Returns
+ -------
+ list[dict, ...]
+ A list of configuration dictionary.
+ """
+ document_list = []
+ if isExist(yaml_json):
+ with open(yaml_json, 'rb') as cfg:
+ try:
+ if '.json' in yaml_json.lower():
+ docs = json.load(cfg)
+ elif '.yaml' in yaml_json.lower():
+ docs = yaml.load_all(cfg, Loader=SafeLoader)
+ except ValueError as e:
+ msg = 'loadListDocument() -> ' + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ for doc in docs:
+ document_list.append(doc)
+ else:
+ msg = "loadListDocument() -> " + str(yaml_json) + "' does not exist!"
+ add_error_log(msg)
+ raise ValueError(msg)
+ return document_list
+
+
+
+[docs]
+def loadRawYAMLString(raw_string):
+ """Return a configuration dictionary of a single document from the given string
+ :obj:`raw_string`.
+
+ Parameters
+ ----------
+ raw_string : str
+ A raw string of JSON or YAML; for example,
+ :code:`raw_string="[{'tk_name': 'SORT'}]"`.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document.
+ """
+ document = {}
+ try:
+ d = yaml.load(raw_string, Loader=SafeLoader)
+ document = next(iter(d))
+ except ValueError as e:
+ msg = 'loadRawYAMLString() -> ' + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ return document
+
+
+
+[docs]
+def loadRawYAMLStringMT(raw_string):
+ """Return a list of multiple YAML dictionary from the given string :obj:`raw_string`.
+
+ Parameters
+ ----------
+ raw_string : str
+ A raw string of JSON or YAML; for example,
+ :code:`raw_string="[{'tk_name': 'SORT'}, {'tk_name': 'DeepSORT'}]"`.
+
+ Returns
+ -------
+ list[dict, ...]
+ A list of configuration dictionary.
+ """
+ document_list = []
+ try:
+ ds = yaml.load_all(raw_string, Loader=SafeLoader)
+ for d in ds:
+ document_list.append(d)
+ except ValueError as e:
+ msg = 'loadRawYAMLStringMT() -> ' + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ return document_list
+
+
+
+[docs]
+def dumpDocDict(output_file, doc, header):
+ """Dump a configuration dictionary of a single document into a YAML file
+ with simple format.
+
+ Parameters
+ ----------
+ output_file : str
+ A path file to dump.
+ doc : dict
+ A configuration dictionary of a single document.
+ header : str
+ A file header descriptoin.
+ """
+ try:
+ with open(output_file, 'w') as dumping:
+ dumping.write(header)
+ for key, value in doc.items():
+ key = str(key).lower()
+ if str(value) == "None" or value is None:
+ value = "null # NULL=Null=null is None in Python"
+ if key.lower() in ["detector", "tracker"]:
+ key = USTR.getUnifiedFormat(key)
+ value = USTR.getUnifiedFormat(value)
+ dumping.write('%s: %s\n' % (key, value))
+ except ValueError as e:
+ msg = 'dumpDocDict() -> ' + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+
+
+
+[docs]
+def dumpListDocDict(output_file, doc_list, header):
+ """Dump a list of YAML dictionary into a YAML file with simple format.
+
+ Parameters
+ ----------
+ output_file : str
+ A path file to dump.
+ doc_list : list[dict, ...]
+ A list of configuration dictionary.
+ header : str
+ A file header descriptoin.
+ """
+ try:
+ with open(output_file, 'w') as dumping:
+ dumping.write(header)
+ sep_index = 1
+ for d in doc_list:
+ for key, value in d.items():
+ key = str(key).lower()
+ if str(value) == "None" or value is None:
+ value = "null # NULL=Null=null is None in Python"
+ if key.lower() in ["detector", "tracker"]:
+ key = USTR.getUnifiedFormat(key)
+ value = USTR.getUnifiedFormat(value)
+ dumping.write('%s: %s\n' % (key, value))
+ if sep_index < len(doc_list):
+ dumping.write("---\n")
+ sep_index += 1
+ except ValueError as e:
+ msg = 'dumpListDocDict() -> ' + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+from .strings import USTR
+from .confighelper import getCFGDict, getListCFGDoc, dumpDocDict, dumpListDocDict
+from vsensebox.utils.logtools import add_warning_log, add_error_log
+from vsensebox.utils.commontools import (joinFPathFull, getGlobalRootDir,
+ getAdaptiveAbsPathFDS, normalizePathFDS)
+
+IN_ROOT_DIR = getGlobalRootDir()
+IN_DATA_DIR = joinFPathFull(IN_ROOT_DIR, 'data')
+IN_CONFIG_DIR = joinFPathFull(IN_ROOT_DIR, 'config')
+DET_CONFIG_DIR = joinFPathFull(IN_CONFIG_DIR, 'detectors')
+TRK_CONFIG_DIR = joinFPathFull(IN_CONFIG_DIR, 'trackers')
+STR_CONFIG_DIR = joinFPathFull(IN_CONFIG_DIR, 'strings')
+
+
+
+[docs]
+class BaseCGF(object):
+
+ """
+ An base CFG class used to store the necessary configurations of a module.
+
+ Attributes
+ ----------
+ unified_strings : MyStrings, auto
+ A :class:`MyStrings` object used to store unified strings.
+ configs : dict or list[dict], default={}
+ A configuration dictionary of a single document or a list of multiple documents
+ of the configurations.
+ """
+
+ def __init__(self):
+ self.unified_strings = USTR
+ self.configs = {}
+
+
+[docs]
+ def loadDoc(self, input):
+ """Load and set dictionary of a single document from a YAML/JSON file or string.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ self.configs = getCFGDict(input)
+
+
+
+[docs]
+ def loadDocs(self, input):
+ """Load and set a list of multiple documents from a YAML/JSON file or string.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ self.configs = getListCFGDoc(input)
+
+
+
+[docs]
+ def dumpDoc(self, output, header=""):
+ """Dump the :attr:`configs` into a YAML file with simple format.
+
+ Parameters
+ ----------
+ output : str
+ A path file to dump.
+ header : str
+ A file header descriptoin.
+ """
+ dumpDocDict(output_file=output, doc=self.configs, header=header)
+
+
+
+[docs]
+ def dumpDocs(self, output, header=""):
+ """Dump the :attr:`configs` into a YAML file with simple format.
+
+ Parameters
+ ----------
+ output : str
+ A path file to dump.
+ header : str
+ A file header descriptoin.
+ """
+ dumpListDocDict(output_file=output, doc=self.configs, header=header)
+
+
+
+
+
+[docs]
+class DCFG_YOLOCLS(BaseCGF):
+
+ """
+ A class used to store the necessary configurations of detector YOLO Classic which
+ use :code:`.weights` model.
+
+ Attributes
+ ----------
+ detector : str
+ Configured name of detector YOLO Classic.
+ nms : float
+ Parameter nms of YOLO Classic.
+ conf : float
+ Parameter conf of YOLO Classic.
+ imgsz : int
+ Input image size of YOLO Classic.
+ min_width : int
+ Minimum width filter.
+ classes : list[int, ...]
+ Filtering classes of detection.
+ class_file : str
+ Path of coco.names file of YOLO Classic.
+ model_cfg_file : str
+ Path of .cfg file of YOLO Classic.
+ model_file : str
+ Path of .weights file of YOLO Classic.
+ from_dir : str
+ Path of the root directory, relative to path of :attr:`model_file`.
+ """
+
+ def __init__(self, cfg=None, relative_to_vsensebox_root=False):
+ """Initialize the class according to the :obj:`relative_to_vsensebox_root` which defines
+ whether all the paths inside your configuration file are relative to :code:`{vsensebox root}`
+ or not. If all the paths inside your configuration file have full absolute paths,
+ setting :obj:`relative_to_vsensebox_root` is optional.
+
+ Parameters
+ ----------
+ cfg : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ relative_to_vsensebox_root : bool, default=False
+ (1) Set :code:`relative_to_vsensebox_root=False` to use your current working directory
+ as where all the paths in your configuration file are relative to; for example,
+ the path of :attr:`model_file` inside your configuration file is set relatively
+ to your current working directory.
+ (2) Set :code:`relative_to_vsensebox_root=True` when all the paths in your configuration
+ file are relative to :code:`{vsensebox root}`.
+ """
+ super().__init__()
+ self.from_dir = ""
+ if relative_to_vsensebox_root:
+ self.from_dir = IN_ROOT_DIR
+ if cfg is not None:
+ self.set(cfg)
+
+
+[docs]
+ def set(self, input):
+ """
+ Set configurations according to :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ super().loadDoc(input)
+ if self.configs:
+ try:
+ self.detector = self.unified_strings.getUnifiedFormat(self.configs['detector'])
+ self.nms = self.configs['nms']
+ self.conf = self.configs['conf']
+ self.imgsz = self.configs['imgsz']
+ self.min_width = self.configs['min_width']
+ self.classes = self.configs['classes']
+ self.class_file = getAdaptiveAbsPathFDS(self.from_dir, self.configs['class_file'])
+ self.model_cfg_file = getAdaptiveAbsPathFDS(self.from_dir, self.configs['model_cfg_file'])
+ self.model_file = getAdaptiveAbsPathFDS(self.from_dir, self.configs['model_file'])
+ self.configs = self.getDocument()
+ except Exception as e:
+ msg = "DCFG_YOLOCLS : set() -> " + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ add_warning_log("DCFG_YOLOCLS : set() -> The configuration is empty.")
+
+
+
+[docs]
+ def getDocument(self):
+ """Return a configuration dictionary of a single document of the attributes which
+ are the parameters of detector YOLO Classic.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document of the configurations.
+ """
+ yolocs_doc = {
+ "detector": self.detector,
+ "nms": self.nms,
+ "conf": self.conf,
+ "imgsz": self.imgsz,
+ "min_width": self.min_width,
+ "classes": self.classes,
+ "class_file": normalizePathFDS(IN_ROOT_DIR, self.class_file),
+ "model_cfg_file": normalizePathFDS(IN_ROOT_DIR, self.model_cfg_file),
+ "model_file": normalizePathFDS(IN_ROOT_DIR, self.model_file),
+ }
+ return yolocs_doc
+
+
+
+
+
+[docs]
+class DCFG_YOLOULT(BaseCGF):
+
+ """
+ A class used to store the necessary configurations of detector Ultralytics.
+
+ Attributes
+ ----------
+ detector : str
+ Configured name of detector YOLO Ultralytics.
+ conf : float
+ Parameter conf of YOLO Ultralytics.
+ iou : float
+ Parameter iou of YOLO Ultralytics.
+ imgsz : int
+ Parameter imgsz of YOLO Ultralytics.
+ half : bool
+ Parameter half of YOLO Ultralytics.
+ min_width : int
+ Minimum width filter.
+ classes : list[int, ...]
+ Parameter classes of YOLO Ultralytics.
+ device : str
+ Parameter of device to compute; for example, 0 or cpu.
+ max_det : int
+ Parameter max_det of YOLO Ultralytics.
+ line_width : int
+ Parameter line_width of YOLO Ultralytics.
+ model_file : str
+ Path of :attr:`model_file` for YOLO Ultralytics.
+ from_dir : str
+ Path of the root directory, relative to path of :attr:`model_file`.
+ """
+
+ def __init__(self, cfg=None, relative_to_vsensebox_root=False):
+ """Initialize the class according to the :obj:`relative_to_vsensebox_root` which
+ defines whether all the paths inside your configuration file are relative to
+ :code:`{vsensebox root}` or not. If all the paths inside your configuration file have
+ full absolute paths, setting :obj:`relative_to_vsensebox_root` is optional.
+
+ Parameters
+ ----------
+ cfg : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ relative_to_vsensebox_root : bool, default=False
+ (1) Set :code:`relative_to_vsensebox_root=False` to use your current working
+ directory as where all the paths in your configuration file are relative to;
+ for example, the path of :attr:`model_file` inside your configuration file is
+ set relatively to your current working directory.
+ (2) Set :code:`relative_to_vsensebox_root=True` when all the paths in your
+ configuration file are relative to :code:{vsensebox root}.
+ """
+ super().__init__()
+ self.from_dir = ""
+ if relative_to_vsensebox_root:
+ self.from_dir = IN_ROOT_DIR
+ if cfg is not None:
+ self.set(cfg)
+
+
+[docs]
+ def set(self, input):
+ """
+ Set configurations according to :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ super().loadDoc(input)
+ if self.configs:
+ try:
+ self.detector = self.unified_strings.getUnifiedFormat(self.configs['detector'])
+ self.conf = self.configs['conf']
+ self.iou = self.configs['iou']
+ self.imgsz = self.configs['imgsz']
+ self.half = self.configs['half']
+ self.min_width = self.configs['min_width']
+ self.classes = self.configs['classes']
+ self.device = self.configs['device']
+ self.max_det = self.configs['max_det']
+ self.line_width = self.configs['line_width']
+ self.model_file = getAdaptiveAbsPathFDS(self.from_dir, self.configs['model_file'])
+ self.configs = self.getDocument()
+ except Exception as e:
+ msg = "DCFG_YOLOULT : set() -> " + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ add_warning_log("DCFG_YOLOULT : set() -> The configuration is empty.")
+
+
+
+[docs]
+ def getDocument(self):
+ """Return yolout_doc, a configuration dictionary of a single document of the attributes which
+ are the parameters of detector YOLO Ultralytics.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document of the configurations.
+ """
+ yolout_doc = {
+ "detector": self.detector,
+ "conf": self.conf,
+ "iou": self.iou,
+ "imgsz": self.imgsz,
+ "half": self.half,
+ "min_width": self.min_width,
+ "classes": self.classes,
+ "device": self.device,
+ "max_det": self.max_det,
+ "line_width": self.line_width,
+ "model_file": normalizePathFDS(IN_ROOT_DIR, self.model_file),
+ }
+ return yolout_doc
+
+
+
+
+
+[docs]
+class TCFG_Centroid(BaseCGF):
+
+ """
+ A class used to store the necessary configurations of tracker Centroid.
+
+ Attributes
+ ----------
+ tracker : str
+ Configured name of tracker Centroid.
+ max_spread : int
+ Maximum distance of the being tracked :class:`Person` object of previous and current state.
+ pref_y : str
+ Prefered Y of represented point; Choices of Top/Center/Bottom.
+ """
+
+ def __init__(self, cfg=None, relative_to_vsensebox_root=False):
+ """Initialize the class.
+
+ Parameters
+ ----------
+ cfg : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ relative_to_vsensebox_root : bool, default=False
+ Being consistent with others, will be ignored.
+ """
+ super().__init__()
+ self.from_dir = ""
+ if relative_to_vsensebox_root:
+ self.from_dir = IN_ROOT_DIR
+ if cfg is not None:
+ self.set(cfg)
+
+
+[docs]
+ def set(self, input):
+ """
+ Set configurations according to :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ super().loadDoc(input)
+ if self.configs:
+ try:
+ self.tracker = self.unified_strings.getUnifiedFormat(self.configs['tracker'])
+ self.max_spread = self.configs['max_spread']
+ self.pref_y = self.configs['pref_y']
+ self.configs = self.getDocument()
+ except Exception as e:
+ msg = "TCFG_Centroid : set() -> " + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ add_warning_log("TCFG_Centroid : set() -> The configuration is empty.")
+
+
+
+[docs]
+ def getDocument(self):
+ """Return a configuration dictionary of a single document of the attributes which
+ are the parameters of tracker Centroid.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document of the configurations.
+ """
+ centroid_doc = {
+ "tracker": self.tracker,
+ "max_spread": self.max_spread,
+ "pref_y": self.pref_y
+ }
+ return centroid_doc
+
+
+
+
+
+[docs]
+class TCFG_BasicIoU(BaseCGF):
+
+ """
+ A class used to store the necessary configurations of tracker BasicIoU.
+
+ Attributes
+ ----------
+ tracker : str
+ Configured name of tracker BasicIoU.
+ min_iou : float
+ Minimum IoU value for matching.
+ device : str
+ Device to compute; for example, 0 or cpu.
+ """
+
+ def __init__(self, cfg=None, relative_to_vsensebox_root=False):
+ """Initialize the class.
+
+ Parameters
+ ----------
+ cfg : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ relative_to_vsensebox_root : bool, default=False
+ Being consistent with others, will be ignored.
+ """
+ super().__init__()
+ self.from_dir = ""
+ if relative_to_vsensebox_root:
+ self.from_dir = IN_ROOT_DIR
+ if cfg is not None:
+ self.set(cfg)
+
+
+[docs]
+ def set(self, input):
+ """
+ Set configurations according to :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ super().loadDoc(input)
+ if self.configs:
+ try:
+ self.tracker = self.unified_strings.getUnifiedFormat(self.configs['tracker'])
+ self.min_iou = self.configs['min_iou']
+ self.device = self.configs['device']
+ self.configs = self.getDocument()
+ except Exception as e:
+ msg = "TCFG_BasicIoU : set() -> " + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ add_warning_log("TCFG_BasicIoU : set() -> The configuration is empty.")
+
+
+
+[docs]
+ def getDocument(self):
+ """Return a configuration dictionary of a single document of the attributes which
+ are the parameters of tracker BasicIoU.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document of the configurations.
+ """
+ basiciou_doc = {
+ "tracker": self.tracker,
+ "min_iou": self.min_iou,
+ "device": self.device
+ }
+ return basiciou_doc
+
+
+
+
+
+[docs]
+class TCFG_SORT(BaseCGF):
+
+ """
+ A class used to store the necessary configurations of tracker SORT.
+
+ Attributes
+ ----------
+ tracker : str
+ Configured name of tracker SORT.
+ max_age : int
+ Parameter :obj:`max_age` of tracker SORT.
+ min_hits : int
+ Parameter :obj:`min_hits` of tracker SORT.
+ iou_threshold : float
+ Parameter :obj:`iou_threshold` of tracker SORT.
+ """
+
+ def __init__(self, cfg=None, relative_to_vsensebox_root=False):
+ """Initialize the class.
+
+ Parameters
+ ----------
+ cfg : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ relative_to_vsensebox_root : bool, default=False
+ Being consistent with others, will be ignored.
+ """
+ super().__init__()
+ self.from_dir = ""
+ if relative_to_vsensebox_root:
+ self.from_dir = IN_ROOT_DIR
+ if cfg is not None:
+ self.set(cfg)
+
+
+[docs]
+ def set(self, input):
+ """
+ Set configurations according to :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ super().loadDoc(input)
+ if self.configs:
+ try:
+ self.tracker = self.unified_strings.getUnifiedFormat(self.configs['tracker'])
+ self.max_age = self.configs['max_age']
+ self.min_hits = self.configs['min_hits']
+ self.iou_threshold = self.configs['iou_threshold']
+ self.configs = self.getDocument()
+ except Exception as e:
+ msg = "TCFG_SORT : set() -> " + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ add_warning_log("TCFG_SORT : set() -> The configuration is empty.")
+
+
+
+[docs]
+ def getDocument(self):
+ """
+ Return a configuration dictionary of a single document of the attributes which
+ are the parameters of tracker SORT.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document of the configurations.
+ """
+ sort_doc = {
+ "tracker": self.tracker,
+ "max_age": self.max_age,
+ "min_hits": self.min_hits,
+ "iou_threshold": self.iou_threshold
+ }
+ return sort_doc
+
+
+
+
+
+[docs]
+class TCFG_DeepSORT(BaseCGF):
+
+ """
+ A class used to store the necessary configurations of tracker DeepSORT.
+
+ Attributes
+ ----------
+ tracker : str
+ Configured name of tracker DeepSORT.
+ nn_budget : int
+ Parameter :obj:`nn_budget` of tracker DeepSORT.
+ batch_size : int
+ Parameter :obj:`batch_size` of tracker DeepSORT.
+ nms_max_overlap : float
+ Parameter :obj:`nms_max_overlap` of tracker DeepSORT.
+ max_cosine_distance : float
+ Parameter :obj:`max_cosine_distance` of tracker DeepSORT.
+ model_file : str
+ Path of model file for tracker DeepSORT.
+ from_dir : str
+ Path of the root directory, relative to path of :attr:`model_file`.
+ """
+
+ def __init__(self, cfg=None, relative_to_vsensebox_root=False):
+ """Initialize the class according to the :obj:`relative_to_vsensebox_root` which defines
+ whether all the paths inside your configuration file are relative to :code:`{vsensebox root}`
+ or not. If all the paths inside your configuration file have full absolute paths, setting
+ :obj:`relative_to_vsensebox_root` is optional.
+
+ Parameters
+ ----------
+ cfg : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ relative_to_vsensebox_root : bool, default=False
+ (1) Set :code:`relative_to_vsensebox_root=False` to use your current working directory
+ as where all the paths in your configuration file are relative to; for example,
+ the path of :attr:`model_file` inside your configuration file is set relatively to
+ your current working directory.
+ (2) Set :code:`relative_to_vsensebox_root=True` when all the paths in your configuration
+ file are relative to :code:`{vsensebox root}`.
+ """
+ super().__init__()
+ self.from_dir = ""
+ if relative_to_vsensebox_root:
+ self.from_dir = IN_ROOT_DIR
+ if cfg is not None:
+ self.set(cfg)
+
+
+[docs]
+ def set(self, input):
+ """
+ Set configurations according to :obj:`input`.
+
+ Parameters
+ ----------
+ input : str or dict
+ A YAML/JSON file path, or a raw/ready dictionary.
+ """
+ super().loadDoc(input)
+ if self.configs:
+ try:
+ self.tracker = self.unified_strings.getUnifiedFormat(self.configs['tracker'])
+ self.nn_budget = self.configs['nn_budget']
+ self.batch_size = self.configs['batch_size']
+ self.nms_max_overlap = self.configs['nms_max_overlap']
+ self.max_cosine_distance = self.configs['max_cosine_distance']
+ self.model_file = getAdaptiveAbsPathFDS(self.from_dir, self.configs['model_file'])
+ self.configs = self.getDocument()
+ except Exception as e:
+ msg = "TCFG_DeepSORT : set() -> " + str(e)
+ add_error_log(msg)
+ raise ValueError(msg)
+ else:
+ add_warning_log("TCFG_DeepSORT : set() -> The configuration is empty.")
+
+
+
+[docs]
+ def getDocument(self):
+ """Return a configuration dictionary of a single document of the attributes which
+ are the parameters of tracker DeepSORT.
+
+ Returns
+ -------
+ dict
+ A configuration dictionary of a single document of the configurations.
+ """
+ deepsort_doc = {
+ "tracker": self.tracker,
+ "nn_budget": self.nn_budget,
+ "batch_size": self.batch_size,
+ "nms_max_overlap": self.nms_max_overlap,
+ "max_cosine_distance": self.max_cosine_distance,
+ "model_file": normalizePathFDS(IN_ROOT_DIR, self.model_file)
+ }
+ return deepsort_doc
+
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import yaml
+from yaml.loader import SafeLoader
+from vsensebox.utils.commontools import joinFPathFull, getGlobalRootDir
+
+
+DEFAULT_STR_YAML = joinFPathFull(getGlobalRootDir(), "config/strings/strings.yaml")
+
+
+[docs]
+class UnifiedStrings(object):
+
+ """
+ A class used to set up unified strings of VSenseBox based on the internal strings.yaml.
+
+ Attributes
+ ----------
+ data : dict, auto
+ Data or documents read from strings.yaml.
+ """
+
+ def __init__(self, strings_yaml=DEFAULT_STR_YAML):
+ """Initailize by calling :meth:`load(strings_yaml=strings_yaml)`.
+
+ Parameters
+ ----------
+ strings_yaml : str, default='{vsensebox root}/config/strings/strings.yaml'
+ A path of a YAML file which stores the unified strings.
+ """
+ self.data = self.load(strings_yaml=strings_yaml)
+
+
+[docs]
+ def load(self, strings_yaml):
+ """Load a configuration dictionary of a single document as a dictionary from
+ a :obj:`strings_yaml` file and automatically pass to :meth:`set()`.
+
+ Parameters
+ ----------
+ strings_yaml : str
+ A path of a YAML file which stores the unified strings.
+ """
+ with open(strings_yaml, 'r') as str_cfg:
+ data = yaml.load(str_cfg, Loader=SafeLoader)
+ return data
+
+
+
+[docs]
+ def getUnifiedFormat(self, input_str):
+ """Return a standard unified format string.
+
+ Parameters
+ ----------
+ input_str : str
+ An input string.
+
+ Returns
+ -------
+ str
+ A unified format string.
+ """
+ res = ""
+ if isinstance(input_str, str):
+ if input_str.lower() in self.data['detector'].lower():
+ res = self.data['detector']
+ elif input_str.lower() in self.data['yolo_classic'].lower():
+ res = self.data['yolo_classic']
+ elif input_str.lower() in self.data['yolo_ultralytics'].lower():
+ res = self.data['yolo_ultralytics']
+ elif input_str.lower() in self.data['tracker'].lower():
+ res = self.data['tracker']
+ elif input_str.lower() in self.data['centroid'].lower():
+ res = self.data['centroid']
+ elif input_str.lower() in self.data['sort'].lower():
+ res = self.data['sort']
+ elif input_str.lower() in self.data['deepsort'].lower():
+ res = self.data['deepsort']
+ elif input_str.lower() in self.data['basiciou'].lower():
+ res = self.data['basiciou']
+ elif input_str.lower in ['top', 'center', 'bottom']:
+ res = input_str.title()
+ else:
+ res = input_str
+ else:
+ res = input_str
+ return res
+
+
+
+USTR = UnifiedStrings()
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import os
+
+from vsensebox.utils.logtools import add_warning_log, add_info_log
+from vsensebox.utils.commontools import isExist, joinFPathFull
+from vsensebox.config.configurator import DET_CONFIG_DIR, TRK_CONFIG_DIR, STR_CONFIG_DIR
+
+
+CUR_DIR = os.path.dirname(__file__)
+UI_TMP = joinFPathFull(CUR_DIR, "ui.tmp")
+
+
+[docs]
+def config(config_yaml=None):
+ """Launch GUI configuration tool of VSenseBox.
+
+ Parameters
+ ----------
+ config_yaml : str, default=None
+ Path of YAML config file.
+ """
+ if config_yaml is not None:
+ if not isExist(str(config_yaml)):
+ add_warning_log("VSenseBox:config() -> Input config_yaml is not valid.")
+ else:
+ with open(UI_TMP, "w+") as UI_TMP_file:
+ UI_TMP_file.write(config_yaml + "\n")
+
+ import sys
+ import subprocess as sp
+ p = sp.Popen([sys.executable, os.path.join(CUR_DIR, 'cfgloader_ui.py')])
+ stdout, stderr = p.communicate()
+
+
+
+[docs]
+def reset():
+ """Reset the internal configurations.
+ """
+ import shutil
+ shutil.unpack_archive(joinFPathFull(DET_CONFIG_DIR, 'detectors.zip'), DET_CONFIG_DIR)
+ shutil.unpack_archive(joinFPathFull(TRK_CONFIG_DIR, 'trackers.zip'), TRK_CONFIG_DIR)
+ shutil.unpack_archive(joinFPathFull(STR_CONFIG_DIR, 'strings.zip'), STR_CONFIG_DIR)
+ add_info_log("Reset successfully!")
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import cv2
+
+from vsensebox.utils.commontools import to_xyxy
+
+
+
+[docs]
+class YOLO_Classic(object):
+
+ """Class used as a custom layer or interface for interacting with detector module
+ YOLO_Classic which uses :code:`.weights` model.
+
+ Attributes
+ ----------
+ cfg : DCFG_YOLOCLS
+ A :class:`DCFG_YOLOCLS` object which manages the configurations of detector
+ YOLO_Classic.
+ model: cv::dnn::DetectionModel
+ A detection model object of OpenCV's deep learning network.
+ """
+
+
+[docs]
+ def __init__(self, cfg):
+ """Initialize according to the given configuration :obj:`cfg`
+ as :class:`DCFG_YOLOCLS` object.
+
+ Parameters
+ ----------
+ cfg : DCFG_YOLOCLS
+ A :class:`DCFG_YOLOCLS` object which manages the configurations of detector
+ YOLO_Classic.
+ """
+ self.cfg = cfg
+ net = cv2.dnn.readNet(cfg.model_file, cfg.model_cfg_file)
+ net.setPreferableBackend(cv2.dnn.DNN_BACKEND_CUDA)
+ net.setPreferableTarget(cv2.dnn.DNN_TARGET_CUDA)
+ self.model = cv2.dnn_DetectionModel(net)
+ self.model.setInputParams(size=(cfg.imgsz, cfg.imgsz), scale=1/255.0)
+
+
+
+[docs]
+ def detect(self, img):
+ """Detect general object with object's class filter :obj:`classes` in a
+ given :obj:`Mat` like object.
+
+ Parameters
+ ----------
+ img : Mat
+ A :obj:`Mat` like object.
+
+ Returns
+ -------
+ Mat
+ A :obj:`Mat` like object.
+ list[ndarray[int int int int], ...]
+ A list of bounding box :code:`ndarray[x y width height]`.
+ list[ndarray[int int int int], ...]
+ A list of bounding box :code:`ndarray[x1 y1 x2 y2]`.
+ list[]
+ A list of detected bodies' keypoints.
+ list[float]
+ A list of the detection confidence of every detected object.
+ list[int]
+ A list of detection classes.
+ """
+ boxes_xywh = []
+ boxes_xyxy = []
+ keypoints = []
+ confs = []
+ cls = []
+ if self.cfg.classes is None:
+ self.cfg.classes = [i for i in range(0, 80)]
+ _classes, confidences, boxes = self.model.detect(
+ img,
+ confThreshold=float(self.cfg.conf),
+ nmsThreshold=float(self.cfg.nms)
+ )
+ if len(boxes) > 0:
+ for cl, conf, box_xywh in zip(_classes.flatten(), confidences, boxes):
+ if cl in self.cfg.classes and box_xywh[2] >= self.cfg.min_width:
+ box_xywh = box_xywh.astype(int)
+ boxes_xywh.append(box_xywh)
+ box_xyxy = to_xyxy(box_xywh)
+ boxes_xyxy.append(box_xyxy)
+ confs.append(float(conf))
+ cls.append(int(cl))
+ return img, boxes_xywh, boxes_xyxy, keypoints, confs, cls
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import cv2
+
+from vsensebox.utils.commontools import to_xywh
+from vsensebox.utils.logtools import ignore_this_logger
+
+
+
+[docs]
+class YOLO_Ultralytics(object):
+
+ """Class used as a custom layer or interface for interacting with
+ detector YOLO Ultralytics which uses PyTorch .pt model.
+
+ Attributes
+ ----------
+ cfg : DCFG_YOLOULT
+ A :class:`DCFG_YOLOULT` object which manages the configurations
+ of detector YOLO_Ultralytics.
+ model: ultralytics.yolo.engine.YOLO
+ A detection model object of YOLO_Ultralytics.
+ """
+
+
+[docs]
+ def __init__(self, cfg):
+ """Initialize according to the given configuration :obj:`cfg`
+ as :class:`DCFG_YOLOULT` object.
+
+ Parameters
+ ----------
+ cfg : DCFG_YOLOULT
+ A :class:`DCFG_YOLOULT` object which manages the configurations
+ of detector YOLO_Ultralytics.
+ """
+ self.cfg = cfg
+ self.cpu_only = False
+ if isinstance(self.cfg.device, str):
+ if self.cfg.device.lower() == 'cpu':
+ self.cpu_only = True
+ ignore_this_logger("ultralytics")
+ if "nas" in self.cfg.model_file:
+ # YOLO NAS isn't stable yet :/
+ from ultralytics import NAS
+ self.model = NAS(self.cfg.model_file)
+ else:
+ from ultralytics import YOLO
+ self.model = YOLO(self.cfg.model_file)
+
+
+
+[docs]
+ def detect(self, img):
+ """Detect general object with object's class filter :obj:`classes`
+ in a given :obj:`Mat` like object.
+
+ Parameters
+ ----------
+ img : Mat
+ A :obj:`Mat` like object.
+
+ Returns
+ -------
+ Mat
+ A :obj:`Mat` like object.
+ list[ndarray[int int int int], ...]
+ A list of bounding box :code:`ndarray[x y width height]`.
+ list[ndarray[int int int int], ...]
+ A list of bounding box :code:`ndarray[x1 y1 x2 y2]`.
+ list[]
+ A list of detected bodies' keypoints.
+ list[float]
+ A list of the detection confidence of every detected object.
+ list[int]
+ A list of detection classes.
+ """
+ numpy_dets = []
+ boxes_xyxy = []
+ boxes_xywh = []
+ keypoints = []
+ confs = []
+ cls = []
+ dets = self.model.predict(
+ img,
+ imgsz=int(self.cfg.imgsz),
+ conf=float(self.cfg.conf),
+ classes=self.cfg.classes,
+ show_boxes=False,
+ device=self.cfg.device,
+ max_det=int(self.cfg.max_det),
+ line_width=self.cfg.line_width,
+ verbose=False
+ )
+ if self.cpu_only:
+ numpy_dets = dets[0].numpy()
+ else:
+ numpy_dets = dets[0].cuda().cpu().to("cpu").numpy()
+ dt_boxes_xyxy = numpy_dets.boxes.xyxy
+ dt_confidences = numpy_dets.boxes.conf
+ dt_classes = numpy_dets.boxes.cls
+ dt_keypoints = dets[0].keypoints
+ if dt_keypoints is not None:
+ for box_xyxy, conf, kp, cl in zip(dt_boxes_xyxy, dt_confidences,
+ reversed(dt_keypoints), dt_classes):
+ box_xyxy = box_xyxy.astype(int)
+ box_xywh = to_xywh(box_xyxy)
+ if box_xywh[2] >= self.cfg.min_width:
+ boxes_xywh.append(box_xywh)
+ boxes_xyxy.append(box_xyxy)
+ keypoint = kp.data[0]
+ keypoints.append(keypoint)
+ confs.append(float(conf))
+ cls.append(int(cl))
+ elif len(dt_boxes_xyxy) > 0:
+ for box_xyxy, conf, cl in zip(dt_boxes_xyxy, dt_confidences, dt_classes):
+ box_xyxy = box_xyxy.astype(int)
+ box_xywh = to_xywh(box_xyxy)
+ if box_xywh[2] >= self.cfg.min_width:
+ boxes_xywh.append(box_xywh)
+ boxes_xyxy.append(box_xyxy)
+ confs.append(float(conf))
+ cls.append(int(cl))
+ return img, boxes_xywh, boxes_xyxy, keypoints, confs, cls
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+from math import dist
+
+
+
+[docs]
+class Centroid(object):
+
+ """Class reprensented a Centroid tracker.
+ """
+
+
+[docs]
+ def __init__(self, cfg):
+ """Initialize according to the given :obj:`cfg` and :obj:`auto_load`.
+
+ Parameters
+ ----------
+ cfg : TCFG_Centroid
+ A :class:`TCFG_Centroid` object which manages the configurations of tracker Centroid.
+ """
+ self.previous_ct = []
+ self.previous_id = []
+ self.current_ct = []
+ self.current_id = []
+ self.max_spread = cfg.max_spread
+ self.pref_y = cfg.pref_y
+
+
+ def _generateID(self):
+ self.used_ids = list(set(self.used_ids))
+ if len(self.used_ids) == 0: aID = 0
+ else: aID = max(self.used_ids) + 1
+ self.used_ids.append(aID)
+ return aID
+
+ def _findRepspoint(self, box_xyxy):
+ x = int((box_xyxy[0] + box_xyxy[2]) / 2)
+ y = 0
+ if self.pref_y.lower() == "center":
+ y = int((box_xyxy[1] + box_xyxy[3]) / 2)
+ elif self.pref_y.lower() == "bottom":
+ y = max(box_xyxy[1], box_xyxy[3])
+ else:
+ y = min(box_xyxy[1], box_xyxy[3])
+ return (x, y)
+
+ def _findPID(self, point):
+ pindex = -1
+ min_d = 8192
+ i = 0
+ for ct in self.previous_ct:
+ d = dist(point, ct)
+ if d < min_d:
+ min_d = d
+ pindex = i
+ i += 1
+ if min_d > self.max_spread: pindex = -1
+ return pindex
+
+
+[docs]
+ def update(self, boxes_xyxy, boxes_confs, boxes_cls=None, img=None):
+ """Update the tracker and return a track list.
+
+ Parameters
+ ----------
+ boxes_xyxy : list[[X1, Y1, X2, Y2], ...]
+ A list of boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ boxes_confs : list[float, ...]
+ Being consistent with other trackers, will be ignored.
+ boxes_cls : list[int, ...], default=None
+ Being consistent with other trackers, will be ignored.
+ img : any, default=None
+ Being consistent with other trackers, will be ignored.
+
+ Returns
+ -------
+ list[[X1, Y1, X2, Y2], ...]
+ A list of boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ list[int]
+ A list of IDs corresponding to the the list of boxes.
+ """
+
+ self.previous_ct = self.current_ct
+ self.previous_id = self.current_id
+
+ self.current_id = []
+ self.used_ids = []
+
+ self.current_ct = [self._findRepspoint(b) for b in boxes_xyxy]
+ hang_indexes_in_clist = []
+ len_bb = len(boxes_xyxy)
+ if len_bb > 0:
+ for i in range(0, len_bb):
+ self.current_id.append(-1)
+ pindex = self._findPID(self.current_ct[i])
+ if pindex >= 0:
+ prev_id = self.previous_id[pindex]
+ if prev_id in self.used_ids:
+ hang_indexes_in_clist.append(i)
+ else:
+ self.current_id[i] = prev_id
+ self.used_ids.append(prev_id)
+ else:
+ hang_indexes_in_clist.append(i)
+ len_hlist = len(hang_indexes_in_clist)
+ if len_hlist > 0:
+ for index in hang_indexes_in_clist:
+ self.current_id[index] = self._generateID()
+
+ return boxes_xyxy, self.current_id
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import numpy as np
+
+from vsensebox.utils.commontools import to_xywh
+from vsensebox.utils.logtools import ignore_this_logger
+
+ignore_this_logger("tensorflow")
+ignore_this_logger("preprocessing")
+ignore_this_logger("nn_matching")
+ignore_this_logger("generate_detections")
+ignore_this_logger("detection")
+ignore_this_logger("tracker")
+
+from .utils import preprocessing
+from .utils import nn_matching
+from .utils import generate_detections as gdet
+from .utils.detection import Detection as DSDetection
+from .utils.tracker import Tracker as DSTracker
+from .utils.box_matching import matchDetTrkByXYXY
+
+
+
+[docs]
+class DeepSORT(object):
+
+ """Class used as a custom layer or interface for interacting with DeepSORT tracker.
+ """
+
+
+[docs]
+ def __init__(self, cfg):
+ """Initialize according to the given :obj:`cfg` and :obj:`auto_load`.
+
+ Parameters
+ ----------
+ cfg : TCFG_DeepSORT
+ A :class:`TCFG_DeepSORT` object which manages the configurations of tracker DeepSORT.
+ """
+ self.nms_max_overlap = cfg.nms_max_overlap
+ self.encoder = gdet.create_box_encoder(cfg.model_file, batch_size=cfg.batch_size)
+ self.metric = nn_matching.NearestNeighborDistanceMetric(
+ "cosine", cfg.max_cosine_distance,cfg.nn_budget
+ )
+ self.tracker = DSTracker(self.metric)
+
+
+
+
+[docs]
+ def update(self, boxes_xyxy, boxes_confs, boxes_cls=None, img=None):
+ """Update the tracker and return a track list.
+
+ Parameters
+ ----------
+ boxes_xyxy : list[[X1, Y1, X2, Y2], ...]
+ A list of boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ boxes_confs : list[float, ...]
+ A list of detection confidences corresponding to boxes_xyxy.
+ boxes_cls : list[int, ...], default=None
+ A list of detection class corresponding to boxes_xyxy.
+ img : any, default=None
+ A :obj:`Mat` like object.
+
+ Returns
+ -------
+ list[[X1, Y1, X2, Y2], ...]
+ A list of boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ list[int]
+ A list of IDs corresponding to the the list of boxes.
+ """
+
+ dconfidences = boxes_confs
+ dclasses = boxes_cls
+ dboxes = [to_xywh(b) for b in boxes_xyxy]
+ dfeatures = self.encoder(img, dboxes)
+ detections = [DSDetection(dbox, dconfidence, dclass, dfeature)
+ for dbox, dconfidence, dclass, dfeature in
+ zip(dboxes, dconfidences, dclasses, dfeatures)]
+ dboxes = np.array([d.tlwh for d in detections])
+ scores = np.array([d.confidence for d in detections])
+ indices = preprocessing.non_max_suppression(dboxes, self.nms_max_overlap, scores)
+ detections = [detections[i] for i in indices]
+
+ self.tracker.predict()
+ self.tracker.update(detections)
+
+ boxes_xyxy_trk = []
+ ids_trk = []
+ for t in self.tracker.tracks:
+ if not t.is_confirmed() or t.time_since_update > 1:
+ continue
+ bxyxy = t.to_tlbr().tolist()
+ boxes_xyxy_trk.append(bxyxy)
+ ids_trk.append(int(t.track_id))
+ ids = matchDetTrkByXYXY(boxes_xyxy, boxes_xyxy_trk, ids_trk)
+
+ return boxes_xyxy, ids
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import numpy as np
+from vsensebox.utils.logtools import ignore_this_logger
+
+ignore_this_logger("sort")
+from .utils.sort import Sort as ST
+from .utils.box_matching import matchDetTrkByXYXYForSORT
+
+
+
+[docs]
+class SORT(object):
+
+ """Class used as a custom layer or interface for interacting with SORT tracker.
+ """
+
+
+[docs]
+ def __init__(self, cfg):
+ """Initialize according to the given :obj:`cfg` and :obj:`auto_load`.
+
+ Parameters
+ ----------
+ cfg : TCFG_SORT
+ A :class:`TCFG_SORT` object which manages the configurations of tracker SORT.
+ """
+ self.st = ST(cfg.max_age, cfg.min_hits, cfg.iou_threshold)
+
+
+
+[docs]
+ def update(self, boxes_xyxy, boxes_confs, boxes_cls=None, img=None):
+ """Update the tracker and return a track list.
+
+ Parameters
+ ----------
+ boxes_xyxy : list[[X1, Y1, X2, Y2], ...]
+ A list of boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ boxes_confs : list[float, ...]
+ A list of detection confidence corresponding to boxes_xyxy.
+ boxes_cls : list[int, ...], default=None
+ Being consistent with other trackers, will be ignored.
+ img : any, default=None
+ Being consistent with other trackers, will be ignored.
+
+ Returns
+ -------
+ list[[X1, Y1, X2, Y2], ...]
+ A list of boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ list[int]
+ A list of IDs corresponding to the the list of boxes.
+ """
+ detection = []
+ for b, c in zip(boxes_xyxy, boxes_confs):
+ b = np.append(b, c)
+ detection.append(b)
+ track = self.st.update(np.array(detection)).astype(int)
+ ids = matchDetTrkByXYXYForSORT(boxes_xyxy, track, max_spread=10)
+ return boxes_xyxy, ids
+
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import os
+import time
+import logging
+
+_TIMESTAMP = str(time.strftime("%Y%m%d_%H%M%S"))
+_VSENSEBOX_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir))
+_LOG_DIR = os.path.join(_VSENSEBOX_ROOT, "data/logs").replace(os.sep, '/')
+_LOG_TXT_PATH = os.path.join(_LOG_DIR, "log_" + _TIMESTAMP + ".txt")
+_MAX_AGE = 86400 * 1 # 1 DAY
+
+# Remove old logs
+if os.path.exists(_LOG_DIR):
+ for filename in os.listdir(_LOG_DIR):
+ if "git" in filename: continue
+ filestamp = os.stat(os.path.join(_LOG_DIR, filename)).st_mtime
+ if filestamp < time.time() - _MAX_AGE:
+ os.remove(os.path.join(_LOG_DIR, filename))
+else: os.makedirs(_LOG_DIR)
+
+# Initial logger
+logging.basicConfig(
+ filename=_LOG_TXT_PATH,
+ filemode='a',
+ format='%(asctime)s %(levelname)-3s %(message)-3s',
+ datefmt='%H:%M:%S',
+ level=logging.INFO
+)
+
+# Add header
+with open(_LOG_TXT_PATH, 'w+') as log_txt:
+ log_txt.write("-------------------------------------------------")
+ log_txt.write("-------------------------------------------------\n")
+ log_txt.write("#################################################")
+ log_txt.write("#################################################\n")
+ log_txt.write("-------------------------------------------------")
+ log_txt.write("-------------------------------------------------\n")
+
+# Global
+__this_logger__ = logging.getLogger(__name__)
+__this_logger__.info(": Here we go!")
+__TRUE__ = True
+
+
+#############################################################################
+
+
+def add_warning_log(msg, terminal_log=__TRUE__, add_new_line=True):
+ """
+ :meta private:
+ """
+ global __this_logger__
+ if terminal_log: print(msg)
+ if add_new_line: msg = ': \n' + str(msg)
+ else: msg = ': ' + str(msg)
+ __this_logger__.warning(msg)
+
+def add_info_log(msg, terminal_log=__TRUE__, add_new_line=False):
+ """
+ :meta private:
+ """
+ global __this_logger__
+ if terminal_log: print(msg)
+ if add_new_line: msg = ': \n' + str(msg)
+ else: msg = ': ' + str(msg)
+ __this_logger__.info(msg)
+
+def add_error_log(msg, terminal_log=__TRUE__, add_new_line=True):
+ """
+ :meta private:
+ """
+ global __this_logger__
+ if terminal_log: print(msg)
+ if add_new_line: msg = ': \n' + str(msg)
+ else: msg = ': ' + str(msg)
+ __this_logger__.error(msg)
+
+def ignore_this_logger(name, level=logging.ERROR):
+ """
+ :meta private:
+ """
+ logger_to_ignore = logging.getLogger(name)
+ logger_to_ignore.setLevel(level)
+
+def disable_this_logger(name, level=logging.ERROR):
+ """
+ :meta private:
+ """
+ ignore_this_logger(name=name, level=level)
+ logger_to_disable = logging.getLogger(name)
+ logger_to_disable.disabled = True
+
+
+[docs]
+def disable_terminal_log():
+ """
+ Disable all console or terminal logging of VSenseBox.
+ """
+ global __TRUE__
+ __TRUE__ = False
+
+
+
+[docs]
+def enable_terminal_log():
+ """
+ Enable all console or terminal logging of VSenseBox.
+ """
+ global __TRUE__
+ __TRUE__ = True
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+import cv2
+
+from .commontools import getCVMat
+from .logtools import add_warning_log
+
+
+[docs]
+def draw_boxes(img,
+ ids=[],
+ show_ids=True,
+ boxes_xyxy=[],
+ boxes_confs=[],
+ boxes_color=(255, 255, 0),
+ boxes_thickness=2,
+ text_color=(150, 200, 50),
+ text_font=cv2.FONT_HERSHEY_COMPLEX_SMALL,
+ text_font_scale=1,
+ text_thickness=1,
+ img_is_mat=True):
+ """Draw the boxes add other decorations for the detected objects in the given image.
+
+ Parameters
+ ----------
+ img : str or Mat
+ An image file or a :obj:`Mat` like object.
+ show_ids : bool, default=True
+ Whether to visualize the IDs.
+ boxes_xyxy : list[[X1, Y1, X2, Y2], ...], default=[]
+ A list of bounding boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ boxes_confs : list[float, ...], default=[]
+ A list of detection confidences corresponding to bounding boxes.
+ boxes_color : tuple(int, int, int), default=(255, 255, 0)
+ Color space of bounding boxes.
+ boxes_thickness : int, default=2
+ Thickness of bounding boxes.
+ text_color : tuple(int, int, int), default=(150, 200, 50)
+ Color space of the texts for IDs and confs.
+ text_font : int, default=cv2.FONT_HERSHEY_COMPLEX_SMALL
+ Font of the texts for IDs and confs.
+ text_font_scale : int, default=1
+ Font scale of the texts for IDs and confs.
+ text_thickness : int, default=2
+ Thickness of the texts for IDs and confs.
+ img_is_mat : bool, default=True
+ Speed up the function by telling whether the :obj:`img` is :obj:`Mat` like object.
+
+ Returns
+ -------
+ Mat
+ A visualized :obj:`Mat` like object.
+ """
+ if img_is_mat and isinstance(img, str): img_is_mat = False
+ if not img_is_mat: img = getCVMat(img)
+ len_boxes = len(boxes_xyxy)
+ if show_ids and len_boxes != len(ids) and len(ids) > 0:
+ add_warning_log("VSenseBox:draw_boxes() -> Numbers of IDs and boxes are different!")
+ show_ids = False
+ if show_ids and len(ids) == 0:
+ ids = [i for i in range(0, len_boxes)]
+ for i in range(0, len_boxes):
+ cv2.rectangle(
+ img,
+ (int(boxes_xyxy[i][0]), int(boxes_xyxy[i][1])), (int(boxes_xyxy[i][2]), int(boxes_xyxy[i][3])),
+ boxes_color,
+ boxes_thickness
+ )
+ conf = 0.0
+ try:
+ conf = format(boxes_confs[i], '.2f')
+ cv2.putText(img, str(conf), (int(boxes_xyxy[i][0]) + 5, int(boxes_xyxy[i][3]) - 5),
+ text_font, text_font_scale, text_color, text_thickness)
+ except Exception as e:
+ add_warning_log("VSenseBox:draw_boxes() -> Numbers of confs and boxes are different!")
+ if show_ids:
+ cv2.putText(img, "#" + str(ids[i]), (int(boxes_xyxy[i][0]), int(boxes_xyxy[i][1] - 5)),
+ text_font, text_font_scale, text_color, text_thickness)
+ return img
+
+
+# VSenseBox - Python toolbox for visual sensing
+# GNU General Public License v3 or later (GPLv3+)
+# Copyright (C) 2024 UMONS-Numediart
+
+
+# Configurations
+from vsensebox.config.configurator import DET_CONFIG_DIR, TRK_CONFIG_DIR, IN_ROOT_DIR
+from vsensebox.config.confighelper import getCFGDict
+
+# Utils & tools
+from vsensebox.modules.detectors import checkDet
+from vsensebox.modules.trackers import checkTrk
+from vsensebox.utils.commontools import getCVMat, joinFPathFull, getAncestorDir
+
+# Default detector
+DEFAULT_DET_CONFIG = getCFGDict(joinFPathFull(DET_CONFIG_DIR, 'default.yaml'))
+DEFAULT_DET_YAML = joinFPathFull(IN_ROOT_DIR, DEFAULT_DET_CONFIG['config_yaml'])
+DET_YAML_TO_ROOT = True if getAncestorDir(DEFAULT_DET_CONFIG['config_yaml']) == 'config/detectors' else False
+
+# Default tracker
+DEFAULT_TRK_CONFIG = getCFGDict(joinFPathFull(TRK_CONFIG_DIR, 'default.yaml'))
+DEFAULT_TRK_YAML = joinFPathFull(IN_ROOT_DIR, DEFAULT_TRK_CONFIG['config_yaml'])
+TRK_YAML_TO_ROOT = True if getAncestorDir(DEFAULT_TRK_CONFIG['config_yaml']) == 'config/trackers' else False
+
+
+
+[docs]
+class VSense(object):
+
+ """
+ VSense is used to operate the object detection and tracking.
+
+ Attributes
+ ----------
+ assets : VSenseAssets
+ A :class:`VSenseAssets` object used to store assets of VSense.
+ """
+
+ def __init__(self):
+ self.assets = VSenseAssets()
+ self._detector = None
+ self._tracker = None
+ self._yaml_det = None
+ self._yaml_trk = None
+ self._det_rel_to_root = False
+ self._trk_rel_to_root = False
+
+
+[docs]
+ def detect(self, img, config_yaml=None, img_is_mat=False):
+ """Detect objects in the given image :obj:`img`.
+
+ Parameters
+ ----------
+ img : str or Mat
+ Image file or a :obj:`Mat` like object.
+ config_yaml : str, default=None
+ Path of YAML config file.
+ img_is_mat : bool, default=False
+ Speed up the function by telling whether the :obj:`img` is :obj:`Mat` like object.
+ """
+ if not img_is_mat: img = getCVMat(img)
+ if config_yaml is None:
+ config_yaml = DEFAULT_DET_YAML
+ self._det_rel_to_root = DET_YAML_TO_ROOT
+ if self._yaml_det != config_yaml:
+ self._detector = checkDet(detector=self._detector, config_yaml=config_yaml,
+ relative_to_vsensebox_root=self._det_rel_to_root)
+ else:
+ self._yaml_det = config_yaml
+ img, boxes_xywh, boxes_xyxy, keypoints, confs, cls = self._detector.detect(img)
+ self.assets.update(
+ boxes_xywh=boxes_xywh,
+ boxes_xyxy=boxes_xyxy,
+ keypoints=keypoints,
+ boxes_confs=confs,
+ boxes_cls=cls
+ )
+
+
+
+[docs]
+ def track(self, img=None, config_yaml=None, img_is_mat=False):
+ """Track the detected objects in the given image :obj:`img`.
+
+ Parameters
+ ----------
+ img : str or Mat
+ Image file or a :obj:`Mat` like object.
+ config_yaml : str, default=None
+ Path of YAML config file.
+ img_is_mat : bool, default=False
+ Speed up the function by telling whether the :obj:`img` is :obj:`Mat` like object.
+ """
+ if not img_is_mat: img = getCVMat(img)
+ if config_yaml is None:
+ config_yaml = DEFAULT_TRK_YAML
+ self._trk_rel_to_root = TRK_YAML_TO_ROOT
+ if self._yaml_trk != config_yaml:
+ self._tracker = checkTrk(tracker=self._tracker, config_yaml=config_yaml,
+ relative_to_vsensebox_root=self._trk_rel_to_root)
+ else:
+ self._yaml_trk = config_yaml
+ boxes_xyxy, self.assets.ids = self._tracker.update(
+ self.assets.boxes_xyxy,
+ self.assets.boxes_confs,
+ boxes_cls=self.assets.boxes_cls,
+ img=img
+ )
+
+
+
+
+
+[docs]
+class VSenseAssets(object):
+
+ """
+ A class used to store assets of VSense.
+
+ Attributes
+ ----------
+ boxes_xyxy : list[[X1, Y1, X2, Y2], ...], optional
+ A list of bounding boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ boxes_xywh : list[[X, Y, W, H], ...], optional
+ A list of bounding boxes; for example, [[X, Y, W, H], [X, Y, W, H], ...].
+ boxes_confs : list[float, ...], optional
+ A list of detection confidences corresponding to bounding boxes.
+ boxes_cls : list[int, ...], optional
+ A list of detection class corresponding to bounding boxes.
+ keypoints : list[], optional
+ A list of detected bodies' keypoints.
+ ids: list[int, ...], optional
+ A list of IDs corresponding to bounding boxes.
+ masks: list[], optional
+ A list of detected masks.
+ misc : list[], optional
+ A list of miscellaneous items.
+ """
+
+ def __init__(self):
+ """Construct a VSenseAssets.
+ """
+ self.boxes_xyxy = []
+ self.boxes_xywh = []
+ self.boxes_confs = []
+ self.boxes_cls = []
+ self.keypoints = []
+ self.ids = []
+ self.masks = []
+ self.miscs = []
+
+
+[docs]
+ def update(self,
+ boxes_xyxy=[],
+ boxes_xywh=[],
+ boxes_confs=[],
+ boxes_cls=[],
+ keypoints=[],
+ ids=[],
+ masks=[],
+ miscs=[]):
+ """Update a VSenseAssets.
+
+ Parameters
+ ----------
+ boxes_xyxy : list[[X1, Y1, X2, Y2], ...], optional
+ A list of bounding boxes; for example, [[X1, Y1, X2, Y2], [X1, Y1, X2, Y2], ...].
+ boxes_xywh : list[[X, Y, W, H], ...], optional
+ A list of bounding boxes; for example, [[X, Y, W, H], [X, Y, W, H], ...].
+ boxes_confs : list[float, ...], optional
+ A list of detection confidences corresponding to bounding boxes.
+ boxes_cls : list[int, ...], optional
+ A list of detection class corresponding to bounding boxes.
+ keypoints : list[], optional
+ A list of detected bodies' keypoints.
+ ids: list[int, ...], optional
+ A list of IDs corresponding to bounding boxes.
+ masks: list[], optional
+ A list of detected masks.
+ misc : list[], optional
+ A list of miscellaneous items.
+ """
+ self.boxes_xyxy = boxes_xyxy
+ self.boxes_xywh = boxes_xywh
+ self.boxes_confs = boxes_confs
+ self.boxes_cls = boxes_cls
+ self.keypoints = keypoints
+ self.ids = ids
+ self.masks = masks
+ self.miscs = miscs
+
+
+
+