cadrumo.domain.deadlines.engine module

Registry-backed deadline computation engine.

Takes an TaxpayerProfile and a year and produces a deterministic, typed Schedule. Filing windows and applicability conditions are read from validated calculation registry data supplied by ValidatedRegistryAuthority. Each window is described by a ModeloRevision paired with its deadline window definitions.

classify_obligation_status(closes_on, today, due_soon_days)[source]

Map a window close date to an ObligationStatus.

This is the single owner of the OVERDUE / DUE_TODAY / DUE_SOON / UPCOMING boundary rules. The deadline engine applies it to every registry-scheduled obligation, and the overview calendar applies it to a locally-created work unit after its own filing-pointer gate has run — that gate is the only intentional difference between the two callers, and it decides whether to call this function at all, never how the boundaries fall.

Parameters:

  • closes_on (date) – The day the AEAT filing window closes.

  • today (date) – The reference date the engine evaluates against.

  • due_soon_days (int) – Window (in days) before closes_on for which the obligation is flagged DUE_SOON.

Return type:

ObligationStatus

Returns:

The ObligationStatus for the window.

indexed_deadline_windows(operation, year)[source]

Project one filing year’s deadline windows from a pinned operation.

Reused for a generation already projected for this year; see _DEADLINE_WINDOW_INDEX.

Return type:

tuple[tuple[str, ModeloRevision, DeadlineWindowDefinition], ...]

Parameters:
class DeadlineEngine(*, due_soon_days=14, authority=None)[source]

Bases: object

Engine that computes typed filing schedules from registry data.

Variables:

due_soon_days – Window before cadrumo.domain.deadlines.models.ModeloDeadline.closes_on that flags cadrumo.domain.deadlines.models.ObligationStatus.DUE_SOON (default 14).

Parameters:
compute(profile, year, *, today=None)[source]

Compute the full Schedule for profile x year.

Pure function: no I/O, no input mutation. Identical (profile, year, today) always yields an equal schedule (modulo Schedule.generated_at).

Parameters:
Return type:

Schedule

Returns:

The Schedule containing every obligation that applies to profile for year.

Raises:

NoDeadlineWindowsError – If no validated registry deadline windows are registered for year — the benign data gap callers degrade around.

explain(profile, modelo, *, year=None)[source]

Return registry-backed deadline applicability text for modelo.

Parameters:

  • profile (TaxpayerProfile) – The TaxpayerProfile to evaluate conditions against.

  • modelo (str) – The AEAT modelo identifier to look up.

  • year (int | None) – Optional fiscal year; defaults to the current year.

Return type:

str

applies_to(profile, modelo, *, year=None)[source]

Return whether registry deadline conditions match for modelo.

Parameters:

  • profile (TaxpayerProfile) – The TaxpayerProfile to evaluate conditions against.

  • modelo (str) – The AEAT modelo identifier to check.

  • year (int | None) – Optional fiscal year; defaults to the current year.

Return type:

bool

deadline_windows(year)[source]

Return validated registry deadline windows for year.

This read-only facade lets application projections inspect the same registry surface used by compute() without reaching into engine implementation details.

Return type:

tuple[tuple[str, ModeloRevision, DeadlineWindowDefinition], ...]

Parameters:

year (int)

schedule_applies(profile, revision, window)[source]

Return whether a validated filing schedule applies to profile.

Parameters:
Return type:

bool

evaluate_conditions(profile, conditions, *, mode)[source]

Evaluate one registry applicability-condition tuple.

Parameters:
Return type:

str | None

next_deadline(schedule, today=None)[source]

Return the next obligation in schedule that has not yet closed.

Pure function. Returns None if every obligation in the schedule is already overdue (or the schedule is empty).

Parameters:
Return type:

ModeloDeadline | None

Returns:

The earliest non-overdue ModeloDeadline, or None if no such obligation exists.

class ScheduleProducer(*args, **kwargs)[source]

Bases: Protocol

Structural surface over DeadlineEngine.compute.

compute_obligation_schedule() is typed against this Protocol rather than the concrete DeadlineEngine so the workflow engine — which injects a protocol-typed deadline engine — and the state projection — which uses a concrete DeadlineEngine — can both feed the same single-producer function.

compute(profile, year, *, today=None)[source]

Return a Schedule for profile in year.

Parameters:
Return type:

Schedule

compute_obligation_schedule(engine, profile, *, today)[source]

Compute the obligation Schedule from one canonical call.

This is the single producer of the pending-obligation datum. Both the operator state read-projection (pending_obligations) and the WorkflowEngine NO_PENDING_OBLIGATION gate route their schedule computation through here, so the gate and the projection cannot draw a divergent obligation set: identical (engine, profile, today) always yields an equal schedule (modulo Schedule.generated_at).

The fiscal year is derived from today so neither consumer can pass a mismatched (year, today) pair.

Parameters:
Return type:

Schedule

Returns:

The Schedule of obligations applicable to profile for today’s fiscal year.

applies_to(profile, modelo)[source]

Return whether registry deadline conditions match for modelo.

Parameters:
Return type:

bool

explain(profile, modelo)[source]

Return registry-backed deadline applicability text for modelo.

Parameters:
Return type:

str