6488803620

Committed 11 Oct 2023 10:42PM UTC coverage: 87.097% (+0.03%) from 87.07%

Build # 6488803620

Build Type

Pull #676

github-actions

Committed by

jchate6

Commit Message

Merge branch 'feature/tom2tom_data_sharing' into feature/tom-tom-target-sharing

Pull Request Pull Request #676: Feature/tom tom target sharing

Run Details

376 of 376 new or added lines in 14 files covered. (100.0%)

8181 of 9393 relevant lines covered (87.1%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

74.15

/tom_dataproducts/sharing.py

import requests
import os

from django.conf import settings
from django.core.exceptions import ImproperlyConfigured
from django.contrib import messages

from tom_targets.models import Target
from tom_dataproducts.models import DataProduct, ReducedDatum
from tom_dataproducts.alertstreams.hermes import publish_photometry_to_hermes, BuildHermesMessage, get_hermes_topics
from tom_dataproducts.serializers import DataProductSerializer, ReducedDatumSerializer


def share_data_with_hermes(share_destination, form_data, product_id=None, target_id=None, selected_data=None):
    """

    :param share_destination:
    :param form_data:
    :param product_id:
    :param target_id:
    :param selected_data:
    :return:
    """
    # Query relevant Reduced Datums Queryset
    accepted_data_types = ['photometry']
    if product_id:
        product = DataProduct.objects.get(pk=product_id)
        target = product.target
        reduced_datums = ReducedDatum.objects.filter(data_product=product)
    elif selected_data:
        reduced_datums = ReducedDatum.objects.filter(pk__in=selected_data)
        target = reduced_datums[0].target
    elif target_id:
        target = Target.objects.get(pk=target_id)
        data_type = form_data.get('data_type', 'photometry')
        reduced_datums = ReducedDatum.objects.filter(target=target, data_type=data_type)
    else:
        reduced_datums = ReducedDatum.objects.none()
        target = Target.objects.none()

    reduced_datums.filter(data_type__in=accepted_data_types)

    # Build and submit hermes table from Reduced Datums
    hermes_topic = share_destination.split(':')[1]
    destination = share_destination.split(':')[0]
    message_info = BuildHermesMessage(title=form_data.get('share_title',
                                                          f"Updated data for {target.name} from "
                                                          f"{getattr(settings, 'TOM_NAME','TOM Toolkit')}."),
                                      submitter=form_data.get('submitter'),
                                      authors=form_data.get('share_authors', None),
                                      message=form_data.get('share_message', None),
                                      topic=hermes_topic
                                      )
    # Run ReducedDatums Queryset through sharing protocols to make sure they are safe to share.
    filtered_reduced_datums = check_for_share_safe_datums(destination, reduced_datums, topic=hermes_topic)
    if filtered_reduced_datums.count() > 0:
        response = publish_photometry_to_hermes(message_info, filtered_reduced_datums)
    else:
        return {'message': f'ERROR: No valid data to share. (Check Sharing Protocol. Note that data types must be in '
                           f'{accepted_data_types})'}
    return response


def share_data_with_tom(share_destination, form_data, product_id=None, target_id=None, selected_data=None):
    """

    :param share_destination:
    :param form_data:
    :param product_id:
    :param target_id:
    :param selected_data:
    :return:
    """
    try:
        destination_tom_base_url = settings.DATA_SHARING[share_destination]['BASE_URL']
        username = settings.DATA_SHARING[share_destination]['USERNAME']
        password = settings.DATA_SHARING[share_destination]['PASSWORD']
    except KeyError as err:
        raise ImproperlyConfigured(f'Check DATA_SHARING configuration for {share_destination}: Key {err} not found.')
    auth = (username, password)
    headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}

    dataproducts_url = destination_tom_base_url + 'api/dataproducts/'
    targets_url = destination_tom_base_url + 'api/targets/'
    reduced_datums_url = destination_tom_base_url + 'api/reduceddatums/'
    reduced_datums = ReducedDatum.objects.none()

    if product_id:
        product = DataProduct.objects.get(pk=product_id)
        target = product.target
        serialized_data = DataProductSerializer(product).data
        destination_target_id, target_search_response = get_destination_target(target, targets_url, headers, auth)
        if destination_target_id is None:
            return {'message': 'ERROR: No matching target found.'}
        elif isinstance(destination_target_id, list) and len(destination_target_id) > 1:
            return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}
        serialized_data['target'] = destination_target_id
        # TODO: this should be updated when tom_dataproducts is updated to use django.core.storage
        dataproduct_filename = os.path.join(settings.MEDIA_ROOT, product.data.name)
        # Save DataProduct in Destination TOM
        with open(dataproduct_filename, 'rb') as dataproduct_filep:
            files = {'file': (product.data.name, dataproduct_filep, 'text/csv')}
            headers = {'Media-Type': 'multipart/form-data'}
            response = requests.post(dataproducts_url, data=serialized_data, files=files, headers=headers, auth=auth)
    elif selected_data or target_id:
        if selected_data:
            reduced_datums = ReducedDatum.objects.filter(pk__in=selected_data)
            targets = set(reduced_datum.target for reduced_datum in reduced_datums)
            target_dict = {}
            for target in targets:
                # get destination Target
                destination_target_id, target_search_response = get_destination_target(target,
                                                                                       targets_url,
                                                                                       headers,
                                                                                       auth)
                if isinstance(destination_target_id, list) and len(destination_target_id) > 1:
                    return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}
                target_dict[target.name] = destination_target_id
            if all(value is None for value in target_dict.values()):
                return {'message': 'ERROR: No matching targets found.'}
        else:
            target = Target.objects.get(pk=target_id)
            reduced_datums = ReducedDatum.objects.filter(target=target)
            destination_target_id, target_search_response = get_destination_target(target, targets_url, headers, auth)
            if destination_target_id is None:
                return {'message': 'ERROR: No matching target found.'}
            elif isinstance(destination_target_id, list) and len(destination_target_id) > 1:
                return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}
            target_dict = {target.name:  destination_target_id}
        response_codes = []
        reduced_datums = check_for_share_safe_datums(share_destination, reduced_datums)
        if not reduced_datums:
            return {'message': 'ERROR: No valid data to share.'}
        for datum in reduced_datums:
            if target_dict[datum.target.name]:
                serialized_data = ReducedDatumSerializer(datum).data
                serialized_data['target'] = target_dict[datum.target.name]
                serialized_data['data_product'] = ''
                if not serialized_data['source_name']:
                    serialized_data['source_name'] = settings.TOM_NAME
                    serialized_data['source_location'] = "TOM-TOM Direct Sharing"
                response = requests.post(reduced_datums_url, json=serialized_data, headers=headers, auth=auth)
                response_codes.append(response.status_code)
        failed_data_count = response_codes.count(500)
        if failed_data_count < len(response_codes):
            return {'message': f'{len(response_codes)-failed_data_count} of {len(response_codes)} '
                               'datums successfully saved.'}
        else:
            return {'message': 'ERROR: No valid data shared. These data may already exist in target TOM.'}
    else:
        return {'message': 'ERROR: No valid data to share.'}

    return response


