129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
import numpy as np
|
||
import requests
|
||
|
||
|
||
url = 'http://kb.athene.tech/api/1.0/ontology/'
|
||
|
||
|
||
def is_ontology_exists(ontology_uid: str, url: str) -> bool:
|
||
'''
|
||
Проверяет, существует ли онтология в сервисе.
|
||
@param ontology_uid: УИД онтологии.
|
||
@param url: Базовый URL сервиса.
|
||
'''
|
||
list_ontologies = requests.get(url).json()
|
||
list_ontologies = list_ontologies['response']['items']
|
||
for onto in list_ontologies:
|
||
if onto['uid'] == ontology_uid:
|
||
return True
|
||
return False
|
||
|
||
|
||
def upload_ontology(name: str, file_buf) -> str:
|
||
"""
|
||
Загружает файл.
|
||
:param name: Имя файла.
|
||
:param file_buf: Содержимое файла.
|
||
:return: УИД загруженной онтологии.
|
||
"""
|
||
files = {'file': file_buf}
|
||
response = requests.post(f'{url}?name={name}', files=files)
|
||
response = response.json()
|
||
return response['response']['uid']
|
||
|
||
|
||
def delete_ontology(ontology_uid: str) -> bool:
|
||
"""
|
||
Загружает файл.
|
||
:param name: Имя файла.
|
||
:param file_buf: Содержимое файла.
|
||
:return: УИД загруженной онтологии.
|
||
"""
|
||
response = requests.delete(url + ontology_uid)
|
||
response = response.json()
|
||
return False if response['error'] else True
|
||
|
||
|
||
def rename_entity(list_names: dict) -> dict:
|
||
'''
|
||
Нормализация названий объектов.
|
||
@param list_names: Список названий объектов.
|
||
'''
|
||
temp_list = list()
|
||
for entity in list_names.values():
|
||
entity: str
|
||
temp_list.append(entity.title().replace(' ', ''))
|
||
return temp_list
|
||
|
||
|
||
def get_entity_square(width: float, height: float) -> float:
|
||
'''
|
||
Получение площади занимаемой области.
|
||
@param width: Ширина области в px.
|
||
@param height: Высота области в px.
|
||
'''
|
||
return abs(width * height)
|
||
|
||
|
||
def get_request_data(entities: dict, objects: np.ndarray, confs: np.ndarray, boxes: np.ndarray) -> tuple[list, list]:
|
||
'''
|
||
Формирование данных для сервиса онтологий.
|
||
@param entities: Список имён объектов.
|
||
@param results_ndarray: Результат распознавания объектов.
|
||
'''
|
||
classroom = 'classroom'
|
||
entities = rename_entity(entities)
|
||
object_properties = list()
|
||
data_properties = list()
|
||
|
||
for entity_idx, entity in enumerate(entities):
|
||
if (entity_idx in objects):
|
||
object_properties.append(
|
||
{'domain': entity, 'property': 'locatedIn', 'range': classroom})
|
||
else:
|
||
object_properties.append(
|
||
{'domain': entity, 'property': 'notLocatedIn', 'range': classroom})
|
||
|
||
for object_idx, object in enumerate(objects):
|
||
conf = confs[object_idx]
|
||
box = boxes[object_idx]
|
||
entity = entities[object.item()]
|
||
data_properties.append(
|
||
{'domain': entity,
|
||
'property': 'hasArea',
|
||
'value': get_entity_square(float(box[2]), float(box[3]))})
|
||
data_properties.append(
|
||
{'domain': entity,
|
||
'property': 'hasConfidence',
|
||
'value': float(conf)})
|
||
|
||
return object_properties, data_properties
|
||
|
||
|
||
def analyze(ontology_uid: str, object_properties: list, data_properties: list, queries: list[str]) -> tuple[2]:
|
||
'''
|
||
Базовая функция анализа.
|
||
@param ontology_uid: УИД онтологии.
|
||
@param object_properties: Объектные свойства.
|
||
@param data_properties: Свойства данных.
|
||
@param queries: Список запросов для запуска.
|
||
'''
|
||
if not is_ontology_exists(ontology_uid, url):
|
||
raise Exception(f'Онтология с uid {ontology_uid} не существует')
|
||
|
||
# Формирование данных для запроса к сервису работы с онтологиями.
|
||
data = {
|
||
'data':
|
||
{
|
||
'objectPropertyAssertions': object_properties,
|
||
'dataPropertyAssertions': data_properties
|
||
}
|
||
}
|
||
params = '&'.join([f'names={query}' for query in queries])
|
||
|
||
# Выполнение запроса.
|
||
response = requests.post(
|
||
url + f'{ontology_uid}/query/multi?{params}', json=data).json()
|
||
|
||
return response
|