mrqart.email_latest_flip

Daily header-compliance summary email based on db.sqlite.

Refactor note: - main() is now a thin orchestration layer. - Core logic is split into testable helpers:

  • get_report_date()

  • fetch_acquisitions()

  • select_eligible_rows()

  • _evaluate_row() <– per-row logic, extracted for testability

  • evaluate_rows()

  • format_seq_result() <– renders a single SeqSummary entry to lines

  • build_email()

  • send_all()

Behavior should be identical to the previous version.

Functions

build_email(*, date_label, marquee_cols, ...)

Render subject + body from aggregated results.

build_jsonl_entries(*, date_label, ...[, ...])

Build a list of JSONL entries for the web dashboard.

compact_error_keys(errors, prefer_cols[, ...])

errors is dict keyed by col name.

dbg(msg)

evaluate_rows(eligible_rows, *, sql, tc, ...)

Core evaluation engine — calls _evaluate_row for each eligible row and aggregates results into seq_summary, missing_templates, and totals.

fetch_acquisitions(sql, yday_str)

first_seen_date_for_seq(sql, project, seqname)

first_seen_from_template_by_count(sql, ...)

format_expected_got(col, exp, got)

format_seq_result(summary, *, marquee_cols)

Render a single SeqSummary to a list of lines for inclusion in the email.

format_series_003(series)

get_physicist_for_project(sql, project)

Look up the assigned physicist for a project by matching the suffix after '^' in the acq_param Project name against the project table.

get_report_date(env[, now])

Determine reporting date:

is_interesting_sequence_with_blacklist(...)

Rules:

load_email_entries(toml_path)

load_reporting_config(toml_path)

Load reporting settings from config/reporting.toml.

log_line(msg)

main(*[, dry_run])

normalize_for_compare(val)

parse_ta_seconds(val)

rebuild_templates(sql)

select_eligible_rows(acq_rows, settings)

@param acq_rows sql query results - acquisitions to check @param settings configuration from load_reporting_config passed to is_interesting_sequence_with_blacklist Apply: - SeriesNumber <= 200 - reporting filter (interesting/deny/blacklist) Returns: - eligible rows - study_counts_today: total eligible rows per project - seq_counts_today: total eligible rows per (project, subid, seqname) - study_subids_today: set of unique SubIDs seen per project (for session count)

send_all(email_entries, subject, body, *[, ...])

Send the same subject/body to all recipients.

send_via_local_mail(subject, body, recipient)

series_int(series)

series_is_posthoc(series)

study_has_any_templates(sql, project)

yyyymmdd_to_iso(s)

Classes

ReportDate(report_date, yday_str, date_label)

RowResult(key[, missing_template, ...])

Result of evaluating a single acquisition row against its template.

SeqSummary(key[, total, nonconforming, ...])

Aggregated conformance data for a single (project, subid, seqname) key.

Totals([total_seen_today, total_checked, ...])

class mrqart.email_latest_flip.ReportDate(report_date: 'datetime', yday_str: 'str', date_label: 'str')[source]
Parameters:

  • report_date (datetime)

  • yday_str (str)

  • date_label (str)

class mrqart.email_latest_flip.RowResult(key, missing_template=False, has_any_diff=False, has_marquee_mismatch=False, errors_dict=<factory>, first_seen=None, study_has_templates=False)[source]

Result of evaluating a single acquisition row against its template.

Parameters:

  • key (Tuple[str, str, str])

  • missing_template (bool)

  • has_any_diff (bool)

  • has_marquee_mismatch (bool)

  • errors_dict (Dict[str, Dict[str, str]])

  • first_seen (str | None)

  • study_has_templates (bool)

class mrqart.email_latest_flip.SeqSummary(key, total=0, nonconforming=0, anydiff=0, examples=<factory>, mismatch_counts=<factory>)[source]

Aggregated conformance data for a single (project, subid, seqname) key. This is the canonical data structure for a single sequence check — populated by evaluate_rows and rendered by format_seq_result.

Parameters:

  • key (Tuple[str, str, str])

  • total (int)

  • nonconforming (int)

  • anydiff (int)

  • examples (List[str])

  • mismatch_counts (Dict[Tuple[str, str, str], int])

class mrqart.email_latest_flip.Totals(total_seen_today: 'int' = 0, total_checked: 'int' = 0, total_nonconforming: 'int' = 0, total_anydiff: 'int' = 0, total_missing_templates: 'int' = 0, mia_actionable: 'int' = 0)[source]
Parameters:

  • total_seen_today (int)

  • total_checked (int)

  • total_nonconforming (int)

  • total_anydiff (int)

  • total_missing_templates (int)

  • mia_actionable (int)

mrqart.email_latest_flip.build_email(*, date_label, marquee_cols, total_seen_today, seq_summary, missing_templates, totals, study_subids_today, physicist_by_project)[source]

Render subject + body from aggregated results. Nonconforming sequences are grouped by project; each acquisition is a flat bullet with diffs below. Project header shows session count today.

Parameters:

  • date_label (str)

  • marquee_cols (List[str])

  • total_seen_today (int)

  • seq_summary (Dict[Tuple[str, str, str], SeqSummary])

  • missing_templates (Dict[Tuple[str, str, str], Dict[str, Any]])

  • totals (Totals)

  • study_subids_today (Mapping[str, set])

  • physicist_by_project (Mapping[str, str | None])

Return type:

Tuple[str, str]

mrqart.email_latest_flip.build_jsonl_entries(*, date_label, seq_summary, missing_templates, marquee_cols, physicist_by_project={})[source]

Build a list of JSONL entries for the web dashboard. One entry per nonconforming sequence or missing template. Conforming sequences are skipped to keep the log lean.

Parameters:

  • date_label (str)

  • seq_summary (Dict[Tuple[str, str, str], SeqSummary])

  • missing_templates (Dict[Tuple[str, str, str], Dict[str, Any]])

  • marquee_cols (List[str])

  • physicist_by_project (Mapping[str, str | None])

Return type:

List[Dict[str, Any]]

mrqart.email_latest_flip.compact_error_keys(errors, prefer_cols, max_items=6)[source]

errors is dict keyed by col name. prefer marquee cols + TA first.

Parameters:

  • errors (Dict[str, Any])

  • prefer_cols (List[str])

  • max_items (int)

Return type:

str

mrqart.email_latest_flip.evaluate_rows(eligible_rows, *, sql, tc, marquee_cols, study_counts_today, seq_counts_today, study_has_templates_fn=<function study_has_any_templates>, first_seen_from_templates_fn=<function first_seen_from_template_by_count>, first_seen_from_acq_fn=<function first_seen_date_for_seq>)[source]

Core evaluation engine — calls _evaluate_row for each eligible row and aggregates results into seq_summary, missing_templates, and totals.

SequenceType is always added to marquee_set here regardless of what is in reporting.toml, so SequenceType changes are always caught as nonconforming.

Parameters:

  • eligible_rows (Iterable[Row])

  • sql (Connection)

  • tc (TemplateChecker)

  • marquee_cols (List[str])

  • study_counts_today (Mapping[str, int])

  • seq_counts_today (Mapping[Tuple[str, str, str], int])

  • study_has_templates_fn (Callable[[Connection, str], bool])

  • first_seen_from_templates_fn (Callable[[Connection, str, str], str | None])

  • first_seen_from_acq_fn (Callable[[Connection, str, str], str | None])

Return type:

Tuple[Dict[Tuple[str, str, str], SeqSummary], Dict[Tuple[str, str, str], Dict[str, Any]], Totals]

mrqart.email_latest_flip.format_seq_result(summary, *, marquee_cols)[source]

Render a single SeqSummary to a list of lines for inclusion in the email. Each nonconforming acquisition is a flat bullet: subid/seqname.acqnum with diffs indented below. No fraction or ex: line — the bullet IS the example. Bullets with no displayable diffs (e.g. only SequenceType) are silently skipped. Can be called independently for testing or iteration.

Parameters:
Return type:

List[str]

mrqart.email_latest_flip.get_physicist_for_project(sql, project)[source]

Look up the assigned physicist for a project by matching the suffix after ‘^’ in the acq_param Project name against the project table. Case-insensitive. Returns None if no match or no physicist assigned.

Parameters:

  • sql (Connection)

  • project (str)

Return type:

str | None

mrqart.email_latest_flip.get_report_date(env, now=None)[source]
Determine reporting date:

  • MRQART_DATE override supports YYYYMMDD or YYYY-MM-DD

  • default: yesterday relative to now

Parameters:

  • env (Mapping[str, str])

  • now (datetime | None)

Return type:

ReportDate

mrqart.email_latest_flip.is_interesting_sequence_with_blacklist(seqname, seqtype, settings)[source]
Rules:

  1. If SequenceName contains any interesting substring, include.

  2. Else if disable_blacklist is True, include.

  3. Else if SequenceType prefix is in blacklist, exclude.

  4. Else include.

Parameters:

  • seqname (str)

  • seqtype (str | None)

  • settings (Dict[str, Any])

Return type:

bool

mrqart.email_latest_flip.load_reporting_config(toml_path)[source]

Load reporting settings from config/reporting.toml.

Expected structure:

[filter] interesting_substrings = […] deny_substrings = […] blacklist_seqtype_prefixes = […] disable_blacklist = bool blacklist_study_regex = [“^Body”, “^7TB”]

[compare] marquee_cols = […]

Note: SequenceType is always added to marquee_set at evaluation time (in evaluate_rows) regardless of what is listed here. BWP and PixelResol are not forced but should be added to reporting.toml if you want them to get their own detail lines in the email.

Parameters:

toml_path (Path)

Return type:

Dict[str, Any]

mrqart.email_latest_flip.select_eligible_rows(acq_rows, settings)[source]

@param acq_rows sql query results - acquisitions to check @param settings configuration from load_reporting_config

passed to is_interesting_sequence_with_blacklist

Apply:

  • SeriesNumber <= 200

  • reporting filter (interesting/deny/blacklist)

Returns:

  • eligible rows

  • study_counts_today: total eligible rows per project

  • seq_counts_today: total eligible rows per (project, subid, seqname)

  • study_subids_today: set of unique SubIDs seen per project (for session count)

Parameters:

  • acq_rows (Iterable[Row])

  • settings (Dict[str, Any])

Return type:

Tuple[List[Row], Dict[str, int], Dict[Tuple[str, str, str], int], Dict[str, set]]

mrqart.email_latest_flip.send_all(email_entries, subject, body, *, send_fn=<function send_via_local_mail>)[source]

Send the same subject/body to all recipients. Returns True if any failures occurred.

Parameters:

  • email_entries (Iterable[Mapping[str, str]])

  • subject (str)

  • body (str)

  • send_fn (Callable[[str, str, str], bool])

Return type:

bool