Spaces:
Sleeping
Sleeping
File size: 3,509 Bytes
1bc4cbd 18c0a20 1bc4cbd | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 | """Bounded, disposable numerical cache; never an authority for publication.
Only application-selected numerical projections enter this cache. Current
publication and recovery checks must happen outside it on every request.
"""
from collections import OrderedDict
from concurrent.futures import Future
import json
from threading import Lock
from .results import canonical_hash
from .storage import encode
VERSION = 'koevd-public-numerical-profile-v2'
FIELDS = ('format', 'artifact_sha256', 'protocol', 'scoring', 'statistics',
'interpretation', 'tasks', 'task3_task5', 'quality', 'signal_thresholds',
'signals', 'charts')
def numerical_profile(profile):
"""Whitelist analytical output; omit originals, cases and source identities."""
value = {key: profile[key] for key in FIELDS}
disagreement = profile['measurement_disagreement']
value['measurement_disagreement'] = {
key: disagreement[key] for key in ('count', 'comparable_sources', 'interpretation')}
# Round trip also detaches every nested object from the analysis engine.
return json.loads(encode(value))
def numerical_comparison(comparison):
"""Exact matched-input counts and rates without run or sample identities."""
value = {key: comparison[key] for key in
('format', 'left_artifact', 'right_artifact', 'tasks', 'note')}
value = json.loads(encode(value))
value['note'] = (comparison['note'].replace('Intervals resample shared sources together.',
'Bootstrap is disabled in this public summary; no confidence intervals are shown.'))
for task in value['tasks']:
for partition in task['partitions']:
partition.pop('sample_ids', None)
return value
class NumericalProfileCache:
def __init__(self, *, max_entries=128, max_bytes=64 * 1024**2):
if (type(max_entries) is not int or max_entries < 1
or type(max_bytes) is not int or max_bytes < 1):
raise ValueError('Positive numerical cache bounds required')
self.max_entries, self.max_bytes = max_entries, max_bytes
self._entries, self._pending = OrderedDict(), {}
self._size, self._lock = 0, Lock()
def get_or_compute(self, identity, compute):
key = canonical_hash({'projection': VERSION, **identity})
with self._lock:
if key in self._entries:
self._entries.move_to_end(key)
return json.loads(self._entries[key])
future = self._pending.get(key)
leader = future is None
if leader:
future = self._pending[key] = Future()
if not leader:
return json.loads(future.result())
try:
blob = encode(compute())
with self._lock:
if len(blob) <= self.max_bytes:
while self._entries and (len(self._entries) >= self.max_entries
or self._size + len(blob) > self.max_bytes):
_, removed = self._entries.popitem(last=False)
self._size -= len(removed)
self._entries[key] = blob
self._size += len(blob)
self._pending.pop(key)
future.set_result(blob)
return json.loads(blob)
except BaseException as exc:
with self._lock:
self._pending.pop(key, None)
future.set_exception(exc)
raise
|