dodil/erp-brick-pos

DODIL ERP brick: point of sale

src/brick_pos/domain/snapshot.py

"""The register snapshot (POS-30): the catalogue, the prices and the tax rates a register sells with, read ONCE when a session opens and
kept as a dated copy of product's and tax's answers. A sale is rung from it, so a till that loses the network keeps selling, and a price or
rate dated tomorrow is in tomorrow's snapshot, never today's. It is versioned per (shop, business date) and carries a digest of its
content: two registers that opened on the same answers share one snapshot, a change between two openings is the next snapshot_no, and the
digest is what the Z report prints. A snapshot is never edited."""
import hashlib
import json
from dataclasses import dataclass, field
from datetime import date, datetime
from decimal import Decimal
from typing import Optional

from brick_pos.domain.pricing import Price, Rate


@dataclass
class Snapshot:
    shop_id: str
    business_date: date
    price_list_id: str = ""
    country: str = ""
    currency: str = ""
    prices_include_tax: bool = True
    valid_until: Optional[date] = None             # the earliest day a price or a rate changes: the snapshot is refreshed before it
    products: list = field(default_factory=list)   # one row per price: product_id, unit_code, from_quantity, unit_price, pricing_quantity, name, tax_category
    rates: list = field(default_factory=list)      # category_code, supply, consumption, tax_code, rate_pct
    snapshot_no: int = 0
    digest: str = ""
    built_at: Optional[datetime] = None

    def content(self) -> dict:
        return {"business_date": self.business_date.isoformat(), "price_list_id": self.price_list_id, "country": self.country,
                "currency": self.currency, "prices_include_tax": self.prices_include_tax,
                "valid_until": self.valid_until.isoformat() if self.valid_until else None,
                "products": sorted(self.products, key=lambda p: (p["product_id"], p["unit_code"], Decimal(p["from_quantity"]))),
                "rates": sorted(self.rates, key=lambda r: (r["category_code"], r["supply"], r["consumption"], r["tax_code"]))}

    def seal(self) -> "Snapshot":
        self.digest = hashlib.sha256(json.dumps(self.content(), sort_keys=True, separators=(",", ":")).encode()).hexdigest()
        return self

    def prices(self) -> list:
        return [Price(product_id=p["product_id"], unit_code=p["unit_code"], from_quantity=Decimal(p["from_quantity"]),
                      unit_price=Decimal(p["unit_price"]), pricing_quantity=int(p["pricing_quantity"])) for p in self.products]

    def rate_rows(self) -> list:
        return [Rate(category_code=r["category_code"], supply=r["supply"], consumption=r["consumption"], tax_code=r["tax_code"],
                     rate_pct=Decimal(r["rate_pct"])) for r in self.rates]

    def info(self) -> dict:
        """product_id -> {tax_category, name}: what pricing needs of the item master."""
        return {p["product_id"]: {"tax_category": p["tax_category"], "name": p["name"]} for p in self.products}


def build(*, shop_id: str, business_date: date, price_list_id: str, country: str, currency: str, prices_include_tax: bool,
          valid_until: Optional[date], prices: list, info: dict, rates: list) -> Snapshot:
    """The snapshot of what product and tax answered: only the products the item master knows and gives a tax category (a price for a
    product nobody can tax is not sellable), each price row with the product's name and tax category."""
    rows = [{"product_id": p.product_id, "unit_code": p.unit_code, "from_quantity": format(p.from_quantity, "f"),
             "unit_price": format(p.unit_price, "f"), "pricing_quantity": p.pricing_quantity, "name": info[p.product_id]["name"],
             "tax_category": info[p.product_id]["tax_category"]} for p in prices if p.product_id in info]
    return Snapshot(shop_id=shop_id, business_date=business_date, price_list_id=price_list_id, country=country, currency=currency,
                    prices_include_tax=prices_include_tax, valid_until=valid_until, products=rows,
                    rates=[{"category_code": r.category_code, "supply": r.supply, "consumption": r.consumption, "tax_code": r.tax_code,
                            "rate_pct": format(r.rate_pct, "f")} for r in rates]).seal()