"""Characterization tests for the BUI diagnostics view modules.
These pin the behaviours that must survive merging
``image_diagnostics_views`` and ``diag_vs_diag_views`` into a single
x-versus-y path: how the ``pixel_quantiles`` pseudo-name expands into
series, and how the time-series figure offsets and groups its series. They
are written against the *current* code, so they must pass before the merge
starts.
Uses a throwaway project database, following ``test_error_render``.
"""
import tempfile
import unittest
from datetime import datetime
from unittest import mock
import matplotlib
# The backend has to be selected before anything imports pyplot, which the
# view module under test does at import time.
matplotlib.use("Agg")
# pylint: disable=wrong-import-position
import numpy
from sqlalchemy import select
from autowisp.database.interface import set_project_home, start_db_session
# pylint: disable=no-name-in-module
from autowisp.database.data_model import (
DiagnosticType,
Image,
ImageDiagnostics,
ImageType,
ObservingSession,
)
# pylint: enable=no-name-in-module
# Imported after set_project_home is available; these only touch the DB
# through start_db_session, so no Django configuration is needed.
from autowisp.diagnostics.expression_series import (
SeriesKey,
get_canonical_images,
)
from autowisp.exceptions import PipelineError
from autowisp.browser_interface.diagnostics.image_diagnostics_views import (
create_diagnostics_figure,
get_series_data,
group_series_by_x_overlap,
)
from autowisp.browser_interface.diagnostics.series_table import (
get_available_diagnostics,
get_available_expressions,
get_available_series,
get_recorded_diagnostics,
get_series_key,
make_row_id,
split_row_id,
)
# pylint: enable=wrong-import-position
#: JD of the first image of the first night.
_first_jd = 2460000.5
#: Nights are one day apart, so their JD ranges cannot overlap.
_night_separation = 1.0
#: The quantile diagnostics, which expand to one series each.
_quantile_names = ("pixel_q99", "pixel_q999")
#: Every diagnostic the fixture records. All are created explicitly: the
#: lazy database initialization behind ``set_project_home`` creates the
#: schema but seeds no ``diagnostic_type`` rows, and depending on that would
#: couple these tests to project-creation behaviour they are not about.
_diagnostic_names = ("bg_center",) + _quantile_names
#: Frames of each type per night. Only the second night is mixed, which is
#: what lets these tests tell a per-type series from one that lumps a whole
#: session together; leaving the first night single-type keeps the plain
#: one-series-per-night cases readable.
_frames_per_night = ({"object": 3}, {"object": 3, "flat": 2})
#: ``bg_center`` of the first frame of each type. Far enough apart that a
#: median over one type cannot be confused with a median over the mixture.
_first_bg_center = {"object": 100.0, "flat": 500.0}
[docs]
class DiagnosticsViewTestCase(unittest.TestCase):
"""Base creating one throwaway project database holding two nights."""
[docs]
@classmethod
def setUpClass(cls):
# Closed in tearDownClass rather than by a context manager, which a
# fixture spanning every test of the class cannot use.
# pylint: disable=consider-using-with
cls._tmp = tempfile.TemporaryDirectory()
# pylint: enable=consider-using-with
set_project_home(cls._tmp.name)
cls._fill_database()
[docs]
@classmethod
def tearDownClass(cls):
cls._tmp.cleanup()
#: ``{(night, image_type): [image_id, ...]}``, in JD order, so a test can
#: say which images a series is supposed to be built from.
images_of = {}
[docs]
def rows_for(self, x_diagnostic, y_diagnostic, expressions=None):
"""Return ``{(session, type, quantile): row}`` for an axis pair.
Keyed by the group rather than by the whole series, a row having
no channels until something binds them.
"""
with start_db_session() as db_session:
context = get_available_series(
x_diagnostic, y_diagnostic, expressions or {}, db_session
)
return {
split_row_id(row["id"]): row for row in context["diagnostics_list"]
}
[docs]
@staticmethod
def bind(row, *channels):
"""Return the row as the client posts it with its dropdowns set."""
return {**row, "channels": list(channels)}
[docs]
@classmethod
def _fill_database(cls):
"""Create two observing sessions, the second holding two image types.
Every frame records ``bg_center`` in channel ``R``; only object
frames record the quantiles, which is the ordinary case of a
diagnostic that is not defined for every type. Provenance foreign
keys are left dangling, as SQLite does not enforce them and the
diagnostics queries only ever join back to ``observing_session`` and
``image_type``.
"""
# False positive: the declarative models are callable.
# pylint: disable=not-callable
with start_db_session() as db_session:
for name in _diagnostic_names:
db_session.add(
DiagnosticType(
name=name, description=f"Test diagnostic {name}"
)
)
for name in ("object", "flat"):
db_session.add(ImageType(name=name, description=f"{name}s"))
db_session.flush()
diagnostic_ids = dict(
db_session.execute(
select(DiagnosticType.name, DiagnosticType.id).where(
DiagnosticType.name.in_(_diagnostic_names)
)
).all()
)
image_type_ids = dict(
db_session.execute(select(ImageType.name, ImageType.id)).all()
)
for night, frame_counts in enumerate(_frames_per_night):
session = ObservingSession(
observer_id=1,
camera_id=1,
telescope_id=1,
mount_id=1,
observatory_id=1,
target_id=1,
label=f"night_{night}",
start_time_utc=datetime(2023, 3, 1 + night, 20, 0, 0),
end_time_utc=datetime(2023, 3, 1 + night, 23, 0, 0),
)
db_session.add(session)
db_session.flush()
jd = _first_jd + night * _night_separation
for image_type, count in frame_counts.items():
cls.images_of[night, image_type] = []
for index in range(count):
image = Image(
raw_fname=(
f"/data/raw/n{night}_{image_type}_{index}.fits"
),
image_type_id=image_type_ids[image_type],
observing_session_id=session.id,
jd=jd,
)
db_session.add(image)
db_session.flush()
cls.images_of[night, image_type].append(image.id)
jd += 0.05
values = {
"bg_center": _first_bg_center[image_type] + index
}
if image_type == "object":
values["pixel_q99"] = 200.0 + index
values["pixel_q999"] = 300.0 + index
for name, value in values.items():
db_session.add(
ImageDiagnostics(
image_id=image.id,
channel="R",
diagnostic_id=diagnostic_ids[name],
value=value,
)
)
# pylint: enable=not-callable
[docs]
class TestRowId(unittest.TestCase):
"""The row id, a round trip through the client and back.
It becomes an HTML element id, four more element ids are built from
it, and it keys the ``datasets`` object the client posts back -- so it
has to survive all of that as an opaque string. What it deliberately
does *not* carry is the channels: those are chosen in the row, and an
id that changed as they were would take every element id with it.
"""
[docs]
def round_trip(self, session_id, image_type, quantile_name, ordinal=0):
"""Return the group recovered from the id built from these."""
return split_row_id(
make_row_id(session_id, image_type, quantile_name, ordinal)
)
[docs]
def test_plain_row(self):
"""No quantile: the field is empty rather than missing."""
self.assertEqual(
self.round_trip(7, "object", None), (7, "object", None)
)
[docs]
def test_quantile_row(self):
"""The quantile survives despite containing underscores.
This is what the previous encoding could not do without guessing
which underscores separated fields and which belonged to the name.
"""
self.assertEqual(
self.round_trip(7, "object", "pixel_q999"),
(7, "object", "pixel_q999"),
)
[docs]
def test_underscores_anywhere_are_harmless(self):
"""Neither the image type nor the quantile has to avoid them."""
self.assertEqual(
self.round_trip(7, "twilight_flat", "pixel_q999"),
(7, "twilight_flat", "pixel_q999"),
)
[docs]
def test_an_ambiguous_field_is_refused(self):
"""Failing loudly beats an id that silently pairs wrong data."""
with self.assertRaises(ValueError):
make_row_id(7, "we|rd", None, 0)
[docs]
def test_the_image_type_is_part_of_the_identity(self):
"""Two types in one session must not collide on one id."""
self.assertNotEqual(
make_row_id(7, "object", None, 0),
make_row_id(7, "flat", None, 0),
)
[docs]
def test_siblings_differ_by_their_ordinal(self):
"""A group holds a row per binding, and they share everything else."""
self.assertNotEqual(
make_row_id(7, "object", None, 0),
make_row_id(7, "object", None, 1),
)
[docs]
def test_the_key_takes_its_channels_from_the_client(self):
"""The half of a series that the table edits, and only that half."""
self.assertEqual(
get_series_key(
{
"id": make_row_id(7, "object", "pixel_q999", 2),
"channels": ["R", "B"],
}
),
SeriesKey(7, "object", ("R", "B"), "pixel_q999"),
)
[docs]
class TestAvailableQuantities(DiagnosticsViewTestCase):
"""What the two axis selectors offer, given what this project holds.
Availability rather than validity: every stored expression is valid in
every project, so the only question a selector can usefully ask is
whether the diagnostics an expression reaches have actually been
recorded here.
"""
[docs]
def _recorded(self):
"""Return the raw in-use diagnostic names."""
with start_db_session() as db_session:
return get_recorded_diagnostics(db_session)
[docs]
def test_recorded_names_are_raw(self):
"""The individual quantiles, not the family that replaces them.
This is the set an expression is judged against, and one may
reference a concrete ``pixel_q999``, so the collapse must happen
after rather than before.
"""
self.assertEqual(sorted(self._recorded()), sorted(_diagnostic_names))
[docs]
def test_selector_offers_the_family_not_its_members(self):
"""One entry expanding to a series per quantile, and ``jd`` first."""
available = get_available_diagnostics(self._recorded(), {})
self.assertEqual(available[0], "jd")
self.assertIn("pixel_quantiles", available)
self.assertIn("bg_center", available)
for name in _quantile_names:
self.assertNotIn(name, available)
[docs]
def test_expressions_join_the_same_flat_list(self):
"""Not a second list beside it.
An axis reads one name, and a recorded diagnostic is an expression
of itself as far as everything downstream is concerned -- which is
the same flat name space that stops an expression taking a
diagnostic's name.
"""
available = get_available_diagnostics(
self._recorded(), {"rel_bg": "bg_center * 2"}
)
self.assertIn("rel_bg", available)
self.assertIn("bg_center", available)
[docs]
def test_a_dependency_makes_its_dependents_unavailable(self):
"""Availability follows the whole subtree, not the direct names."""
library = {
"rel": "astrom_residual / diagonal_fov",
"scaled": "rel * 2",
}
self.assertEqual(
get_available_expressions(library, self._recorded()), []
)
[docs]
def test_a_concrete_quantile_is_judged_against_the_raw_names(self):
"""Which is why the family collapse happens after this check.
Against the collapsed list ``pixel_q999`` would look unrecorded and
a perfectly drawable expression would go missing from the selector.
"""
library = {"contrast": "pixel_q999 / pixel_q99"}
self.assertEqual(
get_available_expressions(library, self._recorded()), ["contrast"]
)
[docs]
def test_a_jd_only_expression_is_always_available(self):
"""``jd`` exists for every image of the canonical list."""
library = {"night_time": "jd - nanmin(jd)"}
self.assertEqual(
get_available_expressions(library, self._recorded()), ["night_time"]
)
[docs]
def test_a_broken_expression_is_hidden_rather_than_raised(self):
"""A stored cycle must not stop the plot page rendering.
Saying what is wrong with it belongs to the management page; here
the only sane answer is not to offer it.
"""
library = {"a": "b + 1", "b": "a + 1", "rel_bg": "bg_center * 2"}
self.assertEqual(
get_available_expressions(library, self._recorded()), ["rel_bg"]
)
[docs]
def test_an_unknown_name_is_hidden_too(self):
"""A typo no version of AutoWISP defines, rather than a cycle."""
library = {"typo": "bg_centre * 2"}
self.assertEqual(
get_available_expressions(library, self._recorded()), []
)
[docs]
class TestQuantileSeriesExpansion(DiagnosticsViewTestCase):
"""``pixel_quantiles`` expands to one series per ``pixel_q*``.
On either axis, since the expansion happens where the series are built
rather than per axis.
"""
[docs]
def _series_ids(self, x_diagnostic, y_diagnostic):
"""Return the series ids offered for the given axis pair."""
with start_db_session() as db_session:
context = get_available_series(
x_diagnostic, y_diagnostic, {}, db_session
)
return [series["id"] for series in context["diagnostics_list"]]
[docs]
def test_quantiles_on_x_axis(self):
"""One row per quantile per night, the name carried in the id."""
# Every night holds object frames, and only those record the
# quantiles, so each quantile is offered once per night.
nights = len(_frames_per_night)
series_ids = self._series_ids("pixel_quantiles", "bg_center")
quantiles = [split_row_id(series_id)[2] for series_id in series_ids]
self.assertEqual(len(series_ids), nights * len(_quantile_names))
for name in _quantile_names:
self.assertEqual(
quantiles.count(name),
nights,
f"expected one {name} row per night, got {series_ids!r}",
)
[docs]
def test_quantiles_on_y_axis(self):
"""Reversing the axes yields the same expansion."""
self.assertEqual(
sorted(self._series_ids("bg_center", "pixel_quantiles")),
sorted(self._series_ids("pixel_quantiles", "bg_center")),
)
[docs]
def test_quantile_column_present(self):
"""A quantile pairing gains the extra ``Quantile`` table column."""
with start_db_session() as db_session:
context = get_available_series(
"pixel_quantiles", "bg_center", {}, db_session
)
self.assertIn("Quantile", context["diagnostics_fields"])
[docs]
class TestSharedTimeOffset(DiagnosticsViewTestCase):
"""The x-offset is one value for the whole figure, not per series."""
[docs]
def _plotted_x_values(self):
"""Return the x arrays that reach the per-series plotting call.
``plot_image_diagnostic_series`` is mocked out, which both captures
the offset values and avoids ``reverse()`` needing Django settings.
"""
with start_db_session() as db_session:
context = get_available_series("jd", "bg_center", {}, db_session)
# Bound as the table's dropdowns would be: a row names no data
# until it is, and is skipped rather than drawn.
series_list = [
self.bind(row, "R") for row in context["diagnostics_list"]
]
# One per (night, image type): night 0 object, night 1 object,
# night 1 flat.
self.assertEqual(len(series_list), 3)
target = (
"autowisp.browser_interface.diagnostics"
".image_diagnostics_views.plot_image_diagnostic_series"
)
with mock.patch(target) as plot_series:
create_diagnostics_figure(
series_list,
x_diagnostic="jd",
y_diagnostic="bg_center",
expressions={},
db_session=db_session,
# Nothing is drawn once plotting is mocked, so asking
# for a legend only produces a warning.
figure_config={"show_legend": False},
)
return [call.args[1] for call in plot_series.call_args_list]
[docs]
def _series_starts(self):
"""Return where each plotted series begins on the shared x axis."""
return sorted(float(min(values)) for values in self._plotted_x_values())
[docs]
def test_only_the_earliest_series_starts_at_zero(self):
"""One offset for the figure, so exactly one series lands on 0."""
starts = self._series_starts()
self.assertAlmostEqual(starts[0], 0.0, places=6)
for start in starts[1:]:
self.assertGreater(start, 0.0)
[docs]
def test_offset_is_not_per_series(self):
"""Guard the exact regression the merge could introduce.
Zeroing each series on its own would start every one of them at 0,
collapsing the day between the two nights. The second night's
series keep that day, wherever in the night each one begins.
"""
for start in self._series_starts()[1:]:
self.assertGreaterEqual(
start,
_night_separation,
msg="a second-night series was zeroed on its own -- the "
"offset became per-series instead of shared",
)
[docs]
class TestImageTypeSplit(DiagnosticsViewTestCase):
"""A session holding several image types yields a series per type."""
[docs]
def test_each_type_gets_its_own_series(self):
"""The mixed night offers object and flat separately."""
types = {
image_type
for session_id, image_type, _ in self.rows_for("jd", "bg_center")
if session_id == 2
}
self.assertEqual(types, {"object", "flat"})
[docs]
def test_a_type_without_the_diagnostic_is_absent(self):
"""Only object frames record the quantiles, so only they appear."""
types = {
image_type
for _, image_type, _ in self.rows_for(
"pixel_quantiles", "bg_center"
)
}
self.assertEqual(types, {"object"})
[docs]
def test_the_type_is_shown_in_the_table(self):
"""Otherwise two rows of the mixed night would look identical."""
with start_db_session() as db_session:
context = get_available_series("jd", "bg_center", {}, db_session)
self.assertIn("Type", context["diagnostics_fields"])
[docs]
def test_canonical_list_holds_only_its_own_type(self):
"""The alignment the whole design rests on is per type.
Every array is padded onto this list, so if it mixed types then so
would every quantity built against it.
"""
with start_db_session() as db_session:
for image_type in ("object", "flat"):
image_ids, _ = get_canonical_images(
SeriesKey(2, image_type, ("R",)), db_session
)
self.assertEqual(
image_ids.tolist(), self.images_of[1, image_type]
)
[docs]
def test_values_are_not_taken_across_types(self):
"""The point of the split: an aggregate sees one population.
A series covering the whole night would hand ``nanmedian`` all five
frames and return the object median, since the objects outnumber the
flats -- silently, and wrongly.
"""
series = self.bind(
self.rows_for("jd", "bg_center")[2, "flat", None], "R"
)
with start_db_session() as db_session:
_, y_values, image_ids = get_series_data(
series, "jd", "bg_center", {}, db_session
)
flat_values = [
_first_bg_center["flat"] + index
for index in range(_frames_per_night[1]["flat"])
]
self.assertEqual(y_values.tolist(), flat_values)
self.assertEqual(image_ids.tolist(), self.images_of[1, "flat"])
self.assertAlmostEqual(
float(numpy.nanmedian(y_values)),
numpy.median(flat_values),
places=6,
)
[docs]
class TestExpressionAxis(DiagnosticsViewTestCase):
"""An expression selected for an axis, as a diagnostic would be.
The library is passed in rather than stored, which is the arrangement
that lets these run against a project database alone: what the view does
with an expression does not depend on where it was kept.
"""
#: Referenced by every test here; ``bg_center`` is recorded for both
#: image types, so the availability answer is interesting.
library = {
"rel_bg": "bg_center[1] - nanmedian(bg_center[1])",
"scaled_bg": "rel_bg[1] * 10",
"q_ratio": "pixel_q999[1] / pixel_q99[1]",
}
[docs]
def test_offered_wherever_its_diagnostics_are(self):
"""Availability follows what the expression reaches, not its name.
Nothing records a diagnostic called ``rel_bg``; the series it can be
drawn for are those recording the ``bg_center`` it is built from.
"""
self.assertEqual(
sorted(self.rows_for("jd", "rel_bg", self.library)),
sorted(self.rows_for("jd", "bg_center", self.library)),
)
[docs]
def test_a_composed_expression_reaches_through(self):
"""``scaled_bg`` needs what ``rel_bg`` needs, transitively."""
self.assertEqual(
sorted(self.rows_for("jd", "scaled_bg", self.library)),
sorted(self.rows_for("jd", "bg_center", self.library)),
)
[docs]
def test_the_values_are_the_expression_evaluated(self):
"""End to end: an expression axis produces its own numbers."""
series = self.bind(
self.rows_for("jd", "rel_bg", self.library)[2, "object", None], "R"
)
with start_db_session() as db_session:
_, y_values, _ = get_series_data(
series, "jd", "rel_bg", self.library, db_session
)
# bg_center is 100, 101, 102 for these frames.
self.assertEqual(y_values.tolist(), [-1.0, 0.0, 1.0])
[docs]
def test_an_expression_against_a_diagnostic(self):
"""Both axes at once, one of each kind, sharing a query."""
series = self.bind(
self.rows_for("bg_center", "rel_bg", self.library)[
2, "object", None
],
"R",
"R",
)
with start_db_session() as db_session:
x_values, y_values, _ = get_series_data(
series, "bg_center", "rel_bg", self.library, db_session
)
self.assertEqual(x_values.tolist(), [100.0, 101.0, 102.0])
self.assertEqual(y_values.tolist(), [-1.0, 0.0, 1.0])
[docs]
def test_an_unknown_name_is_refused(self):
"""Neither a diagnostic nor an expression, so nothing to plot."""
with self.assertRaises(PipelineError):
self.rows_for("jd", "no_such_thing", self.library)
[docs]
class TestSeriesGrouping(unittest.TestCase):
"""``group_series_by_x_overlap`` splits only non-overlapping ranges."""
[docs]
@staticmethod
def _entry(jd_values):
"""Build the tuple shape the grouping helper consumes."""
return ({}, numpy.asarray(jd_values), None, None)
[docs]
def test_disjoint_ranges_split(self):
"""Two nights a day apart occupy separate subplots."""
groups = group_series_by_x_overlap(
[
self._entry([_first_jd, _first_jd + 0.1]),
self._entry(
[
_first_jd + _night_separation,
_first_jd + _night_separation + 0.1,
]
),
]
)
self.assertEqual(len(groups), 2)
[docs]
def test_overlapping_ranges_merge(self):
"""Overlapping ranges share one subplot.
This is the case a non-time x axis reduces to once the grouping is
generalized from JD to arbitrary x, so it must keep holding.
"""
groups = group_series_by_x_overlap(
[
self._entry([0.0, 10.0]),
self._entry([5.0, 15.0]),
self._entry([12.0, 20.0]),
]
)
self.assertEqual(len(groups), 1)
self.assertEqual(len(groups[0]), 3)
if __name__ == "__main__":
unittest.main()