Basic System (#31)

Co-authored-by: Tim Lorsbach <tim@lorsba.ch>
Reviewed-on: enviPath/enviPy#31
This commit is contained in:
2025-07-23 06:47:07 +12:00
parent 49e02ed97d
commit df896878f1
75 changed files with 3821 additions and 1429 deletions

View File

@ -1,5 +1,6 @@
import json
import logging
from functools import wraps
from typing import List, Dict, Any
from django.conf import settings as s
@ -7,29 +8,71 @@ from django.contrib.auth import get_user_model
from django.http import JsonResponse, HttpResponse, HttpResponseNotAllowed, HttpResponseBadRequest
from django.shortcuts import render, redirect
from django.views.decorators.csrf import csrf_exempt
from django.contrib.auth.decorators import login_required
from utilities.chem import FormatConverter, IndigoUtils
from .logic import GroupManager, PackageManager, UserManager, SettingManager
from .logic import GroupManager, PackageManager, UserManager, SettingManager, SearchManager
from .models import Package, GroupPackagePermission, Group, CompoundStructure, Compound, Reaction, Rule, Pathway, Node, \
EPModel, EnviFormer, MLRelativeReasoning, RuleBaseRelativeReasoning, Scenario, SimpleAmbitRule, APIToken, \
UserPackagePermission, Permission, License, User
UserPackagePermission, Permission, License, User, Edge
logger = logging.getLogger(__name__)
def log_post_params(request):
for k, v in request.POST.items():
logger.debug(f"{k}\t{v}")
if s.DEBUG:
for k, v in request.POST.items():
logger.debug(f"{k}\t{v}")
def catch_exceptions(view_func):
@wraps(view_func)
def _wrapped_view(request, *args, **kwargs):
try:
return view_func(request, *args, **kwargs)
except Exception as e:
# Optionally return JSON or plain HttpResponse
if request.headers.get('Accept') == 'application/json':
return JsonResponse(
{'error': 'Internal server error. Please try again later.'},
status=500
)
else:
return render(request, 'errors/error.html', get_base_context(request))
return _wrapped_view
def editable(request, user):
url = request.build_absolute_uri(request.path)
if PackageManager.is_package_url(url):
_package = PackageManager.get_package_lp(request.build_absolute_uri())
return PackageManager.writable(user, _package)
elif GroupManager.is_group_url(url):
_group = GroupManager.get_group_lp(request.build_absolute_uri())
return GroupManager.writable(user, _group)
elif UserManager.is_user_url(url):
_user = UserManager.get_user_lp(request.build_absolute_uri())
return UserManager.writable(user, _user)
elif url in [s.SERVER_URL, f"{s.SERVER_URL}/", f"{s.SERVER_URL}/package", f"{s.SERVER_URL}/user",
f"{s.SERVER_URL}/group", f"{s.SERVER_URL}/search"]:
return True
else:
print(f"Unknown url: {url}")
return False
def get_base_context(request) -> Dict[str, Any]:
current_user = _anonymous_or_real(request)
can_edit = editable(request, current_user)
ctx = {
'title': 'enviPath',
'meta': {
'version': '0.0.1',
'server_url': s.SERVER_URL,
'user': current_user,
'can_edit': can_edit,
'readable_packages': PackageManager.get_all_readable_packages(current_user, include_reviewed=True),
'writeable_packages': PackageManager.get_all_writeable_packages(current_user),
'available_groups': GroupManager.get_groups(current_user),
@ -65,8 +108,9 @@ def breadcrumbs(first_level_object=None, second_level_namespace=None, second_lev
return bread
# @catch_exceptions
def index(request):
current_user = _anonymous_or_real(request)
context = get_base_context(request)
context['title'] = 'enviPath - Home'
context['meta']['current_package'] = context['meta']['user'].default_package
@ -77,6 +121,16 @@ def index(request):
return render(request, 'index/index.html', context)
# def login(request):
# current_user = _anonymous_or_real(request)
# if request.method == 'GET':
# context = get_base_context(request)
# context['title'] = 'enviPath'
# return render(request, 'login.html', context)
# else:
# return HttpResponseBadRequest()
#
# @login_required(login_url='/login')
def packages(request):
current_user = _anonymous_or_real(request)
@ -86,9 +140,10 @@ def packages(request):
context['object_type'] = 'package'
context['meta']['current_package'] = context['meta']['user'].default_package
context['meta']['can_edit'] = True
reviewed_package_qs = Package.objects.filter(reviewed=True)
unreviewed_package_qs = PackageManager.get_all_readable_packages(current_user)
reviewed_package_qs = Package.objects.filter(reviewed=True).order_by('created')
unreviewed_package_qs = PackageManager.get_all_readable_packages(current_user).order_by('name')
context['reviewed_objects'] = reviewed_package_qs
context['unreviewed_objects'] = unreviewed_package_qs
@ -103,7 +158,6 @@ def packages(request):
else:
package_name = request.POST.get('package-name')
package_description = request.POST.get('package-description', s.DEFAULT_VALUES['description'])
# group = GroupManager.get_group_by_url(request.user, request.POST.get('package-group'))
created_package = PackageManager.create_package(current_user, package_name, package_description)
@ -342,7 +396,23 @@ def models(request):
def search(request):
current_user = _anonymous_or_real(request)
if request.method == 'GET':
package_urls = request.GET.getlist('packages')
searchterm = request.GET.get('search')
mode = request.GET.get('mode')
# add HTTP_ACCEPT check to differentiate between index and ajax call
if 'application/json' in request.META.get('HTTP_ACCEPT') and all([searchterm, mode]):
if package_urls:
packages = [PackageManager.get_package_by_url(current_user, p) for p in package_urls]
else:
packages = PackageManager.get_reviewed_packages()
search_result = SearchManager.search(packages, searchterm, mode)
return JsonResponse(search_result, safe=False)
context = get_base_context(request)
context['title'] = 'enviPath - Search'
@ -352,13 +422,21 @@ def search(request):
{'Home': s.SERVER_URL},
{'Search': s.SERVER_URL + '/search'},
]
# TODO perm
reviewed_package_qs = Package.objects.filter(reviewed=True)
unreviewed_package_qs = Package.objects.filter(reviewed=False)
reviewed_package_qs = PackageManager.get_reviewed_packages()
unreviewed_package_qs = PackageManager.get_all_readable_packages(current_user)
context['reviewed_objects'] = reviewed_package_qs
context['unreviewed_objects'] = unreviewed_package_qs
if all([searchterm, mode]):
if package_urls:
packages = [PackageManager.get_package_by_url(current_user, p) for p in package_urls]
else:
packages = PackageManager.get_reviewed_packages()
context['search_result'] = SearchManager.search(packages, searchterm, mode)
return render(request, 'search.html', context)
@ -487,7 +565,7 @@ def package_model(request, package_uuid, model_uuid):
return render(request, 'objects/model.html', context)
if request.method == 'POST':
elif request.method == 'POST':
if hidden := request.POST.get('hidden', None):
if hidden == 'delete-model':
current_model.delete()
@ -496,21 +574,9 @@ def package_model(request, package_uuid, model_uuid):
return HttpResponseBadRequest()
else:
return HttpResponseBadRequest()
#
# new_compound_name = request.POST.get('compound-name')
# new_compound_description = request.POST.get('compound-description')
#
# if new_compound_name:
# current_compound.name = new_compound_name
#
# if new_compound_description:
# current_compound.description = new_compound_description
#
# if any([new_compound_name, new_compound_description]):
# current_compound.save()
# return redirect(current_compound.url)
# else:
# return HttpResponseBadRequest()
else:
return HttpResponseBadRequest()
def package(request, package_uuid):
@ -803,8 +869,7 @@ def package_rules(request, package_uuid):
elif request.method == 'POST':
for k, v in request.POST.items():
print(k, v)
log_post_params(request)
# Generic params
rule_name = request.POST.get('rule-name')
@ -817,8 +882,8 @@ def package_rules(request, package_uuid):
# Obtain parameters as required by rule type
if rule_type == 'SimpleAmbitRule':
params['smirks'] = request.POST.get('rule-smirks')
params['reactant_smarts'] = request.POST.get('rule-reactant-smarts')
params['product_smarts'] = request.POST.get('rule-product-smarts')
params['reactant_filter_smarts'] = request.POST.get('rule-reactant-smarts')
params['product_filter_smarts'] = request.POST.get('rule-product-smarts')
elif rule_type == 'SimpleRDKitRule':
params['reaction_smarts'] = request.POST.get('rule-reaction-smarts')
elif rule_type == 'ParallelRule':
@ -828,7 +893,7 @@ def package_rules(request, package_uuid):
else:
return HttpResponseBadRequest()
r = Rule.create(current_package, rule_type, name=rule_name, description=rule_description, **params)
r = Rule.create(rule_type=rule_type, package=current_package, name=rule_name, description=rule_description, **params)
return redirect(r.url)
else:
@ -854,7 +919,7 @@ def package_rule(request, package_uuid, rule_uuid):
else: # isinstance(current_rule, ParallelRule) or isinstance(current_rule, SequentialRule):
return render(request, 'objects/composite_rule.html', context)
if request.method == 'POST':
elif request.method == 'POST':
if hidden := request.POST.get('hidden', None):
if hidden == 'delete-rule':
current_rule.delete()
@ -862,8 +927,23 @@ def package_rule(request, package_uuid, rule_uuid):
else:
return HttpResponseBadRequest()
# TODO update!
rule_name = request.POST.get('rule-name', '').strip()
rule_description = request.POST.get('rule-description', '').strip()
if rule_name:
current_rule.name = rule_name
if rule_description:
current_rule.description = rule_description
if any([rule_name, rule_description]):
current_rule.save()
return redirect(current_rule.url)
else:
return HttpResponseBadRequest()
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/reaction
def package_reactions(request, package_uuid):
@ -912,6 +992,8 @@ def package_reactions(request, package_uuid):
return redirect(r.url)
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/reaction/<id>
def package_reaction(request, package_uuid, reaction_uuid):
@ -931,7 +1013,7 @@ def package_reaction(request, package_uuid, reaction_uuid):
return render(request, 'objects/reaction.html', context)
if request.method == 'POST':
elif request.method == 'POST':
if hidden := request.POST.get('hidden', None):
if hidden == 'delete-reaction':
current_reaction.delete()
@ -954,6 +1036,8 @@ def package_reaction(request, package_uuid, reaction_uuid):
else:
return HttpResponseBadRequest()
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/pathway
def package_pathways(request, package_uuid):
@ -989,7 +1073,7 @@ def package_pathways(request, package_uuid):
return render(request, 'collections/objects_list.html', context)
if request.method == 'POST':
elif request.method == 'POST':
log_post_params(request)
@ -1002,15 +1086,34 @@ def package_pathways(request, package_uuid):
return HttpResponseBadRequest()
stand_smiles = FormatConverter.standardize(smiles)
pw = Pathway.create(current_package, name, description, stand_smiles)
if pw_mode != 'build':
if pw_mode not in ['predict', 'build', 'incremental']:
return HttpResponseBadRequest()
pw = Pathway.create(current_package, name, description, stand_smiles)
# set mode
pw.kv.update({'mode': pw_mode})
pw.save()
if pw_mode == 'predict' or pw_mode == 'incremental':
# unlimited pred (will be handled by setting)
limit = -1
# For incremental predict first level and return
if pw_mode == 'incremental':
limit = 1
pred_setting = current_user.prediction_settings()
pw.setting = pred_setting
pw.save()
from .tasks import predict
predict.delay(pw.pk, pred_setting.pk)
predict.delay(pw.pk, pred_setting.pk, limit=limit)
return redirect(pw.url)
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/pathway/<id>
def package_pathway(request, package_uuid, pathway_uuid):
@ -1043,7 +1146,7 @@ def package_pathway(request, package_uuid, pathway_uuid):
return render(request, 'objects/pathway.html', context)
# return render(request, 'pathway_playground2.html', context)
if request.method == 'POST':
elif request.method == 'POST':
if hidden := request.POST.get('hidden', None):
if hidden == 'delete-pathway':
current_pathway.delete()
@ -1051,30 +1154,92 @@ def package_pathway(request, package_uuid, pathway_uuid):
else:
return HttpResponseBadRequest()
pathway_name = request.POST.get('pathway-name')
pathway_description = request.POST.get('pathway-description')
#
#
#
# def package_relative_reasonings(request, package_id):
# if request.method == 'GET':
# pass
#
#
# def package_relative_reasoning(request, package_id, relative_reasoning_id):
# current_user = _anonymous_or_real(request)
#
# if request.method == 'GET':
# pass
# elif request.method == 'POST':
# pass
#
# #
# #
# # # https://envipath.org/package/<id>/pathway/<id>/node
# # def package_pathway_nodes(request, package_id, pathway_id):
# # pass
# #
# #
if any([pathway_name, pathway_description]):
if pathway_name is not None and pathway_name.strip() != '':
pathway_name = pathway_name.strip()
current_pathway.name = pathway_name
if pathway_description is not None and pathway_description.strip() != '':
pathway_description = pathway_description.strip()
current_pathway.description = pathway_description
current_pathway.save()
return redirect(current_pathway.url)
node_url = request.POST.get('node')
if node_url:
n = current_pathway.get_node(node_url)
from .tasks import predict
# Dont delay?
predict(current_pathway.pk, current_pathway.setting.pk, node_pk=n.pk)
return JsonResponse({'success': current_pathway.url})
return HttpResponseBadRequest()
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/pathway/<id>/node
def package_pathway_nodes(request, package_uuid, pathway_uuid):
current_user = _anonymous_or_real(request)
current_package = PackageManager.get_package_by_id(current_user, package_uuid)
current_pathway = Pathway.objects.get(package=current_package, uuid=pathway_uuid)
if request.method == 'GET':
context = get_base_context(request)
context['title'] = f'enviPath - {current_package.name} - {current_pathway.name} - Nodes'
context['meta']['current_package'] = current_package
context['object_type'] = 'node'
context['breadcrumbs'] = [
{'Home': s.SERVER_URL},
{'Package': s.SERVER_URL + '/package'},
{current_package.name: current_package.url},
{'Pathway': current_package.url + '/pathway'},
{current_pathway.name: current_pathway.url},
{'Node': current_pathway.url + '/node'},
]
reviewed_node_qs = Node.objects.none()
unreviewed_node_qs = Node.objects.none()
if current_package.reviewed:
reviewed_node_qs = Node.objects.filter(pathway=current_pathway).order_by('name')
else:
unreviewed_node_qs = Node.objects.filter(pathway=current_pathway).order_by('name')
if request.GET.get('all'):
return JsonResponse({
"objects": [
{"name": pw.name, "url": pw.url, "reviewed": current_package.reviewed}
for pw in (reviewed_node_qs if current_package.reviewed else unreviewed_node_qs)
]
})
context['reviewed_objects'] = reviewed_node_qs
context['unreviewed_objects'] = unreviewed_node_qs
return render(request, 'collections/objects_list.html', context)
elif request.method == 'POST':
node_name = request.POST.get('node-name')
node_description = request.POST.get('node-description')
node_smiles = request.POST.get('node-smiles')
current_pathway.add_node(node_smiles, name=node_name, description=node_description)
return redirect(current_pathway.url)
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/pathway/<id>/node/<id>
@ -1091,19 +1256,129 @@ def package_pathway_node(request, package_uuid, pathway_uuid, node_uuid):
svg_data = current_node.as_svg
return HttpResponse(svg_data, content_type="image/svg+xml")
context = get_base_context(request)
context['title'] = f'enviPath - {current_package.name} - {current_pathway.name}'
context['meta']['current_package'] = current_package
context['object_type'] = 'pathway'
context['breadcrumbs'] = [
{'Home': s.SERVER_URL},
{'Package': s.SERVER_URL + '/package'},
{current_package.name: current_package.url},
{'Pathway': current_package.url + '/pathway'},
{current_pathway.name: current_pathway.url},
{'Node': current_pathway.url + '/node'},
{current_node.name: current_node.url},
]
context['node'] = current_node
return render(request, 'objects/node.html', context)
elif request.method == 'POST':
if s.DEBUG:
for k, v in request.POST.items():
print(k, v)
if hidden := request.POST.get('hidden', None):
if hidden == 'delete-node':
current_node.delete()
return redirect(current_pathway.url)
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/pathway/<id>/edge
def package_pathway_edges(request, package_uuid, pathway_uuid):
current_user = _anonymous_or_real(request)
current_package = PackageManager.get_package_by_id(current_user, package_uuid)
current_pathway = Pathway.objects.get(package=current_package, uuid=pathway_uuid)
if request.method == 'GET':
context = get_base_context(request)
context['title'] = f'enviPath - {current_package.name} - {current_pathway.name} - Edges'
context['meta']['current_package'] = current_package
context['object_type'] = 'edge'
context['breadcrumbs'] = [
{'Home': s.SERVER_URL},
{'Package': s.SERVER_URL + '/package'},
{current_package.name: current_package.url},
{'Pathway': current_package.url + '/pathway'},
{current_pathway.name: current_pathway.url},
{'Edge': current_pathway.url + '/edge'},
]
reviewed_edge_qs = Edge.objects.none()
unreviewed_edge_qs = Edge.objects.none()
if current_package.reviewed:
reviewed_edge_qs = Edge.objects.filter(pathway=current_pathway).order_by('name')
else:
unreviewed_edge_qs = Edge.objects.filter(pathway=current_pathway).order_by('name')
if request.GET.get('all'):
return JsonResponse({
"objects": [
{"name": pw.name, "url": pw.url, "reviewed": current_package.reviewed}
for pw in (reviewed_edge_qs if current_package.reviewed else unreviewed_edge_qs)
]
})
context['reviewed_objects'] = reviewed_edge_qs
context['unreviewed_objects'] = unreviewed_edge_qs
return render(request, 'collections/objects_list.html', context)
elif request.method == 'POST':
edge_name = request.POST.get('edge-name')
edge_description = request.POST.get('edge-description')
edge_substrates = request.POST.getlist('edge-substrates')
edge_products = request.POST.getlist('edge-products')
substrate_nodes = [current_pathway.get_node(url) for url in edge_substrates]
product_nodes = [current_pathway.get_node(url) for url in edge_products]
# TODO in the future consider Rules here?
current_pathway.add_edge(substrate_nodes, product_nodes, name=edge_name, description=edge_description)
return redirect(current_pathway.url)
else:
return HttpResponseBadRequest()
# https://envipath.org/package/<id>/pathway/<id>/edge/<id>
def package_pathway_edge(request, package_uuid, pathway_uuid, edge_uuid):
current_user = _anonymous_or_real(request)
current_package = PackageManager.get_package_by_id(current_user, package_uuid)
current_pathway = Pathway.objects.get(package=current_package, uuid=pathway_uuid)
current_edge = Edge.objects.get(pathway=current_pathway, uuid=edge_uuid)
if request.method == 'GET':
is_image_request = request.GET.get('image')
if is_image_request:
if is_image_request == 'svg':
svg_data = current_edge.as_svg
return HttpResponse(svg_data, content_type="image/svg+xml")
elif request.method == 'POST':
if s.DEBUG:
for k, v in request.POST.items():
print(k, v)
if hidden := request.POST.get('hidden', None):
if hidden == 'delete-edge':
current_edge.delete()
return redirect(current_pathway.url)
else:
return HttpResponseBadRequest()
# #
# #
# # # https://envipath.org/package/<id>/pathway/<id>/edge
# # def package_pathway_edges(request, package_id, pathway_id):
# # pass
# #
# #
# # # https://envipath.org/package/<id>/pathway/<id>/edge/<id>
# # def package_pathway_edge(request, package_id, pathway_id, edge_id):
# # pass
# #
# #
# https://envipath.org/package/<id>/scenario
def package_scenarios(request, package_uuid):
current_user = _anonymous_or_real(request)
@ -1158,11 +1433,6 @@ def package_scenario(request, package_uuid, scenario_uuid):
return render(request, 'objects/scenario.html', context)
### END UNTESTED
##############
# User/Group #
##############
@ -1201,7 +1471,7 @@ def users(request):
return render(request, 'errors/user_account_inactive.html', status=403)
email = temp_user.email
except get_user_model().DoesNotExists:
except get_user_model().DoesNotExist:
return HttpResponseBadRequest()
user = authenticate(username=email, password=password)
@ -1289,6 +1559,18 @@ def user(request, user_uuid):
logout(request)
return redirect(s.SERVER_URL)
default_package = request.POST.get('default-package')
default_group = request.POST.get('default-group')
default_prediction_setting = request.POST.get('default-prediction-setting')
if any([default_package, default_group, default_prediction_setting]):
current_user.default_package = PackageManager.get_package_by_url(current_user, default_package)
current_user.default_group = GroupManager.get_group_by_url(current_user, default_group)
current_user.default_setting = SettingManager.get_setting_by_url(current_user, default_prediction_setting)
current_user.save()
return redirect(current_user.url)
prediction_model_pk = request.POST.get('model')
prediction_threshold = request.POST.get('threshold')
prediction_max_nodes = request.POST.get('max_nodes')
@ -1509,3 +1791,9 @@ def layout(request):
def depict(request):
if smiles := request.GET.get('smiles'):
return HttpResponse(IndigoUtils.mol_to_svg(smiles), content_type='image/svg+xml')
elif smirks := request.GET.get('smirks'):
query_smirks = request.GET.get('is_query_smirks', False) == 'true'
return HttpResponse(IndigoUtils.smirks_to_svg(smirks, query_smirks), content_type='image/svg+xml')
else:
return HttpResponseBadRequest()