# -*- coding: utf-8 -*-
"""
Instrumentation Journal Entry submodule.
"""
import os
import shutil
import typing
from werkzeug.wrappers.response import Response
from flask import Blueprint, flash, g, redirect, render_template, url_for, current_app, abort
# import ResearchNotes.instruments as instruments
from ResearchNotes.auth import login_required
import ResearchNotes.database_transactions as dbt
from ResearchNotes.database import (
Instrument,
TemplateInstrumentationJournalEntry,
db,
InstrumentationJournalEntry,
EntryType,
)
from ResearchNotes.form import (
InstrumentationJournalEntryCreate,
UseTemplate,
) # , MeasurementOrder
from ResearchNotes.files import uploaddir_path, make_file_list
from ResearchNotes.url_security import token_decode
bp = Blueprint("entry", __name__, url_prefix="/entry")
[docs]@bp.route("/overview_jentires", methods=("GET", "POST"))
@login_required
def show_all_entire() -> str | Response:
"""
Renders all Instrumentation Journal Entries of the group.
Returns
-------
str | Response
Render template to show all entries.
"""
return render_template("entry/allentries.html")
[docs]@bp.route("/<int:iid>/htmx_fill_template", methods=["PUT"])
@login_required
def htmx_fill_template(iid: int) -> str:
"""
Htmx function to fill the template into the create form.
Parameters
----------
iid : int
ID of the instrument the entry belongs to.
Returns
-------
str
HTML to be replaced in the frontend page.
"""
choose_template_form = UseTemplate()
current_app.logger.debug(f" htmx : Fill template for instrument {iid}")
# As we do not fill the option into the choose form (templates), we cannot validate the input here
# if choose_template_form.template.data is not None:
form = create_create_form(choose_template_form.template.data, iid)
current_app.logger.debug(f" htmx : Form filled with template {choose_template_form.template.data}")
return render_template("entry/inner_create_form.html", create_entry_form=form)
[docs]@bp.route("/<list:ids>/create", methods=("GET", "POST"))
@login_required
def create(ids: typing.List) -> str | Response:
"""
Create a new instrumentation journal entry.
Parameters
----------
ids : List of the sample and template id
Returns
-------
str or Response
"""
iid, tid = ids
# choose template form
choose_template_form = UseTemplate()
choose_template_form.template.choices = [
(t.id, t.tname)
for t in db.session.execute(
db.select(TemplateInstrumentationJournalEntry).order_by("tname")
).scalars()
if (t.group_id == g.user.group_id) or (t.creator_id == g.user.id)
]
choose_template_form.template.choices.insert(0, (0, "None"))
create_entry_form = create_create_form(tid, iid)
if create_entry_form.validate_on_submit():
new_entry_info = {
"identifier": create_entry_form.identifier.data,
"description": create_entry_form.description.data,
"etype": create_entry_form.etype.data,
"creator_id": g.user.id,
"group_id": g.user.group_id,
"instrument_id": iid,
}
eid = dbt.create_instrumentation_journal_entry(db, new_entry_info)
flash(f"P/P/M { new_entry_info['identifier']} created", "alert-info")
return redirect(url_for("entry.entry_view", eid=eid))
return render_template(
"entry/create.html",
create_entry_form=create_entry_form,
choose_template_form=choose_template_form,
entry=None,
iid=iid,
)
[docs]@bp.route("/<int:eid>/update", methods=("GET", "POST"))
@login_required
def update(eid: int) -> str | Response:
"""
Update an instrumentation journal entry.
Parameters
----------
eid : int
Instrumentation journal entry ID
Returns
-------
str | Flask.Response
Render update form or redirect to measurement view.
"""
entry = get_entry(eid)
update_entry_form = InstrumentationJournalEntryCreate(obj=entry)
update_entry_form.etype.choices = [
(etype.id, etype.name)
for etype in db.get_or_404(Instrument, entry.instrument_id).etypes
if etype.id != entry.etype_id
]
# add current etype as first choice
if entry.etype is None:
update_entry_form.etype.choices.insert(0, (0, "None"))
else:
update_entry_form.etype.choices.insert(0, (0, "None"))
update_entry_form.etype.choices.insert(0, (entry.etype.id, entry.etype.name))
# update_entry_form.etype.default = [entry.etype_id]
if update_entry_form.validate_on_submit():
# print(update_entry_form.etype.data)
dbt.update_instrumentation_journal_entry(
db,
entry,
{
"identifier": update_entry_form.identifier.data,
"description": update_entry_form.description.data,
"etype": update_entry_form.etype.data,
},
)
flash("Instrumentation journal entry updated", "alert-info")
return redirect(url_for("entry.entry_view", eid=eid))
return render_template(
"entry/create.html",
create_entry_form=update_entry_form,
entry=entry,
iid=entry.instrument.id,
)
[docs]@bp.route("/<int:eid>/journal_entry", methods=("GET", "POST"))
@login_required
def entry_view(eid: int) -> str | Response:
"""
Display an instrumentation journal entry and all associated files.
Parameters
----------
eid : int
Instrumentation journal entry ID
Returns
-------
str or Response
Renders template of the measurement view
"""
entry = get_entry(eid)
files = make_file_list(
uploaddir_path(
[
"e",
entry.id,
entry.instrument.id,
entry.instrument.identifier,
]
)
)
return render_template("entry/entry.html", entry=entry, files=files)
def _delete(token: str, page: str) -> Response:
"""
Delete an instrumentation journal entry.
Parameters
----------
token : str
Signed string that encodes the id of the journal entry to delete
page: str
Instrument page to redirect to, e.g. "log_view", "instrument_view"
Returns
-------
Flask.Response
Redirect to specified page
"""
eid = token_decode(token, current_app.config["SEC_SESSION_KEY"], g.salt)
entry = get_entry(eid)
iid = entry.instrument.id
delete_entry_data(entry)
flash(f"Deleted report {entry.identifier}", "alert-info")
return redirect(url_for(f"instruments.{page}", iid=iid))
[docs]@bp.route("/<string:token>/delete_iv")
@login_required
def delete_iv(token: str):
"""
Delete an instrumentation journal entry and redirect to "instrument_view".
Parameters
----------
token : str
Signed string that encodes the id of the journal entry to delete
Returns
-------
Flask.Response
Redirect to "instrument_view"
"""
return _delete(token, page="instrument_view")
[docs]@bp.route("/<string:token>/delete_lv")
@login_required
def delete_lv(token: str):
"""
Delete an instrumentation journal entry and redirect to "log_view".
Parameters
----------
token : str
Signed string that encodes the id of the journal entry to delete
Returns
-------
Flask.Response
Redirect to "log_view"
"""
return _delete(token, page="log_view")
[docs]def delete_entry_data(entry: InstrumentationJournalEntry) -> None:
"""
Delete all data associated with an instrumentation journal entry.
This includes all files and the database object itself. This is done
"silently" i.e. without flash() so it can also be used by the function
that deletes instruments to delete all data of the entries associated
with that instrument.
Parameters:
-----------
entry: entry whose data is deleted
"""
# remove all associated files
path = uploaddir_path(
[
"e",
entry.id,
entry.instrument.id,
entry.instrument.identifier,
]
)
if os.path.exists(path):
try:
shutil.rmtree(path)
except shutil.Error as error:
abort(500, description=f"Failed to delete {path}. Exception {error}")
# remove database object
with dbt.Transaction(db) as db_session:
db_session.delete(entry)
[docs]def get_entry(eid: int, check_permission: bool = True) -> InstrumentationJournalEntry:
"""
Retrieve an instrumentation journal entry by ID, optionally verifying permission.
Parameters
----------
eid : int
database id of the measurement.
check_permission : bool, optional
If true, we check for permission. The default is True.
Returns
-------
measurement : ResearchNotes.Measurement
Loaded Measurement database entry.
"""
entry = db.get_or_404(
InstrumentationJournalEntry,
eid,
description=f": get_entry : {g.user} tried to load instrumentation"
+ f" journal entry {eid} which does not exist",
)
if check_permission:
if entry.instrument.active:
if not (
entry.instrument.id in [i.id for i in g.user.group_member.owned_instruments]
or entry.instrument.id in [i.id for i in g.user.shared_instruments]
):
abort(
403,
description=f"instrument.get_instrument: Authorization failure. {g.user} "
+ f"tried to load instrument {entry.instrument.id}",
)
elif g.user.role_member.name == "Supervisor":
if not (entry.instrument.id in [i.id for i in g.user.group_member.owned_instruments]):
abort(
403,
description=f"instrument.get_instrument: Authorization failure. {g.user} "
+ f"tried to load instrument {entry.instrument.id}",
)
#
return entry