-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
add: [core] Adding BOW ORB - Also use as an example commit of "how to…
… add an algorithm to the library". Very very useful and important commit ! :)
- Loading branch information
1 parent
86da5c9
commit 9942a00
Showing
6 changed files
with
263 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,108 @@ | ||
#!/usr/bin/env python3 | ||
# -*- coding: utf-8 -*- | ||
|
||
|
||
import logging | ||
import traceback | ||
from typing import Dict, List | ||
|
||
import cv2 | ||
|
||
import carlhauser_server.Configuration.database_conf as database_conf | ||
import carlhauser_server.Configuration.distance_engine_conf as distance_engine_conf | ||
import carlhauser_server.Configuration.feature_extractor_conf as feature_extractor_conf | ||
import carlhauser_server.DistanceEngine.scoring_datastrutures as sd | ||
from carlhauser_server.Configuration.algo_conf import Algo_conf | ||
from common.CustomException import AlgoFeatureNotPresentError | ||
from carlhauser_server.DistanceEngine.distance_hash import Distance_Hash as dist_hash | ||
|
||
|
||
class Distance_BoW_ORB: | ||
def __init__(self, db_conf: database_conf.Default_database_conf, dist_conf: distance_engine_conf.Default_distance_engine_conf, fe_conf: feature_extractor_conf.Default_feature_extractor_conf): | ||
# STD attributes | ||
self.logger = logging.getLogger(__name__) | ||
self.logger.info("Creation of a Distance ORB Engine") | ||
|
||
# Save configuration | ||
self.db_conf = db_conf # TODO : REMOVE = NOT USEFUL FOR NOW ! | ||
self.dist_conf = dist_conf | ||
self.fe_conf = fe_conf | ||
|
||
# TODO : | ||
self.orb_matcher = cv2.BFMatcher(cv2.NORM_HAMMING, crossCheck=dist_conf.CROSSCHECK) | ||
|
||
def bow_orb_distance(self, pic_package_from: Dict, pic_package_to: Dict) -> Dict[str, sd.AlgoMatch]: | ||
""" | ||
Distance between two provided pictures (dicts) with BoW-ORB methods | ||
:param pic_package_from: first picture dict | ||
:param pic_package_to: second picture dict | ||
:return: A dictionary of algo name to the match detail (distance, decision ..) | ||
""" | ||
|
||
answer = {} | ||
self.logger.info("BoW-Orb distance computation ... ") | ||
|
||
# Verify if what is needed to compute it is present | ||
if pic_package_from.get("BOW_ORB_DESCRIPTOR", None) is None \ | ||
or pic_package_to.get("BOW_ORB_DESCRIPTOR", None) is None: | ||
self.logger.warning(f"BoW-ORB descriptors are NOT presents in the results.") | ||
raise AlgoFeatureNotPresentError("None BoW-ORB descriptors in orb distance.") | ||
|
||
# Add result for enabled algorithms | ||
try: | ||
if self.fe_conf.BOW_ORB.get("is_enabled", False): | ||
answer = self.add_results(self.fe_conf.ORB, pic_package_from, pic_package_to, answer) | ||
|
||
except Exception as e: | ||
self.logger.error(traceback.print_tb(e.__traceback__)) | ||
self.logger.error("Error during bow-orb distance calculation : " + str(e)) | ||
|
||
return answer | ||
|
||
def add_results(self, algo_conf: Algo_conf, pic_package_from: Dict, pic_package_to: Dict, answer: Dict) -> Dict: | ||
""" | ||
Add results to answer dict, depending on the algorithm name we want to compute | ||
Ex : Input {} -> Output {"BOW_ORB":{"name":"BOW_ORB", "distance":0.3,"decision":YES}} | ||
:param algo_conf: An algorithm configuration (to specify which algorithm to launch) | ||
:param pic_package_from: first picture dict | ||
:param pic_package_to: second picture dict | ||
:param answer: Current dict of algo_name to algo match (will be updated and returned) | ||
:return: a dict of algo_name to algo match | ||
""" | ||
|
||
algo_name = algo_conf.get('algo_name') | ||
|
||
# Depending on the type of | ||
self.logger.debug(f"Comparison for BOW : {self.dist_conf.BOW_CMP_HIST} of {type(self.dist_conf.BOW_CMP_HIST)} " | ||
f"and {distance_engine_conf.BOW_CMP_HIST.CORREL.name} of {type(distance_engine_conf.BOW_CMP_HIST.CORREL.name)}") | ||
|
||
if self.dist_conf.BOW_CMP_HIST == distance_engine_conf.BOW_CMP_HIST.CORREL.name: | ||
tmp_dist = 1 - cv2.compareHist(pic_package_from["BOW_ORB_DESCRIPTOR"], | ||
pic_package_to["BOW_ORB_DESCRIPTOR"], | ||
cv2.HISTCMP_CORREL) | ||
elif self.dist_conf.BOW_CMP_HIST == distance_engine_conf.BOW_CMP_HIST.BHATTACHARYYA.name: | ||
tmp_dist = cv2.compareHist(pic_package_from["BOW_ORB_DESCRIPTOR"], | ||
pic_package_to["BOW_ORB_DESCRIPTOR"], | ||
cv2.HISTCMP_BHATTACHARYYA) | ||
else: | ||
raise Exception('BOW ORB : HISTOGRAM COMPARISON MODE INCORRECT') | ||
|
||
# Add the distance as an AlgoMatch | ||
answer[algo_name] = sd.AlgoMatch(name=algo_name, | ||
distance=tmp_dist, | ||
decision=self.compute_decision_from_distance(algo_conf, tmp_dist)) | ||
return answer | ||
|
||
# ==================== ------ DECISIONS ------- ==================== | ||
|
||
@staticmethod | ||
def compute_decision_from_distance(algo_conf: Algo_conf, dist: float) -> sd.DecisionTypes: | ||
""" | ||
From a distance between orb distance, gives a decision : is it a match or not ? Or maybe ? | ||
# TODO : Evolve to more complex calculation if needed for ORB ! | ||
:param algo_conf: An algorithm configuration (to specify which algorithm to launch) | ||
:param dist: a distance between two pictures | ||
:return: a decision (YES,MAYBE,NO) | ||
""" | ||
|
||
return dist_hash.compute_decision_from_distance(algo_conf, dist) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,90 @@ | ||
#!/usr/bin/env python3 | ||
# -*- coding: utf-8 -*- | ||
|
||
|
||
import logging | ||
import cv2 | ||
from typing import Dict | ||
import numpy as np | ||
import carlhauser_server.Configuration.feature_extractor_conf as feature_extractor_conf | ||
from common.environment_variable import load_server_logging_conf_file | ||
from carlhauser_server.Helpers.bow_orb_vocabulary_creator import BoWOrb_Vocabulary_Creator | ||
load_server_logging_conf_file() | ||
|
||
|
||
class Picture_BoW_Orber: | ||
def __init__(self, fe_conf: feature_extractor_conf.Default_feature_extractor_conf): | ||
# STD attributes | ||
self.fe_conf: feature_extractor_conf.Default_feature_extractor_conf = fe_conf | ||
self.logger = logging.getLogger(__name__) | ||
self.logger.info("Creation of a Picture BoW Orber") | ||
|
||
self.algo = cv2.ORB_create(nfeatures=fe_conf.ORB_KEYPOINTS_NB) | ||
# TODO : Dictionnary path / Vocabulary | ||
self.bow_descriptor = cv2.BOWImgDescriptorExtractor(self.algo, cv2.BFMatcher(cv2.NORM_HAMMING)) | ||
vocab = BoWOrb_Vocabulary_Creator.load_vocab_from_file(fe_conf.BOW_VOCAB_PATH) | ||
self.bow_descriptor.setVocabulary(vocab) | ||
|
||
''' | ||
def create_dict_from_folder(self, folder_path: pathlib.Path()): | ||
self.bow_trainer = cv2.BOWKMeansTrainer(self.conf.BOW_SIZE) | ||
for curr_image in picture_list: | ||
self.bow_trainer.add(np.float32(curr_image.description)) | ||
self.vocab = self.bow_trainer.cluster().astype(picture_list[0].description.dtype) | ||
def compute_distance(): | ||
if self.conf.BOW_CMP_HIST == configuration.BOW_CMP_HIST.CORREL: | ||
dist = 1 - cv2.compareHist(pic1.description, pic2.description, cv2.HISTCMP_CORREL) | ||
elif self.conf.BOW_CMP_HIST == configuration.BOW_CMP_HIST.BHATTACHARYYA: | ||
dist = cv2.compareHist(pic1.description, pic2.description, cv2.HISTCMP_BHATTACHARYYA) | ||
else: | ||
raise Exception('BOW WRAPPER : HISTOGRAM COMPARISON MODE INCORRECT') | ||
''' | ||
|
||
def bow_orb_picture(self, curr_picture, orb_dict: Dict): | ||
""" | ||
BoW-Orb a picture and returns the BoW-orb value | ||
:param curr_picture: the picture to bow-orb | ||
:return: the BoW-orb version of the picture | ||
""" | ||
answer = {} | ||
self.logger.info("BoW-Orbing picture ... ") | ||
|
||
# Convert from cv's BRG default color order to RGB | ||
# image = cv2.imread(str(path)) | ||
# image = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) | ||
self.logger.debug(f"Original type {type(curr_picture)}") | ||
arr = np.asarray(bytearray(curr_picture), dtype=np.uint8) | ||
orb_pic = cv2.imdecode(arr, -1) | ||
self.logger.debug(f"Picture converted to CV2 UMAT {type(orb_pic)}") | ||
|
||
# Get keypoints from orb dictionnary OR compute it if not present | ||
key_points = orb_dict.get("ORB_KEYPOINTS", None) | ||
if key_points is None or key_points == []: | ||
self.logger.warning(f"No Keypoints in provided ORB dictionnary.") | ||
try: | ||
self.logger.info(f"Computing Orb Keypoints in BoW-orber.") | ||
|
||
# Compute keypoints by itself | ||
key_points, _ = self.algo.detectAndCompute(orb_pic, None) | ||
|
||
if key_points is None or key_points == []: | ||
raise Exception("NO KEYPOINTS") | ||
|
||
except Exception as e: | ||
self.logger.error(f"Impossible to compute keypoints in BoW-Orber for provided picture : {e}") | ||
raise e | ||
|
||
if self.fe_conf.BOW_ORB.get("is_enabled", False): | ||
try: | ||
description = self.bow_descriptor.compute(orb_pic, key_points) | ||
self.logger.warning(f"TYPE descriptor : {type(description)}") | ||
answer["BOW_ORB_DESCRIPTOR"] = description | ||
except Exception as e: | ||
self.logger.error("Error during BoW-orbing : " + str(e)) | ||
raise e | ||
|
||
return answer |