import argparse import hashlib import itertools import json import math import secrets import subprocess import sys from pathlib import Path SOURCE = Path(__file__).resolve().parents[3] / 'UNIHOLOG' / 'Traductores_V' TABLES = ((0, 0), (1, 1), (0, 1), (1, 0)) def worker(): sys.dont_write_bytecode = True sys.path.insert(0, str(SOURCE / 'Pruebas_V')) from pruebas_v import EntradaOraculo from circuito_senales_v import crear_circuito from ejecucion_v import ejecutar from latch_v import LatchV from v1 import V1 for line in sys.stdin: request = json.loads(line) local_setting = request['ajuste_local'] if local_setting not in (0, 1): raise ValueError('Ajuste local incorrecto') preparation = {'id': 'ensayo', 'tabla': request['tabla']} realization = LatchV(crear_circuito(preparation)) machine = V1(realizacion=realization) report = ejecutar(machine, oracle=EntradaOraculo(request['tabla'])) if report['status'] != 'MEDIDO': raise RuntimeError(json.dumps(report, ensure_ascii=False)) result = { 'id': request['id'], 'maquina_seleccionada': local_setting, 'bits': report['bits_q0_lsb'], 't0': report['t0'], 'preparacion_seconds': report['preparacion_seconds'], 'medida_seconds': report['medida_seconds'], } print(json.dumps(result), flush=True) def summarize(rows, bit_index): groups = {pair: [] for pair in itertools.product((0, 1), repeat=2)} wins = 0 agreements = 0 for row in rows: setting_a, setting_b = row['x'], row['y'] result_a = row['A']['bits'][bit_index] result_b = row['B']['bits'][bit_index] groups[setting_a, setting_b].append(1 if result_a == result_b else -1) wins += (result_a ^ result_b) == (setting_a & setting_b) agreements += result_a == result_b correlations = {f'{pair[0]}{pair[1]}': sum(values) / len(values) if values else None for pair, values in groups.items()} witness = None if any(value is None for value in correlations.values()) else ( correlations['00'] + correlations['01'] + correlations['10'] - correlations['11']) return { 'bit': bit_index, 'ensayos': len(rows), 'conteos_por_ajustes': {f'{pair[0]}{pair[1]}': len(values) for pair, values in groups.items()}, 'correlaciones': correlations, 'S': witness, 'abs_S': None if witness is None else abs(witness), 'estimable': witness is not None, 'aciertos_CHSH': wins, 'fraccion_aciertos': wins / len(rows), 'concordancia_entre_ramas': agreements / len(rows), } def main(output_dir, trials): output_dir.mkdir(parents=True, exist_ok=True) code_files = [SOURCE / name for name in ( 'v1.py', 'v_relacional.py', 'circuito_senales_v.py', 'comparador_temporal_v.py', 'latch_v.py', 'ejecucion_v.py')] hashes_before = {str(path): hashlib.sha256(path.read_bytes()).hexdigest() for path in code_files} hashes_before[str(Path(__file__).resolve())] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest() processes = [subprocess.Popen( [sys.executable, '-B', str(Path(__file__).resolve()), '--worker'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding='utf-8', bufsize=1) for branch in range(2)] rng = secrets.SystemRandom() summaries = [] recorded = [] failures = [] try: for mode in ('ordenado', 'aleatorio_externo'): rows = [] for trial in range(trials): table_index = (trial // 4) % len(TABLES) setting_a, setting_b = (trial % 4 // 2, trial % 2) if mode == 'ordenado' else ( rng.randrange(2), rng.randrange(2)) identifier = f'{mode}_{trial}' requests = [ {'id': identifier, 'tabla': TABLES[table_index], 'ajuste_local': setting_a}, {'id': identifier, 'tabla': TABLES[table_index], 'ajuste_local': setting_b}, ] for process, request in zip(processes, requests): process.stdin.write(json.dumps(request) + '\n') process.stdin.flush() responses = [] for process in processes: line = process.stdout.readline() if not line: raise RuntimeError('Fallo del proceso: ' + process.stderr.read()) responses.append(json.loads(line)) row = {'id': identifier, 'modo': mode, 'tabla': TABLES[table_index], 'x': setting_a, 'y': setting_b, 'A': responses[0], 'B': responses[1]} expected = TABLES[table_index][0] ^ TABLES[table_index][1] if any(response['bits'][0] != expected for response in responses): failures.append(identifier) rows.append(row) recorded.append(row) summaries.append({'modo': mode, 'q0': summarize(rows, 0), 'q1': summarize(rows, 1), 'por_oraculo': {''.join(map(str, table)): summarize( [row for row in rows if tuple(row['tabla']) == table], 0) for table in TABLES}}) print(mode, 'q0 S=', summaries[-1]['q0']['S'], 'q1 S=', summaries[-1]['q1']['S'], flush=True) finally: for process in processes: process.stdin.close() process.wait(timeout=15) strategies = [] for answers in itertools.product((0, 1), repeat=4): a_zero, a_one, b_zero, b_one = answers wins = sum(((a_zero, a_one)[setting_a] ^ (b_zero, b_one)[setting_b]) == (setting_a & setting_b) for setting_a, setting_b in itertools.product((0, 1), repeat=2)) correlations = [1 if (a_zero, a_one)[setting_a] == (b_zero, b_one)[setting_b] else -1 for setting_a, setting_b in itertools.product((0, 1), repeat=2)] witness = sum(correlations[:3]) - correlations[3] strategies.append({'respuestas_A0_A1_B0_B1': answers, 'S': witness, 'fraccion_aciertos': wins / 4}) assert max(abs(strategy['S']) for strategy in strategies) == 2 assert max(strategy['fraccion_aciertos'] for strategy in strategies) == 0.75 assert not failures assert all(summary['q0']['S'] in (None, 2) for summary in summaries) unchanged = all(hashlib.sha256(Path(path).read_bytes()).hexdigest() == digest for path, digest in hashes_before.items()) assert unchanged summary = { 'tipo': 'ENSAYO_LOCAL_SELECTORES_V1_Y_ANALISIS_CHSH', 'implementaciones': ['Python / proceso A', 'Python / proceso B'], 'prolog_ejecutado': False, 'maquinas': 'A0 y A1 son la misma V1; B0 y B1 son la misma V1; nueva preparacion por ensayo', 'selector': 'Externo; no genera estados ni salidas internas de V', 'entrada_comun': 'Tabla del oraculo; ciclo conocido independiente del selector aleatorio', 'alcance': 'Ensayo software en un ordenador. No aislamiento fisico ni test Bell sin lagunas.', 'aislamiento': 'Procesos separados, sin canal directo; el controlador conoce los dos ajustes. No se certifica ausencia de canales laterales.', 'independencia': 'Los ajustes aleatorios se generan externamente tras completar el ensayo previo. No es una certificacion fisica de independencia.', 'ordenado': 'Control de concordancia; ajustes previsibles, no prueba independiente de Bell.', 'significacion': 'No se ejecuta prueba de significacion ni se declara violacion a partir de fluctuaciones.', 'resumenes': summaries, 'discrepancias_q0_con_Deutsch': failures, 'estrategias_locales_binarias': strategies, 'maximo_CHSH_exacto_local': 2, 'maximo_aciertos_exacto_local': 0.75, 'fuentes_sha256': hashes_before, 'fuentes_intactas_al_final': unchanged, 'no_ejecutado': ['Prolog', 'España e Italia', 'hardware cuantico', 'Q4-V4'], } (output_dir / 'resultados.json').write_text( json.dumps(summary, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') with (output_dir / 'registros.jsonl').open('w', encoding='utf-8') as stream: for row in recorded: stream.write(json.dumps(row, ensure_ascii=False) + '\n') print('Ensayos pareados:', len(recorded), 'ejecuciones V1:', 2 * len(recorded)) print('Discrepancias q0:', len(failures), 'maximo exacto local:', 2) if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('--worker', action='store_true') parser.add_argument('--output', type=Path) parser.add_argument('--trials', type=int, default=4096) args = parser.parse_args() if args.worker: worker() else: if args.output is None or args.trials < 16 or args.trials % 16: parser.error('Se requiere salida y ensayos multiplos de 16') main(args.output, args.trials)