""" Registre de catégorie (pool fiscal) — couche au-dessus de l'inventaire. L'inventaire physique (assets : parents/pièces) et le pool fiscal sont SÉPARÉS. Le pool agrège des flux financiers par année et par catégorie DPA, indépendamment du serveur physique auquel une dépense est rattachée. Table `cca_pool_years` : - year, cca_class_id : clé (une ligne par année et catégorie) - additions_override : ajouts saisis manuellement (NULL = dérivé) - dispositions : produit des dispositions de l'année - adjustment_coeff : coeff 0..1 sur ajouts nets (0.5 = demi-année) - pool_physically_empty : 1 si plus aucun actif physique (perte finale) - opening_ucc_override : FNACC d'ouverture initiale (1re année) Les AJOUTS d'une année peuvent être dérivés automatiquement de l'inventaire (somme des coûts des acquisitions dont l'année d'acquisition = year) ou saisis manuellement (override). Les dispositions et le coefficient sont toujours saisis par l'usager. Références législatives : - LIR 13(21) — Définition de la FNACC ; le pool est calculé par catégorie, non par actif individuel - Règl. 1100(1) — Déduction admissible : taux dégressif appliqué au solde du pool en fin d'exercice (méthode du solde dégressif) - Règl. Partie XI — Art. 1100–1105 : règles générales des catégories de biens - Guide T2 (T4012) — Annexe 8 / Schedule 8 : présentation du registre DPA Textes officiels (Justice Canada) : https://laws-lois.justice.gc.ca/fra/lois/I-3.3/section-13.html https://laws-lois.justice.gc.ca/fra/reglements/C.R.C.,_ch._945/section-1100.html https://www.canada.ca/fr/agence-revenu/services/formulaires-publications/publications/t4012.html """ from __future__ import annotations import sqlite3 from decimal import Decimal from threading import Lock from .cca_pool import YearInput, compute_pool_timeline class CcaRegistry: def __init__(self, conn: sqlite3.Connection, store, lock: Lock) -> None: self._conn = conn self._store = store self._lock = lock self._init_schema() def _init_schema(self) -> None: self._conn.execute(""" CREATE TABLE IF NOT EXISTS cca_pool_years ( year INTEGER NOT NULL, cca_class_id INTEGER NOT NULL, additions_override REAL, dispositions REAL NOT NULL DEFAULT 0, adjustment_coeff REAL NOT NULL DEFAULT 0.5, pool_physically_empty INTEGER NOT NULL DEFAULT 0, opening_ucc_override REAL, PRIMARY KEY (year, cca_class_id) ) """) self._conn.commit() # ---- Dérivation des ajouts depuis l'inventaire ---- def _derived_additions(self, year: int, cca_class_id: int) -> Decimal: """Somme des coûts des acquisitions de l'année pour cette catégorie. On compte : le coût des PIÈCES dont acquisition_date tombe dans l'année (qu'elles soient rattachées ou libres) + les PARENTS sans pièce (coût propre). Un parent avec pièces n'est pas compté en double : ses pièces portent déjà le coût.""" total = Decimal("0") for a in self._store.all(): cca = a["view_cpa"]["cca_class"]["id"] if cca != cca_class_id: continue acq = (a["view_cpa"].get("acquisition_date") or "")[:4] if acq != str(year): continue if a["type"] == "piece": total += Decimal(str(a["view_cpa"]["capitalizable_base_cad"])) elif a["type"] == "parent": # Parent sans pièce : on prend son coût propre (sinon 0, ses # pièces sont comptées séparément). if a["view_admin"].get("pieces_count", 0) == 0: total += Decimal(str(a["view_cpa"]["capitalizable_base_cad"])) return total def _physical_count(self, cca_class_id: int) -> int: """Nombre d'actifs physiques actifs dans la catégorie (parents + pièces non au rebut).""" n = 0 for a in self._store.all(): if a["view_cpa"]["cca_class"]["id"] != cca_class_id: continue if a["view_admin"].get("operational_status") == "Rebut": continue n += 1 return n # ---- CRUD années ---- def upsert_year(self, year: int, cca_class_id: int = 50, **kw) -> None: with self._lock: existing = self._conn.execute( "SELECT 1 FROM cca_pool_years WHERE year=? AND cca_class_id=?", (year, cca_class_id)).fetchone() fields = { "additions_override": kw.get("additions_override"), "dispositions": float(kw.get("dispositions", 0) or 0), "adjustment_coeff": float(kw.get("adjustment_coeff", 0.5) or 0.5), "pool_physically_empty": 1 if kw.get("pool_physically_empty") else 0, "opening_ucc_override": kw.get("opening_ucc_override"), } if existing: self._conn.execute(""" UPDATE cca_pool_years SET additions_override=?, dispositions=?, adjustment_coeff=?, pool_physically_empty=?, opening_ucc_override=? WHERE year=? AND cca_class_id=?""", (fields["additions_override"], fields["dispositions"], fields["adjustment_coeff"], fields["pool_physically_empty"], fields["opening_ucc_override"], year, cca_class_id)) else: self._conn.execute(""" INSERT INTO cca_pool_years (year, cca_class_id, additions_override, dispositions, adjustment_coeff, pool_physically_empty, opening_ucc_override) VALUES (?,?,?,?,?,?,?)""", (year, cca_class_id, fields["additions_override"], fields["dispositions"], fields["adjustment_coeff"], fields["pool_physically_empty"], fields["opening_ucc_override"])) self._conn.commit() def delete_year(self, year: int, cca_class_id: int = 50) -> None: with self._lock: self._conn.execute("DELETE FROM cca_pool_years WHERE year=? AND cca_class_id=?", (year, cca_class_id)) self._conn.commit() def _year_rows(self, cca_class_id: int) -> list[sqlite3.Row]: cur = self._conn.execute( "SELECT * FROM cca_pool_years WHERE cca_class_id=? ORDER BY year", (cca_class_id,)) return cur.fetchall() # ---- Calcul du registre ---- def compute(self, cca_class_id: int = 50, cca_rate: Decimal = Decimal("0.55")) -> dict: """Calcule la timeline du pool pour une catégorie. Les ajouts sont dérivés de l'inventaire sauf si un override est saisi.""" rows = self._year_rows(cca_class_id) # Si aucune année déclarée, on déduit les années présentes dans l'inventaire. if not rows: years_seen = set() for a in self._store.all(): if a["view_cpa"]["cca_class"]["id"] == cca_class_id: acq = (a["view_cpa"].get("acquisition_date") or "")[:4] if acq.isdigit(): years_seen.add(int(acq)) for y in years_seen: self.upsert_year(y, cca_class_id) rows = self._year_rows(cca_class_id) opening = Decimal("0") year_inputs = [] for r in rows: add = (Decimal(str(r["additions_override"])) if r["additions_override"] is not None else self._derived_additions(r["year"], cca_class_id)) if r["opening_ucc_override"] is not None and not year_inputs: opening = Decimal(str(r["opening_ucc_override"])) year_inputs.append(YearInput( year=r["year"], additions=add, dispositions=Decimal(str(r["dispositions"])), adjustment_coeff=Decimal(str(r["adjustment_coeff"])), pool_physically_empty=bool(r["pool_physically_empty"]), )) results = compute_pool_timeline(year_inputs, opening_ucc=opening, cca_rate=cca_rate) return { "cca_class_id": cca_class_id, "cca_rate": str(cca_rate), "physical_count": self._physical_count(cca_class_id), "years": [r.to_dict() for r in results], }