#!/usr/bin/env python3 """Reproduce the article's illustrative SWDM v4 arithmetic. This is a small independent implementation of the documented formula, not CO2.js, a browser collector, a power meter, or an SCI conformance assessment. No network access, user telemetry or production observations are used. """ from __future__ import annotations import argparse from decimal import Decimal, InvalidOperation import hashlib import json from pathlib import Path from typing import Any D = Decimal OP = {'data_centre': D('0.055'), 'network': D('0.059'), 'device': D('0.080')} EM = {'data_centre': D('0.012'), 'network': D('0.013'), 'device': D('0.081')} EM_GRID = D('494') SOURCE_BLOBS = {'formula':'e38b0e09851dd5f64247d5205013c3abba47b269', 'constants':'851fa28b3af660436b494f90ed2b5527bc86da41'} def dec(value: Any, name: str, upper: str | None = None) -> Decimal: if value is None or isinstance(value, bool): raise ValueError(f'{name}: an explicit finite number is required; unknown is not zero') try: result = D(str(value)) except (InvalidOperation, ValueError) as exc: raise ValueError(f'{name}: invalid number') from exc if not result.is_finite() or result < 0 or (upper is not None and result > D(upper)): raise ValueError(f'{name}: value outside the permitted range') return result def canonical_hash(obj: Any) -> str: text = json.dumps(obj, ensure_ascii=False, sort_keys=True, separators=(',', ':')) return 'sha256:' + hashlib.sha256(text.encode('utf-8')).hexdigest() def estimate(byte_count: Any, *, first_share: Any='1', return_share: Any='0', reload_share: Any='0', hosting_factor: Any='0', dc_grid: Any='494', network_grid: Any='494', device_grid: Any='494') -> dict: """Inputs are cold first-view bytes plus optional modelled visit weighting. Do not pass already weighted/warm bytes and then apply the same cache saving again. Operational grids are scenario inputs; embodied conversion stays 494. """ b = dec(byte_count, 'bytes') if b != b.to_integral_value(): raise ValueError('bytes must be an integer') first, ret, reload, host = [dec(v,n,'1') for v,n in [ (first_share,'first_share'),(return_share,'return_share'), (reload_share,'reload_share'),(hosting_factor,'hosting_factor')]] if first + ret != 1: raise ValueError('first_share + return_share must equal 1') grids = {k:dec(v,k+'_grid') for k,v in [('data_centre',dc_grid),('network',network_grid),('device',device_grid)]} factor = first + ret * reload gb = b / D('1000000000') raw, adjusted = {}, {} for segment in OP: op = gb * OP[segment] * grids[segment] em = gb * EM[segment] * EM_GRID raw[segment+'_operational'] = op raw[segment+'_embodied'] = em adjusted[segment+'_operational'] = op * (1-host if segment=='data_centre' else 1) * factor adjusted[segment+'_embodied'] = em * factor total = sum(adjusted.values(), D(0)) inputs = {'cold_first_view_bytes':int(b), 'first_share':str(first), 'return_share':str(ret), 'reload_share':str(reload), 'hosting_factor':str(host), 'operational_grid_gco2e_per_kwh':{k:str(v) for k,v in grids.items()}, 'embodied_conversion_gco2e_per_kwh':str(EM_GRID)} result = {'unit':'gCO2e/modelled_view', 'total':str(total), 'display_six_decimals':format(total,'.6f'), 'raw_segments_gco2e':{k:str(v) for k,v in raw.items()}, 'adjusted_segments_gco2e':{k:str(v) for k,v in adjusted.items()}, 'visit_factor':str(factor)} fingerprint = { 'method':'SWDM', 'model_version':'4', 'implementation':'article-arithmetic-1.0.0', 'source_git_blobs':SOURCE_BLOBS, 'inputs':inputs, 'boundary':'modelled data centre, network, device; operational + embodied-equivalent allocation', 'input_evidence_kind':'illustrative, not measured', 'functional_unit':'modelled view under declared cold/return assumptions', 'coefficient_unit':'kWh-equivalent/decimal GB for embodied; kWh/decimal GB for operational', 'operational_coefficients':{k:str(v) for k,v in OP.items()}, 'embodied_coefficients':{k:str(v) for k,v in EM.items()}, 'uncertainty':'not quantified; scenario sensitivity is not a confidence interval', 'exclusions':['no device-specific power observation','no production backend or AI telemetry', 'no causal avoided-emissions claim','no comprehensive lifecycle assessment'], 'rounding':'full Decimal calculation; six decimals only for regression display', 'output_hash':canonical_hash(result), } fingerprint['fingerprint_id']=canonical_hash(fingerprint) return {'inputs':inputs,'result':result,'fingerprint':fingerprint} def examples() -> dict: cases = [ ('one_mb_cold', dict(byte_count=1000000), '0.148200'), ('one_mb_hosting_adjustment',dict(byte_count=1000000,hosting_factor='1'),'0.121030'), ('one_mb_mixed_visits',dict(byte_count=1000000,first_share='.5',return_share='.5',reload_share='.2'),'0.088920'), ('one_mb_hosting_and_mixed_visits',dict(byte_count=1000000,hosting_factor='1',first_share='.5',return_share='.5',reload_share='.2'),'0.072618'), ('two_mb_cold',dict(byte_count=2000000),'0.296400'), ('one_mib_cold',dict(byte_count=1048576),'0.155399'), ('one_mb_lower_dc_grid_scenario',dict(byte_count=1000000,dc_grid='100'),'0.126530'), ('zero_bytes_model_only',dict(byte_count=0),'0.000000'), ] rows=[] for name,args,expected in cases: r=estimate(**args) assert r['result']['display_six_decimals']==expected,(name,r) assert sum(D(v) for v in r['result']['adjusted_segments_gco2e'].values())==D(r['result']['total']) rows.append({'id':name,'expected_display':expected,**r}) rate=D(rows[0]['result']['total']) before=rate*2*100000/1000; after=rate*250000/1000 return {'schema':'cometweb.article-calculation-examples/v1','as_of':'2026-09-13', 'evidence_kind':'illustrative arithmetic, not a site benchmark', 'cases':rows, 'traffic_example':{'before_bytes':2000000,'after_bytes':1000000, 'before_modelled_views':100000,'after_modelled_views':250000, 'before_kgco2e':str(before),'after_kgco2e':str(after), 'intensity_change_percent':'-50','total_change_percent':str((after/before-1)*100)}, 'hosting_example':{'relative_model_reduction_percent':str((D('0.1482')-D('0.12103'))/D('0.1482')*100), 'interpretation':'A model adjustment, not an empirical measure of supplier effectiveness.'}} def test_invalid_inputs() -> list: invalid=[{'byte_count':None},{'byte_count':True},{'byte_count':-1}, {'byte_count':'nan'},{'byte_count':'Infinity'},{'byte_count':1.2}, {'byte_count':1,'first_share':'.4','return_share':'.4'}, {'byte_count':1,'reload_share':'1.1'}, {'byte_count':1,'hosting_factor':'-1'}, {'byte_count':1,'dc_grid':None},{'byte_count':1,'network_grid':'-2'}, {'byte_count':1,'device_grid':'nan'}] results=[] for args in invalid: try: estimate(**args) except ValueError: results.append({'inputs':args,'rejected':True}) else: raise AssertionError(f'Invalid input accepted: {args}') return results def main() -> None: parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--output',type=Path) parser.add_argument('--tests-output',type=Path) args=parser.parse_args() results=examples(); negatives=test_invalid_inputs() text=json.dumps(results,indent=2,ensure_ascii=False)+'\n' if args.output: args.output.parent.mkdir(parents=True,exist_ok=True);args.output.write_text(text,encoding='utf-8') else: print(text) if args.tests_output: args.tests_output.write_text(json.dumps({'positive_cases':len(results['cases']), 'invalid_input_cases':len(negatives),'status':'PASS','negative_controls':negatives, 'scope':'formula and input validation only; not npm package, product, browser energy or causal validation'},indent=2)+'\n') if __name__=='__main__': main()