272 lines
14 KiB
Python
272 lines
14 KiB
Python
"""Repeated, traffic-bounded HTTPS throughput measurements; not line capacity."""
|
||
from dataclasses import asdict, dataclass
|
||
import math
|
||
import os
|
||
import re
|
||
import statistics
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
from .net import opener, read_limited
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Source:
|
||
name: str
|
||
region: str
|
||
download_url: str
|
||
upload_url: str | None = None
|
||
use_range: bool = False
|
||
capped_stream: bool = False
|
||
carrier: str | None = None
|
||
mainland: bool | None = None
|
||
|
||
|
||
# Fixed official test endpoints only. No arbitrary URLs, account keys or binaries.
|
||
SOURCES = {
|
||
'cloudflare': Source('Cloudflare','Anycast','https://speed.cloudflare.com/__down',
|
||
'https://speed.cloudflare.com/__up'),
|
||
'hetzner-fsn': Source('Hetzner','DE / Falkenstein','https://fsn1-speed.hetzner.com/100MB.bin',use_range=True),
|
||
'ovh-rbx': Source('OVHcloud','FR / Roubaix','https://lil1-rbx.speedtest.network.ovh.net/files/100Mb.dat',use_range=True),
|
||
'cn-telecom-sh': Source('China Telecom','CN / Shanghai','https://speedtest1.online.sh.cn:8080/speedtest/random4000x4000.jpg',
|
||
capped_stream=True,carrier='telecom',mainland=True),
|
||
'cn-unicom-sh': Source('China Unicom','CN / Shanghai','https://mobile.shunicomtest.com:8080/speedtest/random4000x4000.jpg',
|
||
capped_stream=True,carrier='unicom',mainland=True),
|
||
'cn-mobile-fz': Source('China Mobile','CN / Fuzhou','https://csfw.fj.chinamobile.com:8080/speedtest/random4000x4000.jpg',
|
||
capped_stream=True,carrier='mobile',mainland=True),
|
||
'mobile-hk': Source('China Mobile Hong Kong','HK / not mainland China','https://speedtestbb.hk.chinamobile.com:8080/speedtest/random4000x4000.jpg',
|
||
capped_stream=True,carrier='mobile',mainland=False),
|
||
}
|
||
CHINA_SOURCES=('cn-telecom-sh','cn-unicom-sh','cn-mobile-fz')
|
||
# Keep the slower, environment-sensitive Hetzner endpoint explicitly selectable.
|
||
DEFAULT_SOURCES = ('cloudflare','ovh-rbx')
|
||
|
||
|
||
def source_ids(selected=None):
|
||
selected = DEFAULT_SOURCES if selected is None else tuple(dict.fromkeys(selected))
|
||
if not selected or any(identifier not in SOURCES for identifier in selected):
|
||
raise ValueError('Invalid bandwidth source')
|
||
return selected
|
||
|
||
|
||
def payload_budget(rounds, sample_mib, selected=None):
|
||
return sum((2 if SOURCES[name].upload_url else 1) * (rounds * sample_mib * 1024**2 + 131072)
|
||
for name in source_ids(selected))
|
||
|
||
|
||
@dataclass
|
||
class Sample:
|
||
direction: str
|
||
size_bytes: int
|
||
elapsed_s: float = 0
|
||
mbps: float | None = None
|
||
error: str | None = None
|
||
pop: str | None = None
|
||
source: str = 'cloudflare'
|
||
|
||
|
||
def transfer(direction, size, source='cloudflare'):
|
||
from .worker import enabled,invoke
|
||
if enabled():
|
||
result=invoke('transfer',{'direction':direction,'size':size,'source':source},timeout=25)
|
||
return Sample(**result) if isinstance(result,dict) else Sample(direction,size,error='request_failed',source=source)
|
||
if direction not in ('download', 'upload') or not 1 <= size <= 20 * 1024**2:
|
||
raise ValueError('Invalid transfer limits')
|
||
if source not in SOURCES:
|
||
raise ValueError('Invalid bandwidth source')
|
||
provider = SOURCES[source]
|
||
result = Sample(direction, size, source=source)
|
||
if direction == 'upload' and provider.upload_url is None:
|
||
result.error = 'unsupported'
|
||
return result
|
||
headers = {'User-Agent': 'VPSMaster/1.0.0b1', 'Accept-Encoding': 'identity',
|
||
'Cache-Control': 'no-cache'}
|
||
url = provider.download_url
|
||
if not provider.use_range and not provider.capped_stream:
|
||
url += '?bytes=' + str(size)
|
||
elif provider.use_range:
|
||
headers['Range'] = f'bytes=0-{size-1}'
|
||
data = None
|
||
if direction == 'upload':
|
||
url = provider.upload_url
|
||
headers.pop('Range',None)
|
||
data = os.urandom(size) # Avoid compressible uploads artificially inflating results.
|
||
headers['Content-Type'] = 'application/octet-stream'
|
||
request = urllib.request.Request(url, data=data, headers=headers)
|
||
start = time.monotonic()
|
||
try:
|
||
with opener().open(request, timeout=12) as response:
|
||
ranged = direction == 'download' and provider.use_range
|
||
if ranged and response.status == 200:
|
||
# Never fall back to downloading an entire 100 MB test file.
|
||
result.error = 'range_not_supported'
|
||
return result
|
||
if response.status != (206 if ranged else 200):
|
||
result.error = 'http_' + str(response.status)
|
||
return result
|
||
if ranged:
|
||
match = re.fullmatch(r'bytes 0-(\d+)/(\d+)',response.headers.get('Content-Range',''))
|
||
length = response.headers.get('Content-Length')
|
||
if (not match or int(match[1]) != size-1 or int(match[2]) < size
|
||
or (length is not None and length != str(size))):
|
||
result.error = 'invalid_content_range'
|
||
return result
|
||
if ('text/html' in response.headers.get('Content-Type','').lower()
|
||
or response.headers.get('Content-Encoding','identity').lower() not in ('','identity')):
|
||
result.error = 'unexpected_content'
|
||
return result
|
||
ray = response.headers.get('CF-Ray', '')
|
||
pop = ray.rsplit('-', 1)[-1]
|
||
if len(pop) == 3 and pop.isalpha():
|
||
result.pop = pop.upper()
|
||
if direction=='download' and provider.capped_stream:
|
||
# Fixed public test images often ignore Range. Read exactly the
|
||
# consented payload, then close; never consume the full object.
|
||
if not response.headers.get('Content-Type','').lower().startswith('image/jpeg'):
|
||
result.error='unexpected_test_content';return result
|
||
chunks=[];remaining=size
|
||
while remaining:
|
||
if time.monotonic()-start>20:raise TimeoutError()
|
||
read=getattr(response,'read1',response.read)
|
||
block=read(min(65536,remaining))
|
||
if not block:break
|
||
chunks.append(block);remaining-=len(block)
|
||
body=b''.join(chunks)
|
||
if not body.startswith(b'\xff\xd8'):
|
||
result.error='unexpected_test_content';return result
|
||
else:
|
||
body = read_limited(response, size if direction == 'download' else 65536, start + 30)
|
||
if direction == 'download' and len(body) != size:
|
||
result.error = 'short_response'
|
||
return result
|
||
result.elapsed_s = time.monotonic() - start
|
||
if result.elapsed_s <= 0 or not math.isfinite(result.elapsed_s):
|
||
result.error = 'invalid_duration'
|
||
else:
|
||
result.mbps = round(size * 8 / result.elapsed_s / 1_000_000, 3)
|
||
except urllib.error.HTTPError as exc:
|
||
result.error = 'http_' + str(exc.code)
|
||
exc.close()
|
||
except ValueError:
|
||
result.error = 'response_too_large'
|
||
except (OSError, urllib.error.URLError):
|
||
result.error = 'request_failed'
|
||
finally:
|
||
result.elapsed_s = round(max(0, time.monotonic() - start), 4)
|
||
return result
|
||
|
||
|
||
def summarize(samples):
|
||
good = [s for s in samples if s.error is None and s.mbps is not None and math.isfinite(s.mbps) and s.mbps > 0]
|
||
speeds = [s.mbps for s in good]
|
||
median = statistics.median(speeds) if speeds else None
|
||
spread = (max(speeds) - min(speeds)) / median if median else None
|
||
enough = len(good) >= 3 and len(good) == len(samples)
|
||
return {'successful': len(good), 'attempted': len(samples),
|
||
'median_mbps': round(median, 3) if median is not None else None,
|
||
'min_mbps': min(speeds) if speeds else None, 'max_mbps': max(speeds) if speeds else None,
|
||
'spread_percent': round(spread * 100, 1) if spread is not None else None,
|
||
'quality': 'unavailable' if not speeds else 'insufficient' if not enough else
|
||
'short_samples' if any(s.elapsed_s < 1 for s in good) else
|
||
'variable' if spread > .3 else 'consistent',
|
||
'samples': [asdict(s) for s in samples]}
|
||
|
||
|
||
def measure(rounds=3, sample_mib=4, on_sample=None, source='cloudflare'):
|
||
if not 3 <= rounds <= 8 or not 1 <= sample_mib <= 20:
|
||
raise ValueError('Invalid benchmark limits')
|
||
source_ids((source,))
|
||
provider = SOURCES[source]
|
||
report = {'source':source,'provider':provider.name,'region':provider.region,
|
||
'carrier_claim':provider.carrier,'mainland_china':provider.mainland,
|
||
'provider_identity':'catalogue_claim' if provider.carrier else 'official_test_endpoint',
|
||
'directions_scope':'download: node_to_this_host; upload: this_host_to_node',
|
||
'download_url':provider.download_url,'upload_supported':provider.upload_url is not None,
|
||
'method': 'sequential_https_wall_time',
|
||
'payload_budget_bytes': payload_budget(rounds,sample_mib,(source,)),
|
||
'warmups': {}, 'directions': {}}
|
||
def sample_transfer(direction,size):
|
||
# Preserve the original single-source Python API.
|
||
return transfer(direction,size) if source=='cloudflare' else transfer(direction,size,source=source)
|
||
for direction in (('download','upload') if provider.upload_url else ('download',)):
|
||
warmup = sample_transfer(direction, 131072)
|
||
report['warmups'][direction] = asdict(warmup)
|
||
samples = []
|
||
for number in range(1, rounds + 1):
|
||
# Do not hammer an endpoint that explicitly refuses or rate-limits us.
|
||
previous = samples[-1] if samples else warmup
|
||
if previous.error in ('http_403','http_429','provider_refused'):
|
||
sample = Sample(direction,sample_mib*1024**2,error='provider_refused',source=source)
|
||
elif previous.error in ('range_not_supported','invalid_content_range','unexpected_content','source_unavailable'):
|
||
sample = Sample(direction,sample_mib*1024**2,error=previous.error,source=source)
|
||
elif samples and previous.error and (samples[-2].error if len(samples)>1 else warmup.error):
|
||
sample = Sample(direction,sample_mib*1024**2,error='source_unavailable',source=source)
|
||
else:
|
||
sample = sample_transfer(direction,sample_mib*1024**2)
|
||
samples.append(sample)
|
||
if on_sample:
|
||
on_sample(direction, number, rounds, sample)
|
||
report['directions'][direction] = summarize(samples)
|
||
return report
|
||
|
||
|
||
def measure_many(rounds=3, sample_mib=4, sources=None, on_source=None, on_sample=None):
|
||
if not 3 <= rounds <= 8 or not 1 <= sample_mib <= 20:
|
||
raise ValueError('Invalid benchmark limits')
|
||
selected=source_ids(sources)
|
||
report={'schema_version':2,'method':'sequential_https_wall_time',
|
||
'payload_budget_bytes':payload_budget(rounds,sample_mib,selected),'sources':{}}
|
||
for identifier in selected:
|
||
if on_source: on_source(identifier,'start',None)
|
||
callback=(lambda *args: on_sample(identifier,*args)) if on_sample else None
|
||
result=measure(rounds,sample_mib,callback,source=identifier)
|
||
report['sources'][identifier]=result
|
||
if on_source: on_source(identifier,'complete',result)
|
||
return report
|
||
|
||
|
||
def is_complete(report):
|
||
return all(result['quality']=='consistent' for source in report['sources'].values()
|
||
for result in source['directions'].values())
|
||
|
||
|
||
def run_console(lang='zh',color='auto',rounds=3,sample_mib=4,sources=None,destination=None,on_result=None,*,out=None):
|
||
from .console import Console
|
||
out=out if out is not None else Console(lang,color);tr=out.text
|
||
selected=source_ids(sources)
|
||
budget=payload_budget(rounds,sample_mib,selected)/1024**2
|
||
out.line(tr(f'{len(selected)} 个来源 · 每方向 {rounds} 轮 × {sample_mib} MiB · 请求载荷最多 {budget:g} MiB',
|
||
f'{len(selected)} sources · {rounds} rounds × {sample_mib} MiB/direction · requested payload ≤ {budget:g} MiB'))
|
||
def source_output(identifier,event,result):
|
||
provider=SOURCES[identifier]
|
||
if event=='start':
|
||
out.line(f'{provider.name} · {provider.region}','title')
|
||
out.row(tr('测速方向','Direction'),tr('下载：节点 → 本机；上传：本机 → 节点','Download: node → this host; upload: this host → node'),'unknown')
|
||
if not provider.upload_url: out.row(tr('上传','Upload'),tr('不支持（该来源仅提供下载）','Not supported (download-only source)'),'unknown')
|
||
return
|
||
for direction,summary in result['directions'].items():
|
||
label=tr('下载统计','Download summary') if direction=='download' else tr('上传统计','Upload summary')
|
||
median=summary['median_mbps']
|
||
value=tr('无有效样本','No valid samples') if median is None else tr(
|
||
f"中位 {median:.2f} · 范围 {summary['min_mbps']:.2f}–{summary['max_mbps']:.2f} Mbps · {summary['successful']}/{summary['attempted']} 成功",
|
||
f"Median {median:.2f} · Range {summary['min_mbps']:.2f}–{summary['max_mbps']:.2f} Mbps · {summary['successful']}/{summary['attempted']} OK")
|
||
out.row(label,value,'ok' if summary['quality']=='consistent' else 'unknown')
|
||
note={'short_samples':tr('样本短于 1 秒，容量未知','Samples below 1s; capacity unknown'),
|
||
'variable':tr('轮间波动较大','High variation between rounds'),
|
||
'insufficient':tr('有效轮次不足','Insufficient successful rounds')}.get(summary['quality'])
|
||
if note: out.row('',note,'unknown')
|
||
def sample_output(identifier,direction,number,total,sample):
|
||
if destination is not None:
|
||
destination.setdefault('sample_events',[]).append(asdict(sample))
|
||
if on_result is not None:on_result()
|
||
label=tr('下载','Download') if direction=='download' else tr('上传','Upload')
|
||
value=f'{sample.mbps:.2f} Mbps · {sample.elapsed_s:.2f}s' if sample.mbps is not None else tr('未成功','Failed')+' · '+str(sample.error)
|
||
if sample.pop: value+=' · '+sample.pop
|
||
out.row(f'{label} {number}/{total}',value,'ok' if sample.mbps is not None else 'unknown')
|
||
result=measure_many(rounds,sample_mib,selected,source_output,sample_output)
|
||
if destination is not None:
|
||
destination.update(result)
|
||
return destination
|
||
return result
|