def get_destination_target(target, targets_url, headers, auth):
    """
    Retrieve the target ID from a destination TOM
    :param target: Target Model
    :param targets_url: Destination API URL for TOM Target List
    :param headers:
    :param auth:
    :return:
    """
    target_names = ','.join(map(str, target.names))
    target_response = requests.get(f'{targets_url}?name_fuzzy={target_names}', headers=headers, auth=auth)
    target_response_json = target_response.json()
    try:
        if target_response_json['results']:
            if len(target_response_json['results']) > 1:
                return target_response_json['results'], target_response
            destination_target_id = target_response_json['results'][0]['id']
            return destination_target_id, target_response
        else:
            return None, target_response
    except KeyError:
        return None, target_response


def check_for_share_safe_datums(destination, reduced_datums, **kwargs):
    """
    Custom sharing protocols used to determine when data is shared with a destination.
    This example prevents sharing if a datum has already been published to the given Hermes topic.
    :param destination: sharing destination string
    :param reduced_datums: selected input datums
    :return: queryset of reduced datums to be shared
    """
    return reduced_datums
    # if 'hermes' in destination:
    #     message_topic = kwargs.get('topic', None)
    #     # Remove data points previously shared to the given topic
    #     filtered_datums = reduced_datums.exclude(Q(message__exchange_status='published')
    #                                              & Q(message__topic=message_topic))
    # else:
    #     filtered_datums = reduced_datums
    # return filtered_datums


def check_for_save_safe_datums():
    return


def get_sharing_destination_options():
    """
    Build the Display options and headers for the dropdown form for choosing sharing topics.
    Customize for a different selection experience.
    :return: Tuple: Possible Destinations and their Display Names
    """
    choices = []
    try:
        for destination, details in settings.DATA_SHARING.items():
            new_destination = [details.get('DISPLAY_NAME', destination)]
            if details.get('USER_TOPICS', None):
                # If topics exist for a destination (Such as HERMES) give topics as sub-choices
                #   for non-selectable Destination
                if destination == "hermes":
                    destination_topics = get_hermes_topics()
                else:
                    destination_topics = details['USER_TOPICS']
                topic_list = [(f'{destination}:{topic}', topic) for topic in destination_topics]
                new_destination.append(tuple(topic_list))
            else:
                # Otherwise just use destination as option
                new_destination.insert(0, destination)
            choices.append(tuple(new_destination))
    except AttributeError:
        pass
    return tuple(choices)


def sharing_feedback_handler(response, request):
    try:
        if 'message' in response.json():
            publish_feedback = response.json()['message']
        else:
            publish_feedback = f"ERROR: {response.text}"
    except AttributeError:
        publish_feedback = response['message']
    except ValueError:
        publish_feedback = f"ERROR: Returned Response code {response.status_code}"
    if "ERROR" in publish_feedback.upper():
        messages.error(request, publish_feedback)
    else:
        messages.success(request, publish_feedback)
    return

1	import requests	1✔
2	import os	1✔
3
4	from django.conf import settings	1✔
5	from django.core.exceptions import ImproperlyConfigured	1✔
6	from django.contrib import messages	1✔
7
8	from tom_targets.models import Target	1✔
9	from tom_dataproducts.models import DataProduct, ReducedDatum	1✔
10	from tom_dataproducts.alertstreams.hermes import publish_photometry_to_hermes, BuildHermesMessage, get_hermes_topics	1✔
11	from tom_dataproducts.serializers import DataProductSerializer, ReducedDatumSerializer	1✔
12
13
14	def share_data_with_hermes(share_destination, form_data, product_id=None, target_id=None, selected_data=None):	1✔
15	"""
16
17	:param share_destination:
18	:param form_data:
19	:param product_id:
20	:param target_id:
21	:param selected_data:
22	:return:
23	"""
24	# Query relevant Reduced Datums Queryset
25	accepted_data_types = ['photometry']	×
26	if product_id:	×
27	product = DataProduct.objects.get(pk=product_id)	×
28	target = product.target	×
29	reduced_datums = ReducedDatum.objects.filter(data_product=product)	×
30	elif selected_data:	×
31	reduced_datums = ReducedDatum.objects.filter(pk__in=selected_data)	×
32	target = reduced_datums[0].target	×
33	elif target_id:	×
34	target = Target.objects.get(pk=target_id)	×
35	data_type = form_data.get('data_type', 'photometry')	×
36	reduced_datums = ReducedDatum.objects.filter(target=target, data_type=data_type)	×
37	else:
38	reduced_datums = ReducedDatum.objects.none()	×
39	target = Target.objects.none()	×
40
41	reduced_datums.filter(data_type__in=accepted_data_types)	×
42
43	# Build and submit hermes table from Reduced Datums
44	hermes_topic = share_destination.split(':')[1]	×
45	destination = share_destination.split(':')[0]	×
46	message_info = BuildHermesMessage(title=form_data.get('share_title',	×
47	f"Updated data for {target.name} from "
48	f"{getattr(settings, 'TOM_NAME','TOM Toolkit')}."),
49	submitter=form_data.get('submitter'),
50	authors=form_data.get('share_authors', None),
51	message=form_data.get('share_message', None),
52	topic=hermes_topic
53	)
54	# Run ReducedDatums Queryset through sharing protocols to make sure they are safe to share.
55	filtered_reduced_datums = check_for_share_safe_datums(destination, reduced_datums, topic=hermes_topic)	×
56	if filtered_reduced_datums.count() > 0:	×
57	response = publish_photometry_to_hermes(message_info, filtered_reduced_datums)	×
58	else:
59	return {'message': f'ERROR: No valid data to share. (Check Sharing Protocol. Note that data types must be in '	×
60	f'{accepted_data_types})'}
61	return response	×
62
63
64	def share_data_with_tom(share_destination, form_data, product_id=None, target_id=None, selected_data=None):	1✔
65	"""
66
67	:param share_destination:
68	:param form_data:
69	:param product_id:
70	:param target_id:
71	:param selected_data:
72	:return:
73	"""
74	try:	1✔
75	destination_tom_base_url = settings.DATA_SHARING[share_destination]['BASE_URL']	1✔
76	username = settings.DATA_SHARING[share_destination]['USERNAME']	1✔
77	password = settings.DATA_SHARING[share_destination]['PASSWORD']	1✔
78	except KeyError as err:	×
79	raise ImproperlyConfigured(f'Check DATA_SHARING configuration for {share_destination}: Key {err} not found.')	×
80	auth = (username, password)	1✔
81	headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}	1✔
82
83	dataproducts_url = destination_tom_base_url + 'api/dataproducts/'	1✔
84	targets_url = destination_tom_base_url + 'api/targets/'	1✔
85	reduced_datums_url = destination_tom_base_url + 'api/reduceddatums/'	1✔
86	reduced_datums = ReducedDatum.objects.none()	1✔
87
88	if product_id:	1✔
89	product = DataProduct.objects.get(pk=product_id)	1✔
90	target = product.target	1✔
91	serialized_data = DataProductSerializer(product).data	1✔
92	destination_target_id, target_search_response = get_destination_target(target, targets_url, headers, auth)	1✔
93	if destination_target_id is None:	1✔
94	return {'message': 'ERROR: No matching target found.'}	1✔
95	elif isinstance(destination_target_id, list) and len(destination_target_id) > 1:	1✔
96	return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}	×
97	serialized_data['target'] = destination_target_id	1✔
98	# TODO: this should be updated when tom_dataproducts is updated to use django.core.storage
99	dataproduct_filename = os.path.join(settings.MEDIA_ROOT, product.data.name)	1✔
100	# Save DataProduct in Destination TOM
101	with open(dataproduct_filename, 'rb') as dataproduct_filep:	1✔
102	files = {'file': (product.data.name, dataproduct_filep, 'text/csv')}	1✔
103	headers = {'Media-Type': 'multipart/form-data'}	1✔
104	response = requests.post(dataproducts_url, data=serialized_data, files=files, headers=headers, auth=auth)	1✔
105	elif selected_data or target_id:	1✔
106	if selected_data:	1✔
107	reduced_datums = ReducedDatum.objects.filter(pk__in=selected_data)	1✔
108	targets = set(reduced_datum.target for reduced_datum in reduced_datums)	1✔
109	target_dict = {}	1✔
110	for target in targets:	1✔
111	# get destination Target
112	destination_target_id, target_search_response = get_destination_target(target,	1✔
113	targets_url,
114	headers,
115	auth)
116	if isinstance(destination_target_id, list) and len(destination_target_id) > 1:	1✔
117	return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}	×
118	target_dict[target.name] = destination_target_id	1✔
119	if all(value is None for value in target_dict.values()):	1✔
120	return {'message': 'ERROR: No matching targets found.'}	1✔
121	else:
122	target = Target.objects.get(pk=target_id)	1✔
123	reduced_datums = ReducedDatum.objects.filter(target=target)	1✔
124	destination_target_id, target_search_response = get_destination_target(target, targets_url, headers, auth)	1✔
125	if destination_target_id is None:	1✔
126	return {'message': 'ERROR: No matching target found.'}	1✔
127	elif isinstance(destination_target_id, list) and len(destination_target_id) > 1:	1✔
128	return {'message': 'ERROR: Multiple targets with matching name found in destination TOM.'}	×
129	target_dict = {target.name: destination_target_id}	1✔
130	response_codes = []	1✔
131	reduced_datums = check_for_share_safe_datums(share_destination, reduced_datums)	1✔
132	if not reduced_datums:	1✔
133	return {'message': 'ERROR: No valid data to share.'}	×
134	for datum in reduced_datums:	1✔
135	if target_dict[datum.target.name]:	1✔
136	serialized_data = ReducedDatumSerializer(datum).data	1✔
137	serialized_data['target'] = target_dict[datum.target.name]	1✔
138	serialized_data['data_product'] = ''	1✔
139	if not serialized_data['source_name']:	1✔
140	serialized_data['source_name'] = settings.TOM_NAME	1✔
141	serialized_data['source_location'] = "TOM-TOM Direct Sharing"	1✔
142	response = requests.post(reduced_datums_url, json=serialized_data, headers=headers, auth=auth)	1✔
143	response_codes.append(response.status_code)	1✔
144	failed_data_count = response_codes.count(500)	1✔
145	if failed_data_count < len(response_codes):	1✔
146	return {'message': f'{len(response_codes)-failed_data_count} of {len(response_codes)} '	1✔
147	'datums successfully saved.'}
148	else:
149	return {'message': 'ERROR: No valid data shared. These data may already exist in target TOM.'}	×
150	else:
151	return {'message': 'ERROR: No valid data to share.'}	×
152
153	return response	1✔
154
155
156	def get_destination_target(target, targets_url, headers, auth):	1✔
157	"""
158	Retrieve the target ID from a destination TOM
159	:param target: Target Model
160	:param targets_url: Destination API URL for TOM Target List
161	:param headers:
162	:param auth:
163	:return:
164	"""
165	target_names = ','.join(map(str, target.names))	1✔
166	target_response = requests.get(f'{targets_url}?name_fuzzy={target_names}', headers=headers, auth=auth)	1✔
167	target_response_json = target_response.json()	1✔
168	try:	1✔
169	if target_response_json['results']:	1✔
170	if len(target_response_json['results']) > 1:	1✔
171	return target_response_json['results'], target_response	1✔
172	destination_target_id = target_response_json['results'][0]['id']	1✔
173	return destination_target_id, target_response	1✔
174	else:
175	return None, target_response	1✔
176	except KeyError:	1✔
177	return None, target_response	1✔
178
179
180	def check_for_share_safe_datums(destination, reduced_datums, **kwargs):	1✔
181	"""
182	Custom sharing protocols used to determine when data is shared with a destination.
183	This example prevents sharing if a datum has already been published to the given Hermes topic.
184	:param destination: sharing destination string
185	:param reduced_datums: selected input datums
186	:return: queryset of reduced datums to be shared
187	"""
188	return reduced_datums	1✔
189	# if 'hermes' in destination:
190	# message_topic = kwargs.get('topic', None)
191	# # Remove data points previously shared to the given topic
192	# filtered_datums = reduced_datums.exclude(Q(message__exchange_status='published')
193	# & Q(message__topic=message_topic))
194	# else:
195	# filtered_datums = reduced_datums
196	# return filtered_datums
197
198
199	def check_for_save_safe_datums():	1✔
200	return	×
201
202
203	def get_sharing_destination_options():	1✔
204	"""
205	Build the Display options and headers for the dropdown form for choosing sharing topics.
206	Customize for a different selection experience.
207	:return: Tuple: Possible Destinations and their Display Names
208	"""
209	choices = []	1✔
210	try:	1✔
211	for destination, details in settings.DATA_SHARING.items():	1✔
212	new_destination = [details.get('DISPLAY_NAME', destination)]	1✔
213	if details.get('USER_TOPICS', None):	1✔
214	# If topics exist for a destination (Such as HERMES) give topics as sub-choices
215	# for non-selectable Destination
216	if destination == "hermes":	1✔
217	destination_topics = get_hermes_topics()	1✔
218	else:
219	destination_topics = details['USER_TOPICS']	×
220	topic_list = [(f'{destination}:{topic}', topic) for topic in destination_topics]	1✔
221	new_destination.append(tuple(topic_list))	1✔
222	else:
223	# Otherwise just use destination as option
224	new_destination.insert(0, destination)	1✔
225	choices.append(tuple(new_destination))	1✔
226	except AttributeError:	×
227	pass	×
228	return tuple(choices)	1✔
229
230
231	def sharing_feedback_handler(response, request):	1✔
232	try:	1✔
233	if 'message' in response.json():	1✔
234	publish_feedback = response.json()['message']	1✔
235	else:
236	publish_feedback = f"ERROR: {response.text}"	×
237	except AttributeError:	1✔
238	publish_feedback = response['message']	1✔
239	except ValueError:	×
240	publish_feedback = f"ERROR: Returned Response code {response.status_code}"	×
241	if "ERROR" in publish_feedback.upper():	1✔
242	messages.error(request, publish_feedback)	1✔
243	else:
244	messages.success(request, publish_feedback)	1✔
245	return	1✔

TOMToolkit / tom_base / 6488803620

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous