"""Views for defining, listing and moving diagnostic expressions.
The library belongs to the open project, stored in its database, because
exclusion rules make expressions part of how a project is processed.
Projects share expressions by exporting and importing them.
Validity does not depend on the project (see
:mod:`autowisp.diagnostics.diagnostic_types`); what does -- whether the
diagnostics an expression needs have been recorded -- is reported as
availability rather than as brokenness. "Meaningless" and "nothing
recorded yet" are not the same complaint and do not read as one here.
"""
import json
from io import StringIO
from django.contrib import messages
from django.http import Http404, HttpResponse
from django.shortcuts import redirect, render
from autowisp.database.interface import start_db_session
from autowisp.database.user_interface import count_cameras_lacking
from autowisp.diagnostics import expression_library
from autowisp.diagnostics.diagnostic_types import time_quantity
from autowisp.diagnostics.exclusion_rules import get_rule_reads
from autowisp.diagnostics.expressions import (
check_expression,
get_bare_aggregates,
get_expression_dependents,
get_expression_names,
get_logical_keywords,
get_quoted_channel_order,
order_expressions,
)
from autowisp.exceptions import PipelineError
from .expression_data import get_expressions
from .forms import DiagnosticExpressionForm
from .quantities import get_recorded_diagnostics
#: Marks an export file as ours and says which shape it is in. A file
#: without the key, or carrying a version this code does not know, is
#: refused rather than guessed at.
_format_key = "autowisp_diagnostic_expressions"
#: The only export shape there has been so far. It is meant to become the
#: configuration format a command-line run is pointed at, which is why it
#: is versioned and why a subset export pulls in what it depends on.
_format_version = 1
[docs]
def _references(name, expressions):
"""
Return the library entries one expression names directly.
Tolerates an unparseable expression by reporting no references: what is
wrong with it is :func:`check_expression`'s to say, and a page listing
expressions must not fail to render because one of them is broken.
Args:
name(str): The expression to look at.
expressions(dict): The library, ``{name: expression}``.
Returns:
set: The names of the expressions it references. Diagnostics
and functions are left out; they are not links to follow.
"""
try:
referenced = get_expression_names(expressions[name])
except SyntaxError:
return set()
return {other for other in referenced if other in expressions} - {name}
[docs]
def _reachable(names, expressions):
"""
Return *names* and every expression they reach, transitively.
Deliberately not :func:`order_expressions`, which refuses a library
with a cycle or an unresolvable name in it: exporting is one way a user
moves expressions somewhere they can be repaired, so it has to work on
a library that does not validate.
Args:
names(iterable): The selected expression names.
expressions(dict): The library, ``{name: expression}``.
Returns:
list: The closure, alphabetically.
"""
reached = set()
pending = [name for name in names if name in expressions]
while pending:
name = pending.pop()
if name in reached:
continue
reached.add(name)
pending.extend(_references(name, expressions))
return sorted(reached)
[docs]
def describe_expression(name, expressions, recorded):
"""
Return one row of the management table.
The two columns that can complain say different things, and the
distinction is the point of the page. *Problems* is whether the
expression means anything, which is the same answer in every project,
while *missing* is whether this one has recorded what it needs, which
is not. An expression naming a diagnostic this project never produced
is unavailable here and perfectly sound; only a typo is broken.
Args:
name(str): The expression to describe.
expressions(dict): The library, ``{name: expression}``.
recorded(set): The diagnostic names in use in the open project.
Returns:
dict: The fields ``diagnostic_expressions.html`` renders.
"""
problems = check_expression(name, expressions[name], expressions)
missing = None
if not problems:
_, needed = order_expressions([name], expressions)
# jd is known for every image of the canonical list, so it never
# counts against availability.
missing = sorted(needed - {time_quantity} - recorded)
return {
"name": name,
"expression": expressions[name],
"depends_on": sorted(_references(name, expressions)),
"problems": problems,
"missing": missing,
}
[docs]
def get_expression_warnings(names, library, db_session):
"""
Return what the given expressions were probably not meant to say.
None of it is refused: each is legitimate somewhere, merely unlikely
to be what was meant. Also for exclusion rules, which are expressions
too, added to the library under their parameter's name. What does not
parse is skipped rather than raised: saying what is wrong with it is
:func:`check_expression`'s or ``check_rule``'s to do.
Args:
names(iterable): The expressions to look at.
library(dict): The library they belong to, themselves included,
``{name: expression}``.
db_session: An active SQLAlchemy database session.
Returns:
list: The warnings, by expression.
"""
element_wise = {"and": "&", "or": "|", "not": "~"}
warnings = []
for name in sorted(names):
expression = library[name]
try:
aggregates = sorted(get_bare_aggregates(expression))
keywords = sorted(get_logical_keywords(expression))
# Read through the expressions it references too: one quoting a
# channel fails without it, and so does anything reading it.
lacking = sorted(
count_cameras_lacking(
get_quoted_channel_order(name, library), db_session
).items()
)
except (SyntaxError, PipelineError):
continue
warnings.extend(
f"{name} calls {aggregate}(), which goes NaN as soon as one image "
"of a series lacks a diagnostic. Did you mean "
f"nan{aggregate}()?"
for aggregate in aggregates
)
warnings.extend(
f"{name} uses {keyword}, which fails on arrays. Did you mean "
f"{element_wise[keyword]}?"
for keyword in keywords
)
warnings.extend(
f"{name} quotes channel {channel}, which {num_lacking} of "
f"{total} cameras do not have."
for channel, (num_lacking, total) in lacking
)
return warnings
[docs]
def _warn_about(request, names, library):
"""
Warn about what the given expressions were probably not meant to say.
Args:
request: The Django request to attach the warnings to.
names(iterable): The expressions just written.
library(dict): The library they were accepted into, themselves
included, ``{name: expression}``.
"""
with start_db_session() as db_session:
for warning in get_expression_warnings(names, library, db_session):
messages.warning(request, warning)
[docs]
def _render_list(request, form, edit_name=""):
"""
Render the management page around *form*, bound or blank.
The library comes from the form rather than being fetched again: the
form was built with the library this request is about, and on a failed
save the table must show what is stored rather than what was typed.
Args:
request: The Django request.
form(DiagnosticExpressionForm): The form to render above the
table, blank when adding and filled when editing.
edit_name(str): The name of the row being replaced, which the
template posts back so a rename stays an edit. Empty when
adding.
"""
expressions = form.expressions
with start_db_session() as db_session:
recorded = set(get_recorded_diagnostics(db_session))
# The description is the one stored column no rule is derived from,
# so it is merged in here rather than threaded through the library,
# which is a name-to-expression mapping everywhere else in the
# feature.
descriptions = expression_library.get_expression_descriptions(
db_session
)
# What to set an exclusion rule option to, to use one as a rule, so that
# nobody has to work out the subscripts.
rule_reads = get_rule_reads(expressions)
return render(
request,
"diagnostics/diagnostic_expressions.html",
{
"form": form,
"edit_name": edit_name,
"expression_rows": [
dict(
describe_expression(name, expressions, recorded),
description=descriptions.get(name, ""),
rule_read=rule_reads.get(name, ""),
)
for name in sorted(expressions)
],
},
)
[docs]
def list_expressions(request, name=None):
"""
Show the library, with a form for adding to it or editing one row.
Editing is a URL rather than a click that fills the form in place, so
that it needs no JavaScript, survives a refresh, and can be linked to.
``Http404`` for a name that is not there is the right answer to a stale
link, and Django's own -- the error middleware deliberately leaves it
alone.
Args:
request: The Django request.
name(str): The expression to open for editing, or ``None`` to
show the blank form.
"""
expressions = get_expressions()
initial = None
if name is not None:
with start_db_session() as db_session:
entries = expression_library.get_expression_entries(
[name], db_session
)
if not entries:
raise Http404(f"No diagnostic expression named {name}.")
initial = entries[0]
return _render_list(
request,
DiagnosticExpressionForm(
initial=initial, expressions=expressions, replacing=name
),
edit_name=name or "",
)
[docs]
def save_expression(request):
"""
Create or update one expression.
Editing is keyed by ``edit_name`` rather than by primary key, so that
the page is driven entirely by the names it displays and a rename is an
edit rather than a delete followed by a create. A rename carries the
expressions referencing the old name with it (see
:func:`~autowisp.diagnostics.expression_library.store_expression`), and
says which ones moved.
"""
assert request.method == "POST"
edit_name = request.POST.get("edit_name", "")
form = DiagnosticExpressionForm(
request.POST, expressions=get_expressions(), replacing=edit_name
)
if not form.is_valid():
# Re-rendered rather than redirected, so the complaints stay
# attached to the fields that caused them -- and still in edit
# mode, so correcting one does not silently create a second row.
return _render_list(request, form, edit_name=edit_name)
name = form.cleaned_data["name"]
with start_db_session() as db_session:
updated = expression_library.store_expression(
db_session,
name=name,
expression=form.cleaned_data["expression"],
description=form.cleaned_data["description"],
replacing=edit_name or None,
)
if updated:
messages.info(
request,
f"Renamed {edit_name} to {name}, and updated "
+ ", ".join(updated)
+ " to match.",
)
_warn_about(request, [name], get_expressions())
return redirect("diagnostics:list_expressions")
[docs]
def delete_expressions(request):
"""
Delete the checked expressions, unless something still needs them.
Dependents are judged against what will *remain*, so a whole chain may
be deleted together while the bottom of it may not be deleted alone.
"""
assert request.method == "POST"
selected = set(request.POST.getlist("expression_names"))
if not selected:
messages.info(request, "Nothing was ticked, so nothing was deleted.")
return redirect("diagnostics:list_expressions")
expressions = get_expressions()
blocked = {}
for name in selected:
dependents = get_expression_dependents(name, expressions) - selected
if dependents:
blocked[name] = sorted(dependents)
if blocked:
messages.error(
request,
"Nothing was deleted: "
+ "; ".join(
f"{name} is used by {', '.join(dependents)}"
for name, dependents in sorted(blocked.items())
)
+ ".",
)
else:
with start_db_session() as db_session:
expression_library.delete_expressions(selected, db_session)
return redirect("diagnostics:list_expressions")
[docs]
def export_expressions(request):
"""
Download the library, or a selection of it, as JSON.
A selection is extended with everything it depends on, since a file
naming an expression it does not carry cannot be imported anywhere
else -- nor read by a command-line run, which is what this format is
ultimately for.
A POST rather than a link, because the selection is the same set of
checkboxes the delete button reads: one form serves both, with
``formaction`` sending each button here or there. Nothing is lost by
it -- the URL of a download whose content depends on what is ticked is
not worth bookmarking.
"""
assert request.method == "POST"
expressions = get_expressions()
selected = request.POST.getlist("expression_names")
names = (
_reachable(selected, expressions) if selected else sorted(expressions)
)
with start_db_session() as db_session:
entries = expression_library.get_expression_entries(names, db_session)
with StringIO() as export_stream:
json.dump(
{
_format_key: _format_version,
"expressions": entries,
},
export_stream,
indent=4,
)
return HttpResponse(
export_stream.getvalue().encode("utf-8"),
headers={
"Content-Type": "application/json",
"Content-Disposition": (
'attachment; filename="diagnostic_expressions.json"'
),
},
)
[docs]
def _staged_expressions(entries):
"""
Return ``{name: fields}`` for the entries of an import file.
Args:
entries: Whatever the file's ``expressions`` key held, which is
not to be trusted to be a list of anything in particular.
Returns:
dict: ``{name: {"expression": …, "description": …}}``, ready to
be laid over the stored library.
Raises:
ValueError: If the entries are not objects carrying a name and
an expression.
"""
staged = {}
for entry in entries:
if not isinstance(entry, dict) or not {"name", "expression"} <= set(
entry
):
raise ValueError(
"every entry must be an object with a name and an expression"
)
staged[str(entry["name"])] = {
"expression": str(entry["expression"]),
"description": str(entry.get("description", "")),
}
return staged
[docs]
def _write_expressions(entries):
"""Store *entries*, returning how many were new and how many replaced."""
with start_db_session() as db_session:
return expression_library.write_expressions(entries, db_session)
[docs]
def import_expressions(request):
"""
Add expressions from a JSON file written by :func:`export_expressions`.
The whole file is staged over the stored library before anything is
checked, so that expressions referencing each other validate whatever
order they appear in. Entries are then written one at a time, and one
that does not validate is reported rather than aborting the rest.
A file naming an expression that already exists is the one thing this
cannot decide alone, so it asks: everything uncontested is written,
and the clashes go to :func:`confirm_import_expressions` with both
versions shown. Asking only when it happens is why there is no
setting to get wrong beforehand -- one that is read once in a hundred
imports would be forgotten in the other ninety-nine.
"""
assert request.method == "POST"
try:
document = json.load(request.FILES["expressions-import"])
if document.get(_format_key) != _format_version:
raise ValueError(
f"expected {_format_key!r} to be {_format_version}, which is "
"what this version of AutoWISP writes"
)
staged = _staged_expressions(document.get("expressions", []))
except (ValueError, UnicodeDecodeError, AttributeError) as error:
messages.error(request, f"Not a diagnostic expression file: {error}.")
return redirect("diagnostics:list_expressions")
stored = get_expressions()
# What each entry is checked against: the file laid over the library,
# so an intra-file reference resolves whether or not its target has
# been written yet.
library = dict(
stored,
**{name: entry["expression"] for name, entry in staged.items()},
)
fresh, clashing, refused = {}, {}, []
for name, entry in staged.items():
problems = check_expression(name, entry["expression"], library)
if problems:
refused.append(f"{name} ({' '.join(problems)})")
elif name in stored:
clashing[name] = entry
else:
fresh[name] = entry
added, _ = _write_expressions(fresh)
# Only what actually happened: "imported 0" beside a question about
# what to do with the rest says the same thing twice, the second time
# wrongly.
if added:
messages.info(request, f"Imported {added} expression(s).")
elif not staged:
messages.info(request, "That file listed no expressions.")
if refused:
messages.error(request, "Refused " + "; ".join(sorted(refused)) + ".")
_warn_about(request, fresh, library)
if not clashing:
return redirect("diagnostics:list_expressions")
return render(
request,
"diagnostics/confirm_import.html",
{
"clashes": [
{
"name": name,
"stored": stored[name],
"incoming": entry["expression"],
"description": entry["description"],
}
for name, entry in sorted(clashing.items())
],
# Carried through the answer rather than kept server-side: the
# page is the whole of the pending state, so an abandoned
# import leaves nothing behind to expire or collide with a
# second tab's.
"clashes_json": json.dumps(clashing, sort_keys=True),
},
)
[docs]
def confirm_import_expressions(request):
"""
Replace the stored expressions an import was asked about.
Reached only from :func:`import_expressions`' question, and only by
the answer that changes something -- keeping the stored versions is a
link back to the list, there being nothing to do.
"""
assert request.method == "POST"
try:
clashing = _staged_expressions(
[
{"name": name, **entry}
for name, entry in json.loads(
request.POST.get("clashes", "{}")
).items()
]
)
except (ValueError, AttributeError, TypeError) as error:
messages.error(request, f"Could not read what to replace: {error}.")
return redirect("diagnostics:list_expressions")
# Checked again rather than trusted: what comes back is a form field,
# and the library may have moved on while the question was open.
library = dict(
get_expressions(),
**{name: entry["expression"] for name, entry in clashing.items()},
)
writable, refused = {}, []
for name, entry in sorted(clashing.items()):
problems = check_expression(name, entry["expression"], library)
if problems:
refused.append(f"{name} ({' '.join(problems)})")
else:
writable[name] = entry
_, updated = _write_expressions(writable)
if updated:
messages.info(request, f"Replaced {updated} expression(s).")
if refused:
messages.error(request, "Refused " + "; ".join(refused) + ".")
_warn_about(request, writable, library)
return redirect("diagnostics:list_expressions")