|
|
|
|
|
import numpy as np |
|
|
import logging |
|
|
|
|
|
|
|
|
logging.basicConfig( |
|
|
level=logging.DEBUG, |
|
|
format="%(asctime)s - %(levelname)s - %(message)s", |
|
|
filename="logs.log", |
|
|
) |
|
|
|
|
|
|
|
|
logger = logging.getLogger("pipline") |
|
|
|
|
|
def get_json_data(json_data,enhanced_images,detected_activity,distances_list): |
|
|
""" |
|
|
Args: |
|
|
json_data (json Array): get json data of image |
|
|
enhanced_images (list of numpy array): list of enhanced images |
|
|
detected_activity (list of strings): list of activities of person |
|
|
distances_list (lsit of integers): list of distances of each object |
|
|
|
|
|
Returns: |
|
|
results(json Array): contains all informations needed for frontend |
|
|
{'zoomed_img':np.array([]) , |
|
|
'actual_boxes':[], |
|
|
'merged_boundries':{}, |
|
|
} |
|
|
""" |
|
|
logger.info("get_json_data function is called... ") |
|
|
try: |
|
|
results = [] |
|
|
object_count = 0 |
|
|
activity_count = 0 |
|
|
for idx,box in enumerate(json_data): |
|
|
final_json_output = {'zoomed_img':np.array([]) , |
|
|
'actual_boxes':[], |
|
|
'merged_boundries':{}, |
|
|
} |
|
|
|
|
|
final_json_output['zoomed_img'] = enhanced_images[idx] |
|
|
final_json_output['merged_boundries'] = { "top_left": box['merged_boundries']['top_left'], |
|
|
"bottom_right": box['merged_boundries']['bottom_right']} |
|
|
|
|
|
for actual_box in box['actual_boundries']: |
|
|
|
|
|
single_object_details = {"top_left": actual_box['top_left'], |
|
|
"bottom_right": actual_box['bottom_right'], |
|
|
"class": actual_box['class'], |
|
|
"distance":distances_list[object_count], |
|
|
"activity":'none'} |
|
|
object_count+=1 |
|
|
|
|
|
if single_object_details['class'] == 'person': |
|
|
single_object_details['activity'] = detected_activity[activity_count] |
|
|
activity_count+=1 |
|
|
|
|
|
final_json_output['actual_boxes'].append(single_object_details) |
|
|
final_json_output = fix_distance(final_json_output) |
|
|
|
|
|
results.append(final_json_output) |
|
|
|
|
|
return results |
|
|
except Exception as e: |
|
|
logger.error("Something went wrong in generated results function") |
|
|
logger.error(e) |
|
|
|
|
|
|
|
|
def fix_distance(final_json_output): |
|
|
""" |
|
|
Args: |
|
|
final_json_output (json Array): array of json object |
|
|
|
|
|
Returns: |
|
|
final_json_output (json Array): array of json object |
|
|
""" |
|
|
logger.info("fix distance function is called... ") |
|
|
try: |
|
|
distances = [] |
|
|
DIFF = 90 |
|
|
|
|
|
for idx,box in enumerate(final_json_output['actual_boxes']): |
|
|
distances.append({'idx':idx,'distance':int(box['distance'][:-1])}) |
|
|
|
|
|
sorted_dist = sorted(distances, key=lambda d: d['distance']) |
|
|
sum_dist = [] |
|
|
idx= 0 |
|
|
sum_dist.append({'sum':sorted_dist[0]['distance'],'idxes':[sorted_dist[0]['idx']]}) |
|
|
|
|
|
for i in range(1,len(sorted_dist)): |
|
|
if abs(sorted_dist[i]['distance']-sorted_dist[i-1]['distance']) <=DIFF: |
|
|
sum_dist[idx]['sum']+= sorted_dist[i]['distance'] |
|
|
sum_dist[idx]['idxes'].append(sorted_dist[i]['idx']) |
|
|
|
|
|
else: |
|
|
sum_dist.append({'sum':sorted_dist[i]['distance'],'idxes':[sorted_dist[i]['idx']]}) |
|
|
idx+=1 |
|
|
|
|
|
|
|
|
for data in sum_dist: |
|
|
count = len(data['idxes']) |
|
|
mean = data['sum']//count |
|
|
for i in data['idxes']: |
|
|
final_json_output['actual_boxes'][i]['distance'] = str(mean)+'m' |
|
|
|
|
|
return final_json_output |
|
|
|
|
|
except Exception as e: |
|
|
logger.error("Something went wrong in get_json_data function...") |
|
|
logger.error(e) |
|
|
|
|
|
|
