dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/services.py
"""What several use cases share: who the caller is, an asset's plan and values built from rows, the gates, the journals.
Nothing here is a use case (no route, no tool): a use case decides WHO MAY and WHICH ROWS, then calls these. The rules
themselves live in `domain/`. Functions take a `Repository` (`repo`) so they run inside the caller's transaction.
"""
import json
from datetime import date
from decimal import Decimal
from fractions import Fraction
from typing import Optional
from erp_platform import seam
from brick_fixed_assets.domain import journals
from brick_fixed_assets.domain.changes import Delta, Values
from brick_fixed_assets.domain.depreciation import Add, DepKey, Open, Params, Plan, Reset, Retire, SetNbv
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import (
ZERO,
met,
period_of,
unmet,
)
from brick_fixed_assets.infrastructure import seams
from brick_fixed_assets.infrastructure.tables import Loaded, Repository
MOVEMENT = {"acquisition": "addition", "grant": "addition", "settlement": "transfer", "transfer": "transfer",
"retirement": "disposal", "revaluation": "revaluation", "impairment": "impairment",
"impairment_reversal": "impairment", "takeover": "addition"}
EVENT = {"acquisition": "acquire", "grant": "acquire", "settlement": "settle", "transfer": "transfer",
"retirement": "retire", "revaluation": "revalue", "impairment": "impair", "impairment_reversal": "impair",
"reversal": "reverse"}
DEC_KEYS = ("cost", "accumulated_depreciation", "depreciation_in_year", "revaluation_reserve", "gain_loss", "proceeds",
"catch_up")
# ── who ───────────────────────────────────────────────────────────────────────────────────────
def who(caller: dict) -> str:
"""The caller as a stored author is: their email, lower-case, else their sub."""
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def email_of(caller: dict) -> str:
return (caller.get("email") or "").strip().lower()
def sub_of(caller: dict) -> str:
return (caller.get("sub") or "").strip().lower()
# ── plans and values ──────────────────────────────────────────────────────────────────────────
def _fy_end(repo: Repository, company_code: str) -> int:
cache = repo.__dict__.setdefault("_fy_end", {})
if company_code not in cache:
cache[company_code] = seams.fiscal_year_end_month(company_code)
return cache[company_code]
def params_on(repo: Repository, a: Loaded, area_code: str, on: date) -> Optional[Params]:
row = a.area_on(area_code, on)
if row is None:
return None
k = repo.key(row["key_code"], on) if row["key_code"] else None
if k is None:
return None
key = DepKey(k["key_code"], k["method"], k["rate_pct"], k["life_periods"], k["base"] or "net_book_value",
k["acq_conv"] or "full_month", k["ret_conv"] or "full_month", k["stop_at"] or "scrap")
return Params(key, row["life_periods"], row["total_units"], row["scrap_value"] or ZERO)
def plan_events(repo: Repository, a: Loaded, area_code: str) -> list:
"""The events that shape the plan: posted transactions in this area (a reversed one and its reversal drop out)."""
out = []
for i, r in enumerate(repo.area_values_rows(a.asset_id, area_code, statuses=("posted",))):
if r["kind"] == "reversal":
continue
prm = json.loads(r["params"] or "{}")
ar = (prm.get("areas") or {}).get(area_code, {})
on, kind = r["value_date"], r["kind"]
if kind in ("acquisition", "grant", "settlement") and r["cost"]:
out.append(Add(on, r["cost"], i))
elif kind == "takeover" and ar:
out.append(Open(on, Decimal(ar["depreciable"]), int(ar["remaining"]), ar["first_period"], i))
elif kind == "transfer":
if r["cost"] > 0 and ar:
out.append(Open(on, Decimal(ar["depreciable"]), int(ar["remaining"]), ar["first_period"], i))
elif r["cost"] < 0:
out.append(Retire(on, Fraction(1), i))
elif kind == "retirement":
out.append(Retire(on, Fraction(prm.get("share_of_remaining", "1")), i))
elif kind in ("revaluation", "impairment", "impairment_reversal") and ar:
out.append(SetNbv(on, Decimal(ar["new_nbv"]), i))
return out
def build_plan(repo: Repository, a: Loaded, area_code: str) -> Optional[Plan]:
"""The planned depreciation of one asset in one area, or None when nothing was capitalised in it."""
events = plan_events(repo, a, area_code)
first = next((e for e in sorted(events, key=lambda e: (e.on, e.order)) if isinstance(e, (Add, Open))), None)
if first is None:
return None
initial = params_on(repo, a, area_code, first.on)
if initial is None:
return None
# a dated change of the asset's area row, or of the key it uses, is a Reset from its date (prospective)
dates = {r["valid_from"] for r in a.areas if r["area_code"] == area_code and r["valid_from"] > first.on}
for code in {r["key_code"] for r in a.areas if r["area_code"] == area_code and r["key_code"]}:
dates |= {k["valid_from"] for k in repo.rows_of_key(code) if k["valid_from"] > first.on}
last = initial
for n, day in enumerate(sorted(dates)):
p = params_on(repo, a, area_code, day)
if p is not None and p != last:
events.append(Reset(day, p, -1))
last = p
return Plan(initial, events, repo.usage(a.asset_id), _fy_end(repo, a.root["company_code"]))
def area_values(repo: Repository, asset_id: str, area_code: str) -> Values:
"""An asset's values in one area: the SUM of its transactions plus its depreciation postings. Never stored."""
cost = acc = reserve = ZERO
for r in repo.area_values_rows(asset_id, area_code):
cost += r["cost"] or ZERO
acc += r["accumulated_depreciation"] or ZERO
reserve += r["revaluation_reserve"] or ZERO
for p in repo.postings(asset_id, area_code):
acc += p["amount"] or ZERO
return Values(cost, acc, reserve)
def posted_total(repo: Repository, asset_id: str, area_code: str) -> Decimal:
return sum((p["amount"] or ZERO for p in repo.postings(asset_id, area_code)), ZERO)
def refresh_status(repo: Repository, a: Loaded) -> Loaded:
"""status and capitalised_on are DERIVED from the transactions: planned until something posts, active while any
area holds cost, retired when everything has left. Deactivated is the one thing set by hand."""
if a.root["status"] == "deactivated":
return a
rows = repo.tx_rows(a.asset_id, statuses=("posted", "reversed"))
if not rows:
a.root["status"] = "planned"
return a
acq = [r for r in rows if r["kind"] in ("acquisition", "takeover", "settlement", "transfer") and r["status"] == "posted"]
if acq and not a.root.get("capitalised_on"):
a.root["capitalised_on"] = min(r["value_date"] for r in acq)
held = any(area_values(repo, a.asset_id, c).cost > 0 for c in a.area_codes())
a.root["status"] = "active" if held else "retired"
return a
def set_dep_start(repo: Repository, a: Loaded) -> None:
"""dep_start (derived by the key's convention from the first capitalisation) on each area row."""
for area in a.area_codes():
plan = build_plan(repo, a, area)
if plan is None or not plan.segs:
continue
start = plan.segs[0].start
for r in a.areas:
if r["area_code"] == area and r["dep_start"] != start and r["valid_from"] <= (a.root.get("capitalised_on") or date.max):
r["dep_start"] = start
# ── gates ─────────────────────────────────────────────────────────────────────────────────────
def money_text(x) -> Optional[str]:
return None if x is None else str(x)
def tx_view(t: dict, areas: dict) -> dict:
"""A transaction as it is shown: the root fields and its values per area, amounts as exact strings."""
out = {k: (v.isoformat() if isinstance(v, date) else v) for k, v in t.items()
if k not in ("params", "version", "created_at", "updated_at")}
out["params"] = json.loads(t.get("params") or "{}")
out["version"] = t.get("version")
out["areas"] = [{**{k: (str(v) if isinstance(v, Decimal) else v) for k, v in a.items()
if k not in ("transaction_id", "created_at", "updated_at")}} for a in
sorted(areas.values(), key=lambda a: a["area_code"])]
return out
def area_row_values(delta: Delta) -> dict:
return {"cost": delta.cost, "accumulated_depreciation": delta.acc_dep, "depreciation_in_year": delta.depreciation_in_year,
"revaluation_reserve": delta.reserve, "gain_loss": delta.gain_loss, "proceeds": delta.proceeds,
"catch_up": delta.catch_up}
def delta_of(row: dict) -> Delta:
z = lambda k: row.get(k) or ZERO # noqa: E731
return Delta(z("cost"), z("accumulated_depreciation"), z("revaluation_reserve"), z("gain_loss"), z("proceeds"),
z("depreciation_in_year"), z("catch_up"))
# ── the journals ──────────────────────────────────────────────────────────────────────────────
def delta_lines(kind: str, d: Delta, acc: dict, asset_id: str, cost_centre: Optional[str], offset: Optional[dict],
memo: str = "") -> list:
"""The balanced lines of one asset event in one area, from its figures. The account the event is balanced against
(clearing, an offset account, inter-company clearing) absorbs the rest; anything else must balance on its own."""
mv = MOVEMENT.get(kind) or MOVEMENT.get(kind, "addition")
common = dict(asset_id=asset_id, cost_centre=cost_centre, movement=mv, memo=memo)
L: list = []
if d.cost:
L.append(journals.line("debit" if d.cost > 0 else "credit", abs(d.cost), account_id=acc["cost_account_id"], **common))
if d.acc_dep:
L.append(journals.line("credit" if d.acc_dep > 0 else "debit", abs(d.acc_dep),
account_id=acc["accumulated_depreciation_account_id"], **common))
if d.reserve:
L.append(journals.line("credit" if d.reserve > 0 else "debit", abs(d.reserve),
account_id=acc["revaluation_reserve_account_id"], **common))
if d.gain_loss:
a_kind = kind
if d.gain_loss > 0:
account = acc["impairment_reversal_account_id"] if a_kind == "impairment_reversal" else acc["gain_account_id"]
account = account or acc["gain_account_id"]
else:
account = acc["impairment_account_id"] if a_kind in ("impairment", "revaluation") else acc["loss_account_id"]
account = account or acc["loss_account_id"]
L.append(journals.line("credit" if d.gain_loss > 0 else "debit", abs(d.gain_loss), account_id=account, **common))
if d.catch_up:
pair_d = dict(account_id=acc["depreciation_expense_account_id"])
pair_c = dict(account_id=acc["accumulated_depreciation_account_id"])
journals.put(L, d.catch_up, pair_d, pair_c, **common)
debit = sum((Decimal(x["debit"]) for x in L if "debit" in x), ZERO)
credit = sum((Decimal(x["credit"]) for x in L if "credit" in x), ZERO)
if debit != credit and offset is not None:
diff = debit - credit
L.append(journals.line("credit" if diff > 0 else "debit", abs(diff), **offset, **{**common, "movement": mv}))
return L # a leg of a group may be unbalanced alone; the journal is checked whole before it is sent
def missing_accounts(acc: Optional[dict], kind: str, d: Delta) -> list:
"""Which account of the class's determination a figure needs and the class lacks."""
if acc is None:
return ["every account"]
need = []
if d.cost:
need.append("cost_account_id")
if d.acc_dep or d.catch_up:
need.append("accumulated_depreciation_account_id")
if d.catch_up:
need.append("depreciation_expense_account_id")
if d.reserve:
need.append("revaluation_reserve_account_id")
if d.gain_loss > 0:
need.append("gain_account_id")
elif d.gain_loss < 0:
need.append("impairment_account_id" if kind in ("impairment", "revaluation") else "loss_account_id")
return [n for n in need if acc.get(n) is None]
def journal_body(source_document: str, event: str, company_code: str, ledger_id: str, journal_date: date, currency: str,
memo: str, lines: list) -> dict:
return {"journal_id": str(seam.derived_id(source_document, event, ledger_id)), "source_module": "fixed-assets",
"source_document": source_document, "event": event, "company_code": company_code, "ledger_id": str(ledger_id),
"journal_date": journal_date.isoformat(), "currency": currency, "memo": memo, "lines": lines}
def period_gates(day: date, company_code: str, areas: list, actor: str = "") -> list:
"""The period_open gate per posting area's ledger — one Gate each, named by area."""
out = []
for a in areas:
st = seams.period_open(day, company_code, a["ledger_id"], actor)
name = f"period_open:{a['area_code']}"
if st.open:
out.append(met(name))
else:
reason = st.reason or f"the period of {day.isoformat()} is closed in ledger {a['ledger_id']}"
out.append(unmet(name, f"area {a['area_code']} (ledger {a['ledger_id']}): {reason}",
st.fix or "ask the accountant to reopen the period in gl"))
return out
# ── posting a transaction group ───────────────────────────────────────────────────────────────
def _offset_of(params: dict) -> Optional[dict]:
o = params.get("offset")
if not o:
return None
return {"account_id": int(o["account_id"])} if o.get("account_id") is not None else {"purpose": o["purpose"]}
def plan_journals(repo: Repository, head_id: str, actor: str) -> list:
"""[(company_code, area, source tx id, body)] for a group: one journal per company and POSTING area. The tax area
(and any area with no ledger or posting none) is recorded but never posted."""
today = repo.today()
legs = repo.group(head_id)
if not legs:
raise RuleBroken(f"no transaction {head_id}")
out = []
by_company: dict = {}
for leg in legs:
by_company.setdefault(leg["company_code"], []).append(leg)
txs = repo.transactions(l["transaction_id"] for l in legs) # every leg's transaction and asset in one round trip each
assets = repo.assets(l["asset_id"] for l in legs)
for company, cl in by_company.items():
areas = {a["area_code"]: a for a in repo.posting_areas(today, company)} # each company posts to ITS areas' ledgers
for code, area in sorted(areas.items()):
lines: list = []
first = cl[0]
for leg in cl:
tx = txs[leg["transaction_id"]]
row = tx["areas"].get(code)
if row is None:
continue
d = delta_of(row)
if not (d.cost or d.acc_dep or d.reserve or d.gain_loss or d.catch_up or d.proceeds):
continue
a = assets[leg["asset_id"]]
acc = repo.account_map(a.root["class_code"], code, today)
prm = json.loads(leg["params"] or "{}")
kind = prm.get("of_kind") or leg["kind"]
gap = missing_accounts(acc, kind, d)
if gap:
raise RuleBroken(f"class {a.root['class_code']} has no {', '.join(gap)} in force for area {code}: the "
"owner sets account determination (Settings)")
cc = (a.master_on(leg["value_date"]) or a.master_on(today) or {}).get("cost_centre")
lines += delta_lines(kind, d, acc, leg["asset_id"], cc, _offset_of(prm), f"{kind} {leg['asset_id']}")
if not lines:
continue
journals.check_balanced(lines)
event = EVENT[first["kind"]]
body = journal_body(f"fa:{first['transaction_id']}", event, company, area["ledger_id"],
first["value_date"] if first["kind"] != "reversal" else today,
area["currency"] or "GBP", f"{event} {first['asset_id']}", lines)
out.append((company, code, first["transaction_id"], body))
return out
def send_journals(bodies: list, actor: str) -> dict:
"""Post each journal to gl; {area: (journal_id, period_id)}. Refused with gl's own words if any is refused — the
ones already accepted stay (journal ids derive from the document, so a retry replays them)."""
done: dict = {}
for company, code, src, body in bodies:
r = seams.post_journal(body, actor)
if not r.ok:
raise RuleBroken(f"gl refused the journal for area {code} ({r.gate or r.status}): {r.reason}"
+ (f" — {r.fix}" if r.fix else ""))
done[(company, code)] = r.journal_id
return done
def mark_posted(repo: Repository, head_id: str, journal_ids: dict, today: date, actor: str) -> list:
"""The write phase: every leg posted, its journal ids kept, depreciation catch-up recorded, assets refreshed."""
legs = repo.group(head_id)
result = []
txs = repo.transactions(l["transaction_id"] for l in legs) # every leg's transaction, asset and reversed original: one round trip each
assets = repo.assets(l["asset_id"] for l in legs)
originals = repo.transactions(o for l in legs for o in json.loads(l["params"] or "{}").get("reverses", []))
for leg in legs:
tx = txs[leg["transaction_id"]]
root, areas = tx["root"], tx["areas"]
prm = json.loads(root["params"] or "{}")
for code, row in areas.items():
jid = journal_ids.get((root["company_code"], code))
if jid is not None:
row["journal_id"] = jid
area = repo.area(code, today)
row["ledger_id"] = area["ledger_id"] if area else None
if row.get("catch_up"):
kind = "reversal" if root["kind"] == "reversal" else "catch_up"
a = assets[root["asset_id"]]
cc = (a.master_on(root["value_date"]) or {}).get("cost_centre")
from erp_platform.repository import upsert
from brick_fixed_assets.infrastructure import tables as T
upsert(repo.conn, T.fa_dep_posting, [{
"asset_id": root["asset_id"], "area_code": code, "period_id": period_of(root["value_date"]),
"run_id": root["transaction_id"], "amount": row["catch_up"], "cost_centre": cc,
"class_code": a.root["class_code"], "journal_id": jid, "kind": kind, "posted_on": today,
"company_code": root["company_code"]}], ("asset_id", "area_code", "period_id", "run_id"))
root["status"] = "posted"
root["posted_on"] = today
root["posting_period_id"] = period_of(root["value_date"])
repo.save_transaction(root, areas, root["version"])
if root["kind"] == "reversal":
for orig in prm.get("reverses", []):
o = originals.get(orig)
if o:
o["root"]["status"] = "reversed"
repo.save_transaction(o["root"], o["areas"], o["version"])
a = assets[root["asset_id"]]
refresh_status(repo, a)
set_dep_start(repo, a)
repo.save_asset(a)
result.append(root["transaction_id"])
return result
# ── writing a transaction ─────────────────────────────────────────────────────────────────────
def new_tx(repo: Repository, caller: dict, *, transaction_id: str, asset: Loaded, kind: str, value_date: date,
deltas: dict, params: dict, status: str = "approved", group_id: Optional[str] = None, leg: str = "main",
source_ref: Optional[str] = None, evidence_key: Optional[str] = None, asset_version: Optional[int] = None,
company_code: Optional[str] = None) -> dict:
"""Write one transaction with its values per area (`deltas`: {area_code: Delta}). Proposed by `caller`."""
from brick_fixed_assets.domain.values import fiscal_year_of
company = company_code or asset.root["company_code"]
root = {"transaction_id": transaction_id, "asset_id": asset.asset_id, "company_code": company,
"group_id": group_id or transaction_id, "leg": leg, "kind": kind, "value_date": value_date,
"posting_period_id": period_of(value_date), "fiscal_year": fiscal_year_of(value_date, _fy_end(repo, company)),
"status": status, "proposed_by_email": email_of(caller), "proposed_by_sub": sub_of(caller),
"approved_by_email": None, "approved_by_sub": None, "approved_on": None, "posted_on": None,
"rejected_reason": None, "reverses_transaction_id": None, "evidence_key": evidence_key,
"source_ref": source_ref, "params": json.dumps(params, sort_keys=True), "asset_version": asset_version}
areas = {code: {"transaction_id": transaction_id, "area_code": code, **area_row_values(d), "journal_id": None,
"ledger_id": None} for code, d in deltas.items()}
root["version"] = repo.save_transaction(root, areas, 0)
return root
def check_year_open(repo: Repository, company_code: str, value_date: date) -> None:
from brick_fixed_assets.domain.values import fiscal_year_of
fy = fiscal_year_of(value_date, _fy_end(repo, company_code))
if repo.year_closed(company_code, fy):
raise RuleBroken(f"the asset year {fy} is closed for company {company_code}: nothing dated in it can be posted "
"(year_open) — correct it in the next year")
def post_group_now(repository, head_id: str, actor: str) -> list:
"""Post a transaction group to gl (one journal per company and posting area), then mark it posted — three short steps:
plan the journals (read), send them (gl; idempotent on journal_id, so a retry replays), record the result (write)."""
with repository() as repo:
bodies = plan_journals(repo, head_id, actor)
ids = send_journals(bodies, actor)
with repository() as repo:
today = repo.today(True)
return mark_posted(repo, head_id, ids, today, actor)
def tx_answer(repo: Repository, transaction_id: str) -> dict:
tx = repo.transaction(transaction_id)
return tx_view(tx["root"], tx["areas"])
def reverse_group(repo: Repository, caller: dict, head_id: str, today: date, reason: str, status: str = "approved") -> str:
"""Write the reversing transactions of a whole group: every leg's values negated, dated today, keyed 'rev:<id>', its
journals the original's the other way round. The originals are marked reversed when the reversal posts. Returns the
reversing head's id."""
legs = repo.group(head_id)
rev_head = f"rev:{head_id}"
txs = repo.transactions(l["transaction_id"] for l in legs)
assets = repo.assets(l["asset_id"] for l in legs)
for leg in legs:
tx = txs[leg["transaction_id"]]
prm = json.loads(leg["params"] or "{}")
deltas = {code: delta_of(row).negated() for code, row in tx["areas"].items()}
a = assets[leg["asset_id"]]
params = {**prm, "of_kind": prm.get("of_kind") or leg["kind"], "reverses": [leg["transaction_id"]], "reason": reason}
t = new_tx(repo, caller, transaction_id=f"rev:{leg['transaction_id']}", asset=a, kind="reversal", value_date=today,
deltas=deltas, params=params, status=status, group_id=rev_head, leg=leg["leg"],
source_ref=f"reverses:{leg['transaction_id']}", asset_version=a.version, company_code=leg["company_code"])
t["reverses_transaction_id"] = leg["transaction_id"]
tx2 = repo.transaction(t["transaction_id"])
tx2["root"]["reverses_transaction_id"] = leg["transaction_id"]
repo.save_transaction(tx2["root"], tx2["areas"], tx2["version"])
return rev_head