Daniel Allaire
691537400f
Remplace les mentions textuelles par des URLs directes dans les docstrings de cca_pool.py, cca_registry.py et le commentaire CCA_CATEGORIES dans store.py. - laws-lois.justice.gc.ca : LIR art. 13, 20 ; Règl. art. 1100 - canada.ca : Guide T2 (T4012) pour l'Annexe 8 / Schedule 8 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
184 lines
8.6 KiB
Python
184 lines
8.6 KiB
Python
"""
|
||
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],
|
||
}
|