dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/uc_assets.py
"""The register: classes' assets with their sub-numbers (FA-03), dated changes (FA-04), AI suggestions (FA-05), the
acquisition from purchasing (FA-06), manual acquisitions and grants (FA-07), the capitalisation threshold (FA-08), assets
under construction and their settlement (FA-09) and units of use (FA-11).
An asset is created PLANNED: nothing is capitalised until an acquisition posts. Its status and capitalised_on are derived
from its transactions. asset_no is stamped once from the class series, never reused; the same request id returns the same
asset. The tax area (and any area that posts nothing) is recorded but never posted."""
import json
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, Limit, Name, Note, text
from brick_fixed_assets.application import services
from brick_fixed_assets.application import use_cases as hub
from brick_fixed_assets.application.inputs import (
AreaChangeIn,
AssetAreaIn,
Blankable,
Code,
Currency,
LineRuleIn,
Money,
Period,
Reason,
Seq,
Units,
day,
)
from brick_fixed_assets.domain import settlement
from brick_fixed_assets.domain.changes import Delta
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import (
MAX_AMOUNT,
POOLS,
ZERO,
bound_amount,
ids_of,
met,
money,
period_of,
unmet,
)
from brick_fixed_assets.infrastructure import ai_steps, seams
from brick_fixed_assets.infrastructure import tables as T
from brick_fixed_assets.infrastructure.tables import EPOCH, Loaded
FA = hub.FA
def _gate_body(gate: str, reason: str, fix: str) -> dict:
return {"gate": gate, "met": False, "reason": reason, "fix": fix}
def _refuse_gate(status: int, gate: str, reason: str, fix: str) -> Refused:
"""The seam shape: a refusal that is {gate, met: false, reason, fix} (purchasing reads it)."""
return Refused(status, _gate_body(gate, reason, fix))
# ── who sees which asset ──────────────────────────────────────────────────────────────────────
def _own(caller: dict, a: Loaded, today: date) -> bool:
"""A custodian sees the assets they look after: the current master row names them (by email or sub)."""
m = a.master_on(today) or {}
mine = ids_of(caller.get("email"), caller.get("sub"))
return bool(mine) and (m.get("custodian_email") or "").strip().lower() in mine
def _asset(repo, asset_id: str) -> Loaded:
a = repo.asset(asset_id)
if a is None:
raise Refused(404, f"no asset {asset_id}")
return a
def _visible(repo, caller: dict, asset_id: str, own_only: bool = False) -> Loaded:
"""The asset, or 404 (never 403) when the caller may not see it."""
a = _asset(repo, asset_id)
if not own_only and hub.sees_all_assets(caller):
return a
if _own(caller, a, hub.today(repo)):
return a
raise Refused(404, f"no asset {asset_id}")
# ── how an asset reads ────────────────────────────────────────────────────────────────────────
def _fmt(v):
if isinstance(v, date):
return v.isoformat()
if isinstance(v, Decimal):
return str(v)
return v
def readiness(repo, a: Loaded, today: date) -> list:
"""purchasing's asset_assignable gates, as {gate, met, reason, fix}: the asset can take an acquisition."""
gates = []
st = a.root["status"]
gates.append(met("not_retired") if st not in ("retired", "deactivated") else
unmet("not_retired", "the asset is fully retired" if st == "retired" else "the asset is deactivated",
"assign a new asset, or a new sub-number"))
k = repo.klass(a.root["class_code"], today)
gates.append(met("class_in_force") if k else unmet("class_in_force", "the asset's class has no row in force today",
"Settings: asset classes (owner)"))
return gates
def asset_view(repo, a: Loaded, caller: dict, full: bool = True) -> dict:
today = hub.today(repo)
r = a.root
m = a.master_on(today) or {}
k = repo.klass(r["class_code"], today)
out = {"asset_id": r["asset_id"], "company_code": r["company_code"], "asset_no": r["asset_no"], "sub_no": r["sub_no"],
"class_code": r["class_code"], "description": r["description"], "status": r["status"],
"is_auc": bool(r["is_auc"]), "cost_centre": m.get("cost_centre"),
"capitalisation_threshold": str(repo.setting("capitalisation_threshold", today)),
"readiness": [g.as_dict() for g in readiness(repo, a, today)], "version": a.version}
if not full:
return out
out.update({"serial_no": r["serial_no"], "inventory_no": r["inventory_no"], "bp_id": r["bp_id"],
"capitalised_on": _fmt(r["capitalised_on"]), "last_counted_on": _fmt(r["last_counted_on"]),
"quantity": _fmt(r["quantity"]), "ca_pool": r["ca_pool"], "is_car": bool(r["is_car"]),
"new_and_unused": bool(r["new_and_unused"]), "location_id": m.get("location_id"),
"custodian_email": m.get("custodian_email"), "class_name": k["name"] if k else None})
out["master"] = [{k2: _fmt(v) for k2, v in row.items() if k2 not in ("asset_id", "created_at", "updated_at")}
for row in a.master]
out["areas"] = [{k2: _fmt(v) for k2, v in row.items() if k2 not in ("asset_id", "created_at", "updated_at")}
for row in a.areas]
period = period_of(today)
values = []
for code in a.area_codes():
v = services.area_values(repo, a.asset_id, code)
plan = services.build_plan(repo, a, code)
values.append({"area_code": code, "cost": str(v.cost), "accumulated_depreciation": str(v.acc_dep),
"net_book_value": str(v.nbv), "revaluation_reserve": str(v.reserve),
"planned_to_date": str(plan.cum(period)) if plan else "0.00",
"posted_to_date": str(services.posted_total(repo, a.asset_id, code))})
out["values"] = values
if hub.auth.has_permission(caller, hub.TX_WRITE) or hub.auth.has_permission(caller, hub.TX_APPROVE) \
or hub.auth.has_permission(caller, hub.REPORT):
out["transactions"] = [{"transaction_id": t["transaction_id"], "kind": t["kind"], "status": t["status"],
"value_date": _fmt(t["value_date"])} for t in
repo.tx_rows(a.asset_id, statuses=("proposed", "approved", "posted", "rejected", "reversed"))]
out["suggestions"] = [{k2: _fmt(v) for k2, v in f.items() if k2 not in ("created_at", "updated_at")}
for f in repo.findings(f"asset:{a.asset_id}")]
return out
def _summary(repo, a: Loaded, masters: dict) -> dict:
m = masters.get(a.asset_id) or {}
r = a.root
return {"asset_id": r["asset_id"], "company_code": r["company_code"], "asset_no": r["asset_no"], "sub_no": r["sub_no"],
"class_code": r["class_code"], "description": r["description"], "status": r["status"],
"inventory_no": r["inventory_no"], "cost_centre": m.get("cost_centre"), "location_id": m.get("location_id")}
def _page(repo, caller, company_code, class_code, status, q, limit, after, own_only) -> dict:
today = hub.today(repo)
masters = repo.masters_on(today)
out = []
mine = ids_of(caller.get("email"), caller.get("sub"))
needle = (q or "").strip().lower()
for root in repo.asset_roots(company_code, class_code, [status] if status else None):
if after and root["asset_id"] <= after:
continue
m = masters.get(root["asset_id"]) or {}
if own_only or not hub.sees_all_assets(caller):
if not mine or (m.get("custodian_email") or "").strip().lower() not in mine:
continue
if needle and needle not in " ".join(str(root.get(f) or "") for f in
("asset_id", "description", "inventory_no", "serial_no")).lower():
continue
out.append(Loaded(root, [], [], root["version"]))
if len(out) >= limit:
break
rows = [_summary(repo, a, masters) for a in out]
return {"assets": rows, "next_after": rows[-1]["asset_id"] if len(rows) >= limit else None}
@FA.use_case(kind="query", route="GET /assets", permission=hub.READ)
def list_assets(caller: dict, company_code: Optional[Code] = None, class_code: Optional[Code] = None,
status: Optional[Code] = None, q: Optional[Name] = None, limit: Limit = 50,
after: Optional[Code] = None) -> dict:
"""The assets of the companies you keep, filtered by company, class, status (planned | active | retired |
deactivated) or a search over the number, description, tag and serial number, ordered by asset_id. `after` is the last
asset_id of the previous page."""
with hub.repository() as repo:
return _page(repo, caller, company_code, class_code, status, q, limit, after, False)
@FA.use_case(kind="query", route="GET /my/assets", permission=hub.READ_OWN)
def list_my_assets(caller: dict, q: Optional[Name] = None, limit: Limit = 50, after: Optional[Code] = None) -> dict:
"""The assets you look after (the custodian): those whose current master row names you. Another site's asset does not
appear and is not found when asked for."""
with hub.repository() as repo:
return _page(repo, caller, None, None, None, q, limit, after, True)
@FA.use_case(kind="query", route="GET /my/assets/{asset_id}", permission=hub.READ_OWN)
def get_my_asset(caller: dict, asset_id: Code) -> dict:
"""One asset you look after, with its values. An asset that is not yours is not found."""
with hub.repository() as repo:
a = _visible(repo, caller, asset_id, own_only=True)
return asset_view(repo, a, caller, full=False) | {"values": asset_view(repo, a, caller)["values"]}
@FA.use_case(kind="query", route="GET /assets/{asset_id}", permission=hub.READ)
def get_asset(caller: dict, asset_id: Code) -> dict:
"""One asset: its master over time, the key and life per area, its values per area (cost, accumulated depreciation,
net book value: the SUM of its transactions — never stored), planned against posted depreciation, its transactions
and AI suggestions. Purchasing reads `readiness` before it lets an order line name the asset."""
with hub.repository() as repo:
return asset_view(repo, _visible(repo, caller, asset_id), caller)
# ── creating assets ───────────────────────────────────────────────────────────────────────────
def _area_rows(repo, class_code: str, today: date, overrides: Optional[list], master_start: date = EPOCH) -> list:
"""The asset's area rows: the class's defaults per area (key, life), each overridable; scrap 0."""
over = {o.area_code: o for o in overrides or []}
defaults = {c["area_code"]: c for c in repo.class_areas(class_code, today)}
codes = sorted(set(defaults) | set(over))
rows = []
for code in codes:
area = repo.area(code, today)
if area is None:
raise RuleBroken(f"area {code} is not defined today: define it, or leave it out")
d, o = defaults.get(code), over.get(code)
key_code = (o.key_code if o and o.key_code else None) or (d["key_code"] if d else None)
if key_code is None:
raise RuleBroken(f"area {code} has no depreciation key: the class has no default and none was given")
k = repo.key(key_code, today)
if k is None:
raise RuleBroken(f"depreciation key {key_code} is not defined today")
life = (o.life_periods if o and o.life_periods else None) or (d["life_periods"] if d else None)
rows.append({"area_code": code, "valid_from": master_start, "key_code": key_code,
"life_periods": life or k["life_periods"], "total_units": o.total_units if o else None,
"scrap_value": (o.scrap_value if o and o.scrap_value is not None else ZERO), "dep_start": None})
return rows
def _new_asset(repo, caller: dict, *, company_code: str, class_code: str, description: str, request_id: str,
asset_no: str, sub_no: int, serial_no, inventory_no, bp_id, quantity, cost_centre, location_id,
custodian_email, is_car, new_and_unused, ca_pool, areas, today: date) -> Loaded:
k = repo.klass(class_code, today)
asset_id = f"{company_code}-{asset_no}-{sub_no}"
if location_id:
locs = seams.locations(services.who(caller))
if locs is not None and location_id not in {x.get("location_id") for x in locs}:
raise RuleBroken(f"inventory has no location {location_id}: pick one it has")
root = {"asset_id": asset_id, "company_code": company_code, "asset_no": asset_no, "sub_no": sub_no,
"class_code": class_code, "description": description, "serial_no": serial_no, "inventory_no": inventory_no,
"bp_id": bp_id, "capitalised_on": None, "status": "planned", "last_counted_on": None,
"quantity": quantity, "request_id": request_id, "is_auc": bool(k["is_auc"]), "is_car": bool(is_car),
"new_and_unused": bool(new_and_unused), "ca_pool": ca_pool or k["default_ca_pool"],
"deactivated_reason": None, "created_by": services.who(caller), "version": 0}
master = [{"asset_id": asset_id, "valid_from": EPOCH, "cost_centre": cost_centre, "location_id": location_id,
"custodian_email": (custodian_email or "").strip().lower() or None,
"set_by_email": services.who(caller)}]
arows = [{"asset_id": asset_id, **r} for r in _area_rows(repo, class_code, today, areas)]
a = Loaded(root, master, arows, 0)
a.version = 0
return a
def _gate_class(repo, class_code: str, today: date) -> dict:
k = repo.klass(class_code, today)
if k is None:
raise RuleBroken(f"class {class_code} has no row in force today (class_in_force): the owner sets the asset "
"classes in Settings")
return k
@FA.use_case(kind="command", route="POST /assets", permission=hub.WRITE, idempotent=True)
def create_asset(caller: dict, request_id: Code, company_code: Code, class_code: Code, description: Name,
serial_no: Optional[Name] = None, inventory_no: Optional[Name] = None, bp_id: Optional[BigId] = None,
quantity: Units = Decimal("1"), cost_centre: Optional[Code] = None, location_id: Optional[Code] = None,
custodian_email: Optional[text(254)] = None, is_car: bool = False, new_and_unused: bool = True,
ca_pool: Optional[Code] = None, areas: Optional[list[AssetAreaIn]] = None) -> dict:
"""Create an asset (sub-number 0) in a class and company. Its number (asset_no, e.g. 2026-PLANT-0007) is stamped ONCE
from the class series and never reused; the asset_id is <company>-<asset_no>-0. Each area takes the class's key and
life, which you may override. A second create with the same request_id returns the same asset. Nothing is
capitalised until an acquisition posts: the asset is PLANNED. Use add_sub_asset for a component that wears out at
a different rate (IAS 16.43)."""
if ca_pool is not None and ca_pool not in POOLS:
raise Refused(422, f"ca_pool is one of {', '.join(POOLS)}")
def unit():
with hub.repository() as repo:
known = repo.asset_by_request(request_id)
if known:
return asset_view(repo, _asset(repo, known), caller) | {"replay": True}
today = hub.today(repo, True)
k = _gate_class(repo, class_code, today)
fmt = k["number_format"] or repo.setting("asset_number_format", today)
if "{seq" not in fmt:
raise RuleBroken("the class has no number series for this year (number_series): the owner sets the "
"number format in Settings")
asset_no = repo.claim_number(class_code, today.year, fmt, request_id)
a = _new_asset(repo, caller, company_code=company_code, class_code=class_code, description=description,
request_id=request_id, asset_no=asset_no, sub_no=0, serial_no=serial_no,
inventory_no=inventory_no, bp_id=bp_id, quantity=quantity, cost_centre=cost_centre,
location_id=location_id, custodian_email=custodian_email, is_car=is_car,
new_and_unused=new_and_unused, ca_pool=ca_pool, areas=areas, today=today)
from sqlalchemy import insert
repo.conn.execute(insert(T.fa_asset_request).values(request_id=request_id, asset_id=a.asset_id,
company_code=company_code))
a.version = repo.save_asset(a)
return asset_view(repo, a, caller) | {"replay": False}
with hub.decide():
return hub.retrying(unit)
@FA.use_case(kind="command", route="POST /assets/{asset_id}/sub-assets", permission=hub.WRITE, idempotent=True)
def add_sub_asset(caller: dict, asset_id: Code, request_id: Code, description: Name, serial_no: Optional[Name] = None,
inventory_no: Optional[Name] = None, areas: Optional[list[AssetAreaIn]] = None,
quantity: Units = Decimal("1")) -> dict:
"""Add a component (or a later addition) of an asset as the next sub-number: it keeps the main asset's number, company
and class, and has its own key and life per area (the roof and the building, the engine and the lorry). The same
request_id returns the same sub-asset."""
def unit():
with hub.repository() as repo:
known = repo.asset_by_request(request_id)
if known:
return asset_view(repo, _asset(repo, known), caller) | {"replay": True}
main = _visible(repo, caller, asset_id)
if main.root["status"] == "deactivated":
raise RuleBroken("the main asset is deactivated: add the component to a live asset")
today = hub.today(repo, True)
_gate_class(repo, main.root["class_code"], today)
same = [r["sub_no"] for r in repo.asset_roots(main.root["company_code"], main.root["class_code"])
if r["asset_no"] == main.root["asset_no"]]
sub_no = repo.first_free(same)
repo.claim_slot("SUB", f"{main.root['company_code']}-{main.root['asset_no']}-{sub_no}", request_id)
m = main.master_on(today) or {}
a = _new_asset(repo, caller, company_code=main.root["company_code"], class_code=main.root["class_code"],
description=description, request_id=request_id, asset_no=main.root["asset_no"],
sub_no=sub_no, serial_no=serial_no, inventory_no=inventory_no, bp_id=main.root["bp_id"],
quantity=quantity, cost_centre=m.get("cost_centre"), location_id=m.get("location_id"),
custodian_email=m.get("custodian_email"), is_car=main.root["is_car"],
new_and_unused=main.root["new_and_unused"], ca_pool=main.root["ca_pool"], areas=areas,
today=today)
from sqlalchemy import insert
repo.conn.execute(insert(T.fa_asset_request).values(request_id=request_id, asset_id=a.asset_id,
company_code=a.root["company_code"]))
a.version = repo.save_asset(a)
return asset_view(repo, a, caller) | {"replay": False}
with hub.decide():
return hub.retrying(unit)
# ── dated changes ─────────────────────────────────────────────────────────────────────────────
def _apply_changes(repo, a: Loaded, caller: dict, start: date, today: date, cost_centre, location_id, custodian_email,
areas: Optional[list]) -> list:
"""Append the dated master / area rows for one asset; returns what changed. Posted periods are never recomputed:
the plan is prospective from `start`."""
if start <= today:
raise RuleBroken(f"a change starts after today ({today.isoformat()}) at the earliest (starts_after_today): "
"pick a later date")
if a.root["status"] in ("retired", "deactivated"):
raise RuleBroken("a retired or deactivated asset takes no change")
changed = []
if cost_centre is not None or location_id is not None or custodian_email is not None:
cur = a.master_on(start) or {}
if location_id:
locs = seams.locations(services.who(caller))
if locs is not None and location_id not in {x.get("location_id") for x in locs}:
raise RuleBroken(f"inventory has no location {location_id}: pick one it has")
row = {"asset_id": a.asset_id, "valid_from": start,
"cost_centre": cost_centre if cost_centre is not None else cur.get("cost_centre"),
"location_id": location_id if location_id is not None else cur.get("location_id"),
"custodian_email": ((custodian_email or "").strip().lower() or None) if custodian_email is not None
else cur.get("custodian_email"), "set_by_email": services.who(caller)}
a.master = [m for m in a.master if m["valid_from"] != start] + [row]
a.master.sort(key=lambda m: m["valid_from"])
changed.append("master")
for ch in areas or []:
cur = a.area_on(ch.area_code, start)
if cur is None:
raise RuleBroken(f"the asset has no area {ch.area_code}")
key_code = ch.key_code or cur["key_code"]
if repo.key(key_code, start) is None:
raise RuleBroken(f"depreciation key {key_code} is not defined on {start.isoformat()}")
row = {"asset_id": a.asset_id, "area_code": ch.area_code, "valid_from": start, "key_code": key_code,
"life_periods": ch.life_periods or cur["life_periods"],
"total_units": ch.total_units if ch.total_units is not None else cur["total_units"],
"scrap_value": ch.scrap_value if ch.scrap_value is not None else cur["scrap_value"],
"dep_start": cur["dep_start"]}
a.areas = [r for r in a.areas if not (r["area_code"] == ch.area_code and r["valid_from"] == start)] + [row]
changed.append(f"area {ch.area_code}")
a.areas.sort(key=lambda r: (r["area_code"], r["valid_from"]))
return changed
@FA.use_case(kind="command", route="POST /assets/{asset_id}/master", permission=hub.WRITE, if_match="version")
def change_asset_master(caller: dict, asset_id: Code, valid_from: IsoDate, version: Optional[int] = None,
cost_centre: Optional[Code] = None, location_id: Optional[Code] = None,
custodian_email: Optional[text(254)] = None, areas: Optional[list[AreaChangeIn]] = None) -> dict:
"""Change where an asset is, who looks after it, or its life, key, scrap or total units, FROM a date after today. Both
are new dated rows, never edits: depreciation from that date is charged to the new cost centre; from it the remaining
net book value is spread over the new remaining life (prospective, IAS 8.36) and posted periods are never
recomputed. A date of today or earlier is refused."""
def unit():
with hub.repository() as repo:
a = _visible(repo, caller, asset_id)
if version is not None and int(version) != a.version:
raise Refused(409, f"asset {asset_id} changed since you opened it (you have version {version}, it is at "
f"{a.version}) — load it again")
today = hub.today(repo, True)
start = day(valid_from)
changed = _apply_changes(repo, a, caller, start, today, cost_centre, location_id, custodian_email, areas)
if not changed:
raise RuleBroken("say what changes: a cost centre, location, custodian, or a life, key, scrap or units")
a.version = repo.save_asset(a)
return asset_view(repo, a, caller) | {"changed": changed, "valid_from": start.isoformat()}
with hub.decide():
return hub.retrying(unit, 1)
@FA.use_case(kind="command", route="POST /asset-classes/{class_code}/mass-change", permission=hub.WRITE)
def mass_change_assets(caller: dict, class_code: Code, valid_from: IsoDate, company_code: Optional[Code] = None,
cost_centre: Optional[Code] = None, areas: Optional[list[AreaChangeIn]] = None) -> dict:
"""The same dated change for every live asset of a class (and company): one mass change that lists each asset it
touched and each it skipped, with the reason. Prospective from the date, like change_asset_master."""
with hub.repository() as repo, hub.decide():
today = hub.today(repo, True)
start = day(valid_from)
touched, skipped = [], []
roots = repo.asset_roots(company_code, class_code, ["planned", "active"])
loaded = repo.assets(r["asset_id"] for r in roots) # every asset of the class in sets, not three queries each
for root in roots:
a = loaded[root["asset_id"]]
try:
_apply_changes(repo, a, caller, start, today, cost_centre, None, None, areas)
repo.save_asset(a)
touched.append(a.asset_id)
except RuleBroken as why:
skipped.append({"asset_id": a.asset_id, "reason": str(why)})
return {"class_code": class_code, "valid_from": start.isoformat(), "touched": touched, "skipped": skipped}
@FA.use_case(kind="command", route="POST /assets/{asset_id}/deactivate", permission=hub.WRITE, idempotent=True,
destructive=True)
def deactivate_asset(caller: dict, asset_id: Code, reason: Reason) -> dict:
"""Deactivate an asset that was never capitalised (no transaction in any area). It is kept with its number, which is
never reused. An asset with transactions is retired instead (propose_retirement)."""
with hub.repository() as repo, hub.decide():
a = _visible(repo, caller, asset_id)
if a.root["status"] == "deactivated":
return asset_view(repo, a, caller)
if repo.tx_rows(a.asset_id, statuses=("proposed", "approved", "posted", "reversed")):
raise RuleBroken("the asset has transactions (never_capitalised): retire it instead (propose_retirement)")
a.root["status"] = "deactivated"
a.root["deactivated_reason"] = reason
a.version = repo.save_asset(a)
return asset_view(repo, a, caller)
# ── acquisition ───────────────────────────────────────────────────────────────────────────────
def _acquisition_answer(repo, acquisition_ref: str, transaction_id: str, replay: bool) -> dict:
tx = repo.transaction(transaction_id)
return {"acquisition_ref": acquisition_ref, "transaction_id": transaction_id, "asset_id": tx["root"]["asset_id"],
"status": tx["root"]["status"],
"areas": [{"area_code": c, "cost": str(r["cost"]), "journal_id": r["journal_id"], "ledger_id": r["ledger_id"]}
for c, r in sorted(tx["areas"].items())], "replay": replay}
def _post_gates(repo, company_code: str, value_date: date, actor: str) -> None:
gates = services.period_gates(value_date, company_code, repo.posting_areas(hub.today(repo), company_code), actor)
bad = [g for g in gates if not g.met]
if bad:
raise _refuse_gate(409, "period_open", "; ".join(g.reason for g in bad), bad[0].fix)
@FA.use_case(kind="command", route="POST /acquisitions", permission=hub.INTAKE, idempotent=True, tool=False)
def take_acquisition(caller: dict, acquisition_ref: Code, source_kind: Code, company_code: Blankable, bill_id: Code,
line_no: Seq, value_date: IsoDate, amount: Money, currency: Currency, asset_id: Optional[Code] = None,
class_code: Optional[Blankable] = None, description: Optional[Name] = None, bp_id: Optional[BigId] = None,
purchase_order_id: Optional[Code] = None, quantity: Optional[Units] = None,
cost_centre: Optional[Code] = None) -> dict:
"""Purchasing hands the value of an asset bought on a supplier's invoice to the register (FA-06): one transaction of
kind acquisition per bill line, dated the BILL's date (a fact dated by its document), with the amount in every
area; one journal per posting area debits the class's cost account and credits asset clearing (purchasing's own journal
debited it in every ledger, so clearing nets to zero). Idempotent on acquisition_ref ('pur:<bill_id>:<line_no>'): a
retry posts nothing new. A credit is the same call with a negative amount. The asset is `asset_id`, or created from
`class_code` + `description`. 409 {gate: same_company | not_retired | period_open}, 422 {gate: bounds}."""
actor = services.who(caller)
if source_kind not in ("purchase_bill", "purchase_credit"):
raise Refused(422, _gate_body("bounds", "source_kind is purchase_bill or purchase_credit", "fix the call"))
if amount == 0:
raise Refused(422, _gate_body("bounds", "amount is not zero", "send the line's amount (negative for a credit)"))
if source_kind == "purchase_credit" and amount > 0:
raise Refused(422, _gate_body("bounds", "a credit has a negative amount", "send the credit as a negative amount"))
if not company_code:
raise _refuse_gate(409, "company", "the bill carries no company", "send the bill's company_code")
if not asset_id and not class_code:
raise _refuse_gate(409, "class", "the order line names neither an asset nor an asset class",
"assign the line to an asset class (or an existing asset) in purchasing")
if not asset_id and not (class_code and description):
raise Refused(422, _gate_body("bounds", "name the asset (asset_id) or the class and description to create one",
"fix the call"))
def unit():
with hub.repository() as repo:
from sqlalchemy import insert, select
claimed = repo.conn.execute(select(T.fa_acquisition_claim.c.transaction_id).where(
T.fa_acquisition_claim.c.acquisition_ref == acquisition_ref)).scalar()
if claimed:
return claimed, True
today = hub.today(repo, True)
when = day(value_date)
if when > today:
raise _refuse_gate(409, "not_future", "an acquisition cannot be dated after today", "post it on the bill's date")
bound_amount(amount, "amount")
if asset_id:
a = repo.asset(asset_id)
if a is None:
raise _refuse_gate(409, "asset_exists", f"no asset {asset_id}", "assign the order line to an asset")
else:
k = _gate_class(repo, class_code, today)
fmt = k["number_format"] or repo.setting("asset_number_format", today)
number = repo.claim_number(class_code, today.year, fmt, acquisition_ref)
a = _new_asset(repo, caller, company_code=company_code, class_code=class_code, description=description,
request_id=acquisition_ref, asset_no=number, sub_no=0, serial_no=None, inventory_no=None,
bp_id=bp_id, quantity=quantity or Decimal("1"), cost_centre=cost_centre, location_id=None,
custodian_email=None, is_car=False, new_and_unused=True, ca_pool=None, areas=None, today=today)
repo.conn.execute(insert(T.fa_asset_request).values(request_id=acquisition_ref, asset_id=a.asset_id,
company_code=company_code))
a.version = repo.save_asset(a)
if a.root["company_code"] != company_code:
raise _refuse_gate(409, "same_company", f"the asset belongs to {a.root['company_code']}, the bill to "
f"{company_code}", "assign the order line to an asset of the bill's company")
if a.root["status"] in ("retired", "deactivated"):
raise _refuse_gate(409, "not_retired", "the asset is fully retired", "assign a new asset, or a new sub-number")
try:
services.check_year_open(repo, company_code, when)
except RuleBroken as closed:
raise _refuse_gate(409, "year_open", str(closed), "post it in an open year") from None
_post_gates(repo, company_code, when, actor)
functional = None
extra: dict = {}
fn_amount = amount
if seams.GL.installed:
try:
functional = seams.company_ledgers(company_code, actor).get("functional_currency")
except seams.Unavailable:
functional = None
if functional and currency != functional:
fx = seams.fx_rate(currency, functional, when, actor)
if fx is None:
raise _refuse_gate(409, "fx_rate", f"gl has no {currency} to {functional} rate on {when.isoformat()}",
"ask the accountant to enter the rate in gl")
fn_amount = amount * Decimal(str(fx["rate"]))
if abs(fn_amount) > MAX_AMOUNT:
raise Refused(422, _gate_body("bounds", f"{amount} {currency} at {fx['rate']} is {fn_amount} {functional}, "
f"above {MAX_AMOUNT}", "send a smaller amount"))
fn_amount = money(fn_amount)
extra["foreign"] = {"currency": currency, "amount": str(amount), "rate": str(fx["rate"])}
deltas = {}
for code in a.area_codes():
v = services.area_values(repo, a.asset_id, code)
if abs(fn_amount) > MAX_AMOUNT or abs(v.cost + fn_amount) > MAX_AMOUNT:
raise Refused(422, _gate_body("bounds", f"the amount in {functional or currency} ({fn_amount}) or the "
f"asset's cost in area {code} would exceed {MAX_AMOUNT}",
"send a smaller amount"))
if v.cost + fn_amount < 0:
raise Refused(422, _gate_body("bounds", f"a credit of {fn_amount} takes cost in area {code} below zero "
f"(cost is {v.cost})", "send at most what the asset cost"))
deltas[code] = Delta(cost=fn_amount)
if not deltas:
raise _refuse_gate(409, "areas", "the asset has no depreciation area", "Settings: the class's areas")
txid = f"acq:{acquisition_ref}"
params = {"offset": {"purpose": "asset_clearing"}, "acquisition_ref": acquisition_ref, "bill_id": bill_id,
"line_no": line_no, "purchase_order_id": purchase_order_id, "currency": currency,
"source_kind": source_kind, "quantity": str(quantity) if quantity is not None else None, **extra}
services.new_tx(repo, caller, transaction_id=txid, asset=a, kind="acquisition", value_date=when, deltas=deltas,
params=params, source_ref=acquisition_ref, asset_version=a.version)
repo.conn.execute(insert(T.fa_acquisition_claim).values(
acquisition_ref=acquisition_ref, bill_id=bill_id, line_no=line_no, asset_id=a.asset_id,
amount=fn_amount, transaction_id=txid, taken_at=datetime.now(timezone.utc)))
return txid, False
with hub.decide():
txid, replay = hub.retrying(unit)
with hub.repository() as repo:
state = repo.transaction(txid)["root"]["status"]
if state != "posted":
services.post_group_now(hub.repository, txid, actor) # a first take, or a retry that resumes the posting
with hub.repository() as repo:
return _acquisition_answer(repo, acquisition_ref, txid, replay)
@FA.use_case(kind="command", route="POST /assets/{asset_id}/acquisitions", permission=hub.ACQ_WRITE, idempotent=True)
def post_manual_acquisition(caller: dict, asset_id: Code, request_id: Code, value_date: IsoDate, amount: Money,
offset_account_id: BigId, tax_area_applies: bool = False, memo: Optional[Note] = None) -> dict:
"""Post an acquisition that did not come through purchasing (FA-07): a laptop paid by card, an asset built in-house, a
government GRANT (a negative amount: it reduces cost in the book and IFRS areas under the deduction method, IAS 20.24, and
the tax area only if `tax_area_applies`). The journal credits the offset account NAMED (never asset clearing: that
belongs to purchasing). The value date is the document's: today or earlier, never later. Below the capitalisation
threshold, a class that is not low-value is refused (expense it, or use class LVA). The same request_id returns
the same transaction."""
actor = services.who(caller)
def unit():
with hub.repository() as repo:
txid = f"man:{asset_id}:{request_id}"
if repo.transaction(txid):
return txid, True
a = _visible(repo, caller, asset_id)
today = hub.today(repo, True)
when = day(value_date)
if when > today:
raise RuleBroken("an acquisition cannot be dated after today (not_future): post it on the day of the document")
bound_amount(amount, "amount")
if amount == 0:
raise RuleBroken("an acquisition is not zero")
clearing = repo.setting("asset_clearing_account_id", today)
if clearing is not None and int(offset_account_id) == int(clearing):
raise RuleBroken("asset clearing is purchasing's: a manual acquisition offsets another account "
"(offset_not_clearing): pick the account the money or credit really came from")
if a.root["status"] in ("retired", "deactivated"):
raise RuleBroken("the asset is retired or deactivated (not_retired): assign a new asset or sub-number")
k = repo.klass(a.root["class_code"], today)
thr = Decimal(str(repo.setting("capitalisation_threshold", today)))
if amount > 0 and amount < thr and not (k and k["is_low_value"]):
raise RuleBroken(f"{amount} is below the capitalisation threshold {thr} for a class that is not LVA "
"(above_threshold): expense it, or use class LVA")
services.check_year_open(repo, a.root["company_code"], when)
_post_gates(repo, a.root["company_code"], when, actor)
grant = amount < 0
deltas = {}
for code in a.area_codes():
area = repo.area(code, today)
if grant and area and area["purpose"] == "tax" and not tax_area_applies:
continue
v = services.area_values(repo, a.asset_id, code)
if v.cost + amount < 0:
raise Refused(422, f"a grant of {amount} takes cost in area {code} below zero (cost is {v.cost})")
deltas[code] = Delta(cost=amount)
services.new_tx(repo, caller, transaction_id=txid, asset=a, kind="grant" if grant else "acquisition",
value_date=when, deltas=deltas, asset_version=a.version, source_ref=f"manual:{request_id}",
params={"offset": {"account_id": int(offset_account_id)}, "memo": memo})
a.version = repo.save_asset(a)
return txid, False
with hub.decide():
txid, replay = hub.retrying(unit)
with hub.repository() as repo:
state = repo.transaction(txid)["root"]["status"]
if state != "posted":
services.post_group_now(hub.repository, txid, actor)
with hub.repository() as repo:
return services.tx_answer(repo, txid) | {"replay": replay}
# ── assets under construction (FA-09) ─────────────────────────────────────────────────────────
def _line_items(repo, auc: Loaded) -> dict:
"""{line_item transaction_id: {area_code: cost}} — what was collected on the AuC (posted acquisitions and grants)."""
out: dict = {}
for r in repo.area_values_rows(auc.asset_id, statuses=("posted",)):
if r["kind"] in ("acquisition", "grant") and (r["cost"] or ZERO) != 0:
out.setdefault(r["transaction_id"], {})[r["area_code"]] = r["cost"]
return out
def _claimed(repo, auc_id: str, line_id: str) -> Decimal:
from sqlalchemy import select
rows = repo.conn.execute(select(T.fa_settlement_claim.c.amount).where(
(T.fa_settlement_claim.c.auc_asset_id == auc_id) & (T.fa_settlement_claim.c.line_item_transaction_id == line_id)))
return sum((r[0] or ZERO for r in rows), ZERO)
def _ref_area(areas: dict) -> str:
return sorted(areas)[0]
def _settlement_row(repo, auc_id: str, when: date, seq: int) -> Optional[dict]:
from sqlalchemy import and_, select
r = repo.conn.execute(select(T.fa_auc_settlement).where(and_(
T.fa_auc_settlement.c.auc_asset_id == auc_id, T.fa_auc_settlement.c.settlement_date == when,
T.fa_auc_settlement.c.seq == seq))).first()
return dict(r._mapping) if r else None
def _settlement_view(row: dict, extra: Optional[dict] = None) -> dict:
return {"auc_asset_id": row["auc_asset_id"], "settlement_date": row["settlement_date"].isoformat(), "seq": row["seq"],
"status": row["status"], "rule": json.loads(row["rule"] or "[]"), "settled_by_email": row["settled_by_email"],
"settled_on": _fmt(row["settled_on"]), **(extra or {})}
def _check_rules(repo, auc: Loaded, rules: list) -> list:
"""Validate a settlement rule against the line items; [(line_id, [(receiver, ref_amount, percent|None)], areas_of_line)]."""
lines = _line_items(repo, auc)
out = []
seen = set()
receivers = repo.assets(r["asset_id"] for rule in rules for r in rule["receivers"]) # every receiver in one round trip
for rule in rules:
lid = rule["line_item_transaction_id"]
if lid not in lines:
raise RuleBroken(f"{lid} is not a line item of {auc.asset_id}")
if lid in seen:
raise RuleBroken(f"line item {lid} appears twice in the rule")
seen.add(lid)
areas = lines[lid]
ref = areas[_ref_area(areas)]
unsettled = ref - _claimed(repo, auc.asset_id, lid)
recv = []
for r in rule["receivers"]:
b = receivers.get(r["asset_id"])
if b is None:
raise RuleBroken(f"no receiving asset {r['asset_id']}")
if b.root["company_code"] != auc.root["company_code"]:
raise RuleBroken(f"receiver {b.asset_id} is of another company: a settlement stays in one company")
if b.root["is_auc"]:
raise RuleBroken("an asset under construction does not settle into another one: name the finished asset")
if b.root["status"] in ("retired", "deactivated"):
raise RuleBroken(f"receiver {b.asset_id} is {b.root['status']}")
recv.append((r["asset_id"], Decimal(str(r["percent"])) if r.get("percent") is not None else None,
Decimal(str(r["amount"])) if r.get("amount") is not None else None))
if len({x[0] for x in recv}) != len(recv):
raise RuleBroken(f"a receiver appears twice for {lid}")
shares = settlement.distribute(ref, unsettled, recv)
out.append((lid, shares, areas, ref))
return out
@FA.use_case(kind="command", route="POST /assets/{auc_asset_id}/settlements", permission=hub.AUC_SETTLE)
def draft_auc_settlement(caller: dict, auc_asset_id: Code, rules: list[LineRuleIn], settlement_date: Optional[IsoDate] = None) -> dict:
"""Draft the rule that settles an asset under construction into the finished assets (FA-09): for each line item (a
bill, own labour) its receivers and their shares by percent (exactly 100) or by amount (never above what is still
unsettled). Nothing posts until settle_auc. The AuC never depreciates (key NONE)."""
with hub.repository() as repo, hub.decide():
auc = _visible(repo, caller, auc_asset_id)
if not auc.root["is_auc"]:
raise RuleBroken(f"{auc_asset_id} is not an asset under construction")
today = hub.today(repo, True)
when = day(settlement_date) if settlement_date else today
if when > today:
raise RuleBroken("a settlement cannot be dated after today")
plain = [r.model_dump(mode="json") for r in rules]
_check_rules(repo, auc, plain)
from sqlalchemy import func, select
seq = (repo.conn.execute(select(func.max(T.fa_auc_settlement.c.seq)).where(
(T.fa_auc_settlement.c.auc_asset_id == auc_asset_id) & (T.fa_auc_settlement.c.settlement_date == when))).scalar()
or 0) + 1
from erp_platform.repository import upsert
row = {"auc_asset_id": auc_asset_id, "settlement_date": when, "seq": seq, "status": "draft",
"rule": json.dumps(plain), "settled_by_email": None, "settled_on": None, "reversed_reason": None}
upsert(repo.conn, T.fa_auc_settlement, [row], ("auc_asset_id", "settlement_date", "seq"))
return _settlement_view(row)
@FA.use_case(kind="command", route="POST /assets/{auc_asset_id}/settlements/{settlement_date}/{seq}/settle",
permission=hub.AUC_SETTLE, idempotent=True)
def settle_auc(caller: dict, auc_asset_id: Code, settlement_date: IsoDate, seq: Seq) -> dict:
"""Settle the drafted rule: each AuC line item is settled ONCE (a claim row per line item and receiver — two
accountants settling the same line at once settle it once), the AuC is credited and each receiver debited per area
with the settlement date as its capitalisation date. Settling more than 100% of a line item, or one already settled, is
refused; a partial settlement leaves the rest on the AuC. A repeat of a settled one only finishes any posting that
failed."""
actor = services.who(caller)
def unit():
with hub.repository() as repo:
from sqlalchemy import insert, select
auc = _visible(repo, caller, auc_asset_id)
when = day(settlement_date)
st = _settlement_row(repo, auc_asset_id, when, seq)
if st is None:
raise Refused(404, f"no settlement {auc_asset_id} {settlement_date} #{seq}")
if st["status"] == "settled":
return [r["transaction_id"] for r in repo.tx_rows(statuses=("approved",)) if
(json.loads(r["params"] or "{}").get("settlement") == f"{auc_asset_id}|{settlement_date}|{seq}")
and r["leg"] == "main"], True
if st["status"] != "draft":
raise RuleBroken(f"this settlement is {st['status']}")
today = hub.today(repo, True)
if when > today:
raise RuleBroken("a settlement cannot be dated after today")
services.check_year_open(repo, auc.root["company_code"], when)
_post_gates(repo, auc.root["company_code"], when, actor)
checked = _check_rules(repo, auc, json.loads(st["rule"]))
per_receiver: dict = {}
for lid, shares, areas, ref in checked:
n = repo.conn.execute(select(T.fa_settlement_line_seq.c.n).where(
(T.fa_settlement_line_seq.c.auc_asset_id == auc_asset_id) &
(T.fa_settlement_line_seq.c.line_item_transaction_id == lid))).all()
repo.conn.execute(insert(T.fa_settlement_line_seq).values(
auc_asset_id=auc_asset_id, line_item_transaction_id=lid, n=len(n) + 1, settlement_seq=seq))
for rid, amt, pct in shares:
repo.conn.execute(insert(T.fa_settlement_claim).values(
auc_asset_id=auc_asset_id, line_item_transaction_id=lid, receiver_asset_id=rid,
settlement_seq=seq, settlement_date=when, percent=pct, amount=amt))
slot = per_receiver.setdefault(rid, {})
for code, line_amt in areas.items():
if code == _ref_area(areas):
part = amt
else:
part = money(line_amt * amt / ref) if ref else ZERO
slot[code] = slot.get(code, ZERO) + part
heads = []
receivers = repo.assets(per_receiver)
for i, (rid, per_area) in enumerate(sorted(per_receiver.items()), start=1):
b = receivers[rid]
head = f"set:{auc_asset_id}:{when.isoformat()}:{seq}:{i}"
prm = {"settlement": f"{auc_asset_id}|{when.isoformat()}|{seq}", "auc_asset_id": auc_asset_id}
services.new_tx(repo, caller, transaction_id=head, asset=b, kind="settlement", value_date=when,
deltas={c: Delta(cost=v) for c, v in per_area.items()}, params=prm,
source_ref=f"settle:{auc_asset_id}:{seq}", asset_version=b.version)
services.new_tx(repo, caller, transaction_id=f"{head}:out", asset=auc, kind="settlement", value_date=when,
deltas={c: Delta(cost=-v) for c, v in per_area.items()}, params=prm, group_id=head,
leg="out", source_ref=f"settle:{auc_asset_id}:{seq}", asset_version=auc.version)
heads.append(head)
from erp_platform.repository import upsert
upsert(repo.conn, T.fa_auc_settlement, [{"auc_asset_id": auc_asset_id, "settlement_date": when, "seq": seq,
"status": "settled", "settled_by_email": services.who(caller),
"settled_on": today}], ("auc_asset_id", "settlement_date", "seq"))
return heads, False
with hub.decide():
heads, replay = hub.retrying(unit)
for head in heads:
with hub.repository() as repo:
state = repo.transaction(head)["root"]["status"]
if state != "posted":
services.post_group_now(hub.repository, head, actor)
with hub.repository() as repo:
st = _settlement_row(repo, auc_asset_id, day(settlement_date), seq)
return _settlement_view(st, {"transactions": heads, "replay": replay})
@FA.use_case(kind="command", route="POST /assets/{auc_asset_id}/settlements/{settlement_date}/{seq}/reverse",
permission=hub.AUC_SETTLE, destructive=True)
def reverse_auc_settlement(caller: dict, auc_asset_id: Code, settlement_date: IsoDate, seq: Seq, reason: Reason) -> dict:
"""Reverse a settlement: reversing transactions dated today put the cost back on the AuC and take it off the receivers,
and the line items are unsettled again. Refused once a receiver has been depreciated, retired or transferred since
(receivers_untouched): retire or transfer from the receiver instead."""
actor = services.who(caller)
def unit():
with hub.repository() as repo:
from sqlalchemy import and_, delete
auc = _visible(repo, caller, auc_asset_id)
when = day(settlement_date)
st = _settlement_row(repo, auc_asset_id, when, seq)
if st is None:
raise Refused(404, f"no settlement {auc_asset_id} {settlement_date} #{seq}")
if st["status"] != "settled":
raise RuleBroken(f"this settlement is {st['status']}: only a settled one is reversed")
ref = f"{auc_asset_id}|{when.isoformat()}|{seq}"
heads = [r for r in repo.tx_rows(statuses=("posted",)) if r["leg"] == "main" and
json.loads(r["params"] or "{}").get("settlement") == ref]
for h in heads:
if repo.postings(h["asset_id"]):
raise RuleBroken(f"receiver {h['asset_id']} has been depreciated since (receivers_untouched): retire or "
"transfer from the receiver instead")
later = [t for t in repo.tx_rows(h["asset_id"], statuses=("proposed", "approved", "posted"))
if t["transaction_id"] != h["transaction_id"]]
if later:
raise RuleBroken(f"receiver {h['asset_id']} has changed since (receivers_untouched): retire or "
"transfer from the receiver instead")
today = hub.today(repo, True)
services.check_year_open(repo, auc.root["company_code"], today)
_post_gates(repo, auc.root["company_code"], today, actor)
revs = [services.reverse_group(repo, caller, h["transaction_id"], today, reason) for h in heads]
repo.conn.execute(delete(T.fa_settlement_claim).where(and_(
T.fa_settlement_claim.c.auc_asset_id == auc_asset_id, T.fa_settlement_claim.c.settlement_date == when,
T.fa_settlement_claim.c.settlement_seq == seq)))
repo.conn.execute(delete(T.fa_settlement_line_seq).where(and_(
T.fa_settlement_line_seq.c.auc_asset_id == auc_asset_id, T.fa_settlement_line_seq.c.settlement_seq == seq)))
from erp_platform.repository import upsert
upsert(repo.conn, T.fa_auc_settlement, [{"auc_asset_id": auc_asset_id, "settlement_date": when, "seq": seq,
"status": "reversed", "reversed_reason": reason}],
("auc_asset_id", "settlement_date", "seq"))
return revs
with hub.decide():
revs = hub.retrying(unit, 1)
for rev in revs:
services.post_group_now(hub.repository, rev, actor)
with hub.repository() as repo:
st = _settlement_row(repo, auc_asset_id, day(settlement_date), seq)
return _settlement_view(st, {"reversing_transactions": revs})
@FA.use_case(kind="query", route="GET /assets/{auc_asset_id}/line-items", permission=hub.READ)
def list_auc_line_items(caller: dict, auc_asset_id: Code) -> dict:
"""The line items collected on an asset under construction, each with what is settled and what is still unsettled."""
with hub.repository() as repo:
auc = _visible(repo, caller, auc_asset_id)
out = []
for lid, areas in _line_items(repo, auc).items():
ref = areas[_ref_area(areas)]
done = _claimed(repo, auc_asset_id, lid)
out.append({"line_item_transaction_id": lid, "amount": str(ref), "settled": str(done),
"unsettled": str(ref - done)})
return {"auc_asset_id": auc_asset_id, "line_items": out}
# ── units of use (FA-11) ──────────────────────────────────────────────────────────────────────
@FA.use_case(kind="command", route="POST /assets/{asset_id}/usage", permission=hub.USAGE)
def record_usage(caller: dict, asset_id: Code, period_id: Period, units: Units, reason: Optional[Reason] = None) -> dict:
"""Record the hours or miles an asset depreciated by use did in a period (zero is a reading). A second reading for the
same period replaces nothing: it is a new row WITH its reason, and the next run posts the difference. Units beyond the
asset's remaining total are refused (within_total). Only for a key that is units of production, and only for an asset
you look after."""
def unit():
with hub.repository() as repo:
from sqlalchemy import func, insert, select
a = _visible(repo, caller, asset_id, own_only=not hub.auth.has_permission(caller, hub.READ))
today = hub.today(repo, True)
if period_id > period_of(today):
raise RuleBroken(f"{period_id} is after the current period {period_of(today)}: record the units in their own period")
uop = []
for r in a.areas:
k = repo.key(r["key_code"], today) if r["key_code"] else None
if k and k["method"] == "units_of_production":
uop.append(r)
if not uop:
raise RuleBroken("the asset's key is not units of production: nothing to record")
existing = repo.usage(asset_id)
total_used = sum((u for p, u in existing.items() if p != period_id), ZERO) + units
for r in uop:
if r["total_units"] is not None and total_used > r["total_units"]:
left = r["total_units"] - sum((u for p, u in existing.items() if p != period_id), ZERO)
raise RuleBroken(f"{units} is more than the asset has left ({left}) of its {r['total_units']} total "
"units (within_total): check the reading, or extend the total units (a dated life change)")
n = repo.conn.execute(select(func.max(T.fa_usage.c.seq)).where(
(T.fa_usage.c.asset_id == asset_id) & (T.fa_usage.c.period_id == period_id))).scalar() or 0
if n and not reason:
raise RuleBroken("a second reading for a period is a correction: say why (reason)")
repo.conn.execute(insert(T.fa_usage).values(asset_id=asset_id, period_id=period_id, seq=n + 1, units=units,
recorded_by_email=services.who(caller), reason=reason,
created_at=datetime.now(timezone.utc),
updated_at=datetime.now(timezone.utc)))
return {"asset_id": asset_id, "period_id": period_id, "seq": n + 1, "units": str(units), "reason": reason,
"correction": bool(n)}
with hub.decide():
return hub.retrying(unit)
# ── the AI step: class, life and pool (FA-05) ─────────────────────────────────────────────────
@FA.use_case(kind="command", route="POST /assets/{asset_id}/suggest-class", permission=hub.AI_SUGGEST)
def ai_suggest_asset_class(caller: dict, asset_id: Code) -> dict:
"""The system suggests the class, the key and life per area and the capital allowance pool of a PLANNED asset from its
description, supplier and amount (suggest mode: it changes nothing). With no provider wired the step leaves a task for
a person; the class defaults are always shown as the baseline. Each field is a row in fa_ai_finding, which the
accountant accepts, corrects or dismisses (accept_asset_suggestion)."""
with hub.repository() as repo, hub.decide():
a = _asset(repo, asset_id)
if a.root["status"] != "planned":
raise RuleBroken("the suggestion is for an asset not yet capitalised")
mode = repo.setting("ai.ai_suggest_asset_class.mode")
today = hub.today(repo, True)
klass = repo.klass(a.root["class_code"], today)
base = {"class_code": a.root["class_code"], "ca_pool": klass["default_ca_pool"] if klass else None,
"areas": [{"area_code": c["area_code"], "key_code": c["key_code"], "life_periods": c["life_periods"]}
for c in repo.class_areas(a.root["class_code"], today)]}
context = {"description": a.root["description"], "bp_id": a.root["bp_id"],
"classes": ",".join(c["class_code"] for c in repo.classes(today)), "current_class": a.root["class_code"]}
answer = ai_steps.ask("ai_suggest_asset_class", mode, f"asset:{asset_id}", a.version,
["class_code", "key_code", "life_periods", "ca_pool"], context)
out = {"asset_id": asset_id, "baseline": base, **answer.as_dict()}
if answer.task is not None:
from erp_platform.repository import upsert
row = answer.task.as_row()
tid = f"{asset_id}:{a.version}"
upsert(repo.conn, T.fa_task, [{"task_id": tid, "step": answer.step, "subject_ref": f"asset:{asset_id}",
"status": "open", "capability": row["capability"],
"input_json": row["input_json"], "plan": row["plan"], "reason": row["reason"],
"destructive": bool(row["destructive"]), "summary": row["summary"],
"created_by": services.who(caller)}], ("task_id",))
out["task_id"] = tid
if answer.kind == "done" and answer.suggestion:
from sqlalchemy import func, select
run_no = (repo.conn.execute(select(func.max(T.fa_ai_finding.c.run_no)).where(
T.fa_ai_finding.c.subject_ref == f"asset:{asset_id}")).scalar() or 0) + 1
import hashlib
digest = hashlib.sha256(json.dumps(context, sort_keys=True).encode()).hexdigest()[:16]
from erp_platform.repository import upsert
for n, (field, value) in enumerate(sorted(answer.suggestion.items()), start=1):
upsert(repo.conn, T.fa_ai_finding, [{
"subject_ref": f"asset:{asset_id}", "step": answer.step, "run_no": run_no, "finding_no": n,
"model": answer.model, "inputs_digest": digest, "field": field, "value": json.dumps(value),
"evidence": answer.rationale, "decision": None, "decided_by_email": None, "decided_at": None}],
("subject_ref", "step", "run_no", "finding_no"))
out["run_no"] = run_no
return out
@FA.use_case(kind="command", route="POST /assets/{asset_id}/suggestions/{run_no}/{finding_no}/decide", permission=hub.WRITE,
idempotent=True)
def accept_asset_suggestion(caller: dict, asset_id: Code, run_no: Seq, finding_no: Seq, decision: Code,
value: Optional[Name] = None) -> dict:
"""Accept, correct (give `value`) or dismiss one suggested field of a planned asset. Each decision is recorded in
fa_ai_finding with who and when; nothing is applied without you. Fields: class_code, ca_pool, key_code:<area>,
life_periods:<area>."""
with hub.repository() as repo, hub.decide():
from sqlalchemy import and_, select
a = _visible(repo, caller, asset_id)
if decision not in ("accepted", "corrected", "dismissed"):
raise RuleBroken("decision is accepted | corrected | dismissed")
f = repo.conn.execute(select(T.fa_ai_finding).where(and_(
T.fa_ai_finding.c.subject_ref == f"asset:{asset_id}", T.fa_ai_finding.c.run_no == run_no,
T.fa_ai_finding.c.finding_no == finding_no))).first()
if f is None:
raise Refused(404, "no such suggestion")
f = dict(f._mapping)
if f["decision"] is not None:
if f["decision"] == decision:
return {"asset_id": asset_id, "field": f["field"], "decision": decision, "applied": False}
raise RuleBroken(f"this suggestion was already {f['decision']}")
applied = None
if decision != "dismissed":
if a.root["status"] != "planned":
raise RuleBroken("the asset is capitalised: a suggestion is no longer applied")
chosen = value if decision == "corrected" else json.loads(f["value"])
if decision == "corrected" and value is None:
raise RuleBroken("give the corrected value")
applied = _apply_field(repo, a, f["field"], chosen)
a.version = repo.save_asset(a)
from erp_platform.repository import upsert
upsert(repo.conn, T.fa_ai_finding, [{"subject_ref": f"asset:{asset_id}", "step": f["step"], "run_no": run_no,
"finding_no": finding_no, "decision": decision,
"decided_by_email": services.who(caller),
"decided_at": datetime.now(timezone.utc)}],
("subject_ref", "step", "run_no", "finding_no"))
return {"asset_id": asset_id, "field": f["field"], "decision": decision, "applied": applied}
def _apply_field(repo, a: Loaded, field: str, value) -> str:
today = hub.today(repo)
if field == "class_code":
k = _gate_class(repo, str(value), today)
a.root["class_code"] = str(value)
a.root["is_auc"] = bool(k["is_auc"])
return f"class_code={value}"
if field == "ca_pool":
if str(value) not in POOLS:
raise RuleBroken(f"ca_pool is one of {', '.join(POOLS)}")
a.root["ca_pool"] = str(value)
return f"ca_pool={value}"
name, _, area = field.partition(":")
if name in ("key_code", "life_periods") and area:
for r in a.areas:
if r["area_code"] == area:
if name == "key_code":
if repo.key(str(value), today) is None:
raise RuleBroken(f"depreciation key {value} is not defined")
r["key_code"] = str(value)
else:
r["life_periods"] = int(value)
return f"{name}[{area}]={value}"
raise RuleBroken(f"unknown field {field}")