#!/usr/bin/env python3 """Offline planning arithmetic. No requests, forecasts, or vendor measurements.""" import argparse import json from decimal import Decimal, DecimalException, ROUND_CEILING from pathlib import Path GATES = ('coverage', 'allowed_use', 'freshness', 'history') def number(value, label, positive=False, integer=False): if value is None: return None if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f'{label}: expected a JSON number or null') n = Decimal(str(value)) if not n.is_finite() or n < 0 or n > Decimal('1e18') or (positive and n == 0) or (integer and n != n.to_integral_value()): raise ValueError(f'{label}: invalid range or noninteger value') return n def mul(*values): if any(v is None for v in values): return None result = Decimal(1) for v in values: result *= v return result def emit(value): if value is None: return None return int(value) if value == value.to_integral_value() else float(value) def require_note(value, label): if not isinstance(value, str) or not value.strip(): raise ValueError(f'{label}: provide a nonempty evidence/assumption note') def evaluate(data): if not isinstance(data, dict): raise ValueError('worksheet must be a JSON object') if type(data.get('schema_version')) is not int or data['schema_version'] != 1: raise ValueError('schema_version must be 1') require_note(data['currency'], 'currency') require_note(data['workload']['description'], 'workload.description') days = number(data['workload']['active_days'], 'active_days', integer=True) hours = number(data['workload']['hours_per_day'], 'hours_per_day') if hours is not None and hours > 24: raise ValueError('hours_per_day must not exceed 24') if not data['scenarios'] or not data['candidates']: raise ValueError('provide at least one scenario and candidate') result = [] for scenario in data['scenarios']: require_note(scenario['name'], 'scenario.name') interval = number(scenario['poll_interval_seconds'], 'poll_interval_seconds', positive=True) seconds = mul(hours, Decimal(3600)) # Poll at t=0, interval, ... strictly before the daily window ends. polls_daily = None if seconds is None or interval is None else (seconds / interval).to_integral_value(rounding=ROUND_CEILING) polls = mul(days, polls_daily) for candidate in data['candidates']: require_note(candidate['name'], 'candidate.name') kind = candidate['kind'] if kind not in ('api', 'collector'): raise ValueError('kind must be api or collector') blockers = [] for gate in GATES: entry = candidate['gates'][gate] if entry['status'] not in ('yes', 'no', 'unknown'): raise ValueError(f'{gate}: expected yes, no or unknown') require_note(entry['note'], gate) if entry['status'] != 'yes': blockers.append(f'{gate}:{entry["status"]}') units = number(candidate['units_per_poll'], 'units_per_poll', positive=True, integer=True) retry = number(candidate['planned_retry_fraction'], 'planned_retry_fraction') require_note(candidate['request_note'], 'request_note') scheduled = mul(polls, units) extra = mul(scheduled, retry) if extra is not None: extra = extra.to_integral_value(rounding=ROUND_CEILING) attempts = None if scheduled is None or extra is None else scheduled + extra credits = wire_bytes = billed_bytes = None if kind == 'api': credits = mul(attempts, number(candidate['credits_per_attempt'], 'credits_per_attempt')) usage_units, usage_basis = credits, 'credits' else: wire_bytes = mul(attempts, number(candidate['wire_bytes_per_attempt'], 'wire_bytes_per_attempt', integer=True)) billed_bytes = mul(attempts, number(scenario['collector_billed_bytes_per_attempt'], 'collector_billed_bytes_per_attempt', integer=True)) usage_units, usage_basis = None if billed_bytes is None else billed_bytes / Decimal(10**9), 'decimal_GB' quota = candidate['quota'] require_note(quota['note'], 'quota.note') if quota['basis'] not in ('requests', 'credits') or (kind == 'collector' and quota['basis'] != 'requests'): raise ValueError('quota basis must be requests, or credits for an API') if not isinstance(quota['unlimited'], bool): raise ValueError('quota.unlimited must be true or false') limit = number(quota['limit'], 'quota.limit') if quota['unlimited'] and limit is not None: raise ValueError('unlimited quota cannot also have a limit') quantity = credits if quota['basis'] == 'credits' else attempts quota_status = 'pass' if quota['unlimited'] else 'unknown' if limit is None or quantity is None else 'exceeded' if quantity > limit else 'pass' if quota_status != 'pass': blockers.append(f'quota:{quota_status}') costs = candidate['costs'] require_note(costs['note'], 'costs.note') fixed = number(costs['fixed_cost'], 'fixed_cost') usage = mul(usage_units, number(costs['usage_price_per_unit'], 'usage_price_per_unit')) labor = mul(number(costs['maintenance_hours'], 'maintenance_hours'), number(costs['hourly_rate'], 'hourly_rate')) components = {'fixed': fixed, 'usage': usage, 'maintenance': labor} missing = [key for key, value in components.items() if value is None] # The supplied quote is not valid above its quota or with unknown quota. total = None if missing or quota_status != 'pass' else sum(components.values()) if scheduled is None or attempts is None: blockers.append('workload:unknown') if missing: blockers.append('costs:unknown') result.append({ 'scenario': scenario['name'], 'candidate': candidate['name'], 'kind': kind, 'polls_per_active_day': emit(polls_daily), 'polls': emit(polls), 'scheduled_requests': emit(scheduled), 'planned_retry_requests': emit(extra), 'total_request_attempts': emit(attempts), 'api_credits': emit(credits), 'wire_bytes': emit(wire_bytes), 'billed_bytes': emit(billed_bytes), 'billed_decimal_GB': emit(None if billed_bytes is None else billed_bytes / Decimal(10**9)), 'billed_GiB': emit(None if billed_bytes is None else billed_bytes / Decimal(2**30)), 'usage_basis': usage_basis, 'quota_status': quota_status, 'cost_components': {k: emit(v) for k, v in components.items()}, 'missing_cost_components': missing, 'modeled_total': emit(total), 'decision': 'hold' if blockers else 'eligible_for_source_trial', 'blockers': blockers, }) return {'schema_version': 1, 'currency': data['currency'], 'results': result} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('input', type=Path) args = parser.parse_args() try: output = evaluate(json.loads(args.input.read_text())) rendered = json.dumps(output, indent=2, sort_keys=True, allow_nan=False) except (ValueError, KeyError, TypeError, OSError, OverflowError, DecimalException) as error: parser.exit(2, f'Invalid worksheet: {error}\n') print(rendered) if __name__ == '__main__': main()