import argparse import json import random from argparse import Namespace from pathlib import Path import plotly.io as pio from flask import Flask, Response, jsonify, request, send_file from RWLS_3D import loop_soup_3d, loop_stats from sprinkling_gluing import ( build_figure as make_sprinkling_figure, choose_closest_macro_clusters, clipped_translated_local_dust, local_box_from_clusters, make_records, sprinkle_until_glued, ) from view_rwls3d_plotly import make_figure as make_loop_soup_figure from view_rwls3d_two_clusters import make_figure as make_cluster_figure APP_ROOT = Path(__file__).resolve().parent OUTPUT_DIR = APP_ROOT / "output" app = Flask(__name__) INDEX_HTML = """ 3D RWLS Sampler

3D RWLS Sampler

View

Seed

Min Length

Macro Cutoff L

Epsilon Min

Cutoff Metric

Min Macro Cluster

Dust c

Dust Seed

Local Dust

Box Padding

Min Diameter

Max Points

Max Loops

Cluster Choice

Ready.

Generating sample...

""" def plotly_js_path(): import plotly return Path(plotly.__file__).resolve().parent / "package_data" / "plotly.min.js" @app.get("/") def index(): return Response(INDEX_HTML, mimetype="text/html") @app.get("/plotly.min.js") def plotly_min_js(): return send_file(plotly_js_path(), mimetype="application/javascript") def clamp(value, low, high): return max(low, min(high, value)) def build_payload(data): N = clamp(int(data.get("N", 90)), 5, 240) c = clamp(float(data.get("c", 0.5)), 0.0, 1.0) seed = data.get("seed") if seed is None: seed = random.randrange(2**31) seed = int(seed) return { "N": N, "c": c, "seed": seed, "min_length": max(2, int(data.get("minLength", 8))), "min_diameter": max(0.0, float(data.get("minDiameter", 0.0))), "max_points": max(10_000, int(data.get("maxPoints", 220_000))), "max_loops": max(100, int(data.get("maxLoops", 5_000))), "mode": data.get("mode", "clusters"), "rank_by": data.get("rankBy", "steps"), "macro_cutoff": max(0.0, float(data.get("macroCutoff", 8.0))), "epsilon_min": max(0.0, float(data.get("epsilonMin", 1.0))), "cutoff_metric": data.get("cutoffMetric", "length"), "min_macro_cluster_loops": max(1, int(data.get("minMacroClusterLoops", 8))), "dust_c": clamp(float(data.get("dustC", 1.0)), 0.0, 1.0), "dust_seed": int(data.get("dustSeed", random.randrange(2**31))), "local_dust": bool(data.get("localDust", True)), "local_padding": max(1, int(data.get("localPadding", 8))), } def source_payload(loopsoup, params): return { "N": params["N"], "c": params["c"], "seed": params["seed"], "dimension": 3, "loopsoup": loopsoup, "stats": loop_stats(loopsoup), } def make_soup_view(payload, params): fig, shown_count, shown_steps = make_loop_soup_figure( payload, min_length=params["min_length"], min_diameter=params["min_diameter"], max_loops=params["max_loops"], max_points=params["max_points"], bucket_count=7, ) return fig, { "mode_label": "Loop Soup", "shown_loops": shown_count, "shown_steps": shown_steps, } def make_cluster_view(payload, params): cluster_args = Namespace( clusters=2, min_length=params["min_length"], min_diameter=params["min_diameter"], min_cluster_loops=1, rank_by=params["rank_by"], max_points_per_cluster=params["max_points"], line_width=7.0, height=900, show_boxes=True, cdn=False, ) fig, filtered_count, cluster_count, selected_clusters = make_cluster_figure(payload, cluster_args) shown_loops = sum(cluster["stats"]["loop_count"] for cluster in selected_clusters) shown_steps = sum(cluster["stats"]["total_steps"] for cluster in selected_clusters) closest_distance = selected_clusters[0].get("selection_distance") fig.update_layout( title=dict( text=( f"{'Two Closest Clusters' if params['rank_by'] == 'closest' else 'Two Intersection Clusters'} " f"(N={params['N']}, c={params['c']}, seed={params['seed']})" ) ) ) return fig, { "mode_label": "Two Clusters", "shown_loops": shown_loops, "shown_steps": shown_steps, "filtered_loops": filtered_count, "cluster_count": cluster_count, "closest_distance": closest_distance, "closest_distance_rescaled": ( closest_distance / max(1, params["N"] - 1) if closest_distance is not None else None ), } def make_sprinkling_view(macro_payload, params): macro_records = make_records( macro_payload["loopsoup"], min_size=params["macro_cutoff"], cutoff_metric=params["cutoff_metric"], ) if len(macro_records) < 2: raise ValueError("Not enough macro loops. Lower Macro Cutoff L.") selected_clusters, closest_distance, cluster_count, eligible_count = choose_closest_macro_clusters( macro_records, min_macro_cluster_loops=params["min_macro_cluster_loops"], ) dust_loops, local_box = generate_sprinkling_dust(macro_records, selected_clusters, params) dust_records = make_records( dust_loops, min_size=params["epsilon_min"], max_size=params["macro_cutoff"], cutoff_metric=params["cutoff_metric"], ) if not dust_records: raise ValueError("No dust loops in the epsilon window. Lower Epsilon Min or raise Macro Cutoff L.") sprinkle_result = sprinkle_until_glued( dust_records, macro_records, selected_clusters[0], selected_clusters[1], params["epsilon_min"], ) fig = make_sprinkling_figure( { "macro_c": params["c"], "dust_c": params["dust_c"], "macro_seed": params["seed"], "dust_seed": params["dust_seed"], }, macro_records, selected_clusters, sprinkle_result, params["N"], params["macro_cutoff"], params["epsilon_min"], params["cutoff_metric"], show_boxes=True, show_context_dust=True, max_context_points=params["max_points"], height=900, ) gluing_cutoff = sprinkle_result["gluing_cutoff"] cutoff_metric_label = "time steps" if params["cutoff_metric"] == "length" else "lattice diameter" return fig, { "mode_label": "Sprinkling", "shown_loops": sum(cluster["stats"]["loop_count"] for cluster in selected_clusters) + len(sprinkle_result["activated_dust_indices"]), "shown_steps": sum(cluster["stats"]["total_steps"] for cluster in selected_clusters), "filtered_loops": len(macro_records), "cluster_count": cluster_count, "eligible_cluster_count": eligible_count, "closest_distance": closest_distance, "closest_distance_rescaled": closest_distance / max(1, params["N"] - 1), "gluing_cutoff": gluing_cutoff, "gluing_cutoff_rescaled": ( gluing_cutoff / max(1, params["N"] - 1) if gluing_cutoff is not None and params["cutoff_metric"] == "diameter" else None ), "epsilon_min": params["epsilon_min"], "cutoff_metric": params["cutoff_metric"], "cutoff_metric_label": cutoff_metric_label, "min_macro_cluster_loops": params["min_macro_cluster_loops"], "dust_loops": len(dust_records), "bridge_dust_loops": len(sprinkle_result["bridge_dust_indices"]), "activated_dust_loops": len(sprinkle_result["activated_dust_indices"]), "local_box": local_box, "local_padding": params["local_padding"], } def generate_sprinkling_dust(macro_records, selected_clusters, params): if not params["local_dust"]: dust_loops = loop_soup_3d( params["N"], params["dust_c"], seed=params["dust_seed"], progress_every=0, ) return dust_loops, None lower, upper, box_shape = local_box_from_clusters( macro_records, selected_clusters, params["N"], params["local_padding"], ) dust_loops = clipped_translated_local_dust( box_shape, lower, params["dust_c"], params["dust_seed"], progress_every=0, ) return dust_loops, {"lower": lower, "upper": upper, "shape": box_shape} @app.post("/api/sample") def api_sample(): import time started = time.time() params = build_payload(request.get_json(force=True) or {}) try: loopsoup = loop_soup_3d(params["N"], params["c"], seed=params["seed"], progress_every=0) payload = source_payload(loopsoup, params) if params["mode"] == "soup": fig, view_meta = make_soup_view(payload, params) elif params["mode"] == "sprinkling": fig, view_meta = make_sprinkling_view(payload, params) else: fig, view_meta = make_cluster_view(payload, params) elapsed = time.time() - started meta = { "N": params["N"], "c": params["c"], "seed": params["seed"], "source_loops": len(loopsoup), "elapsed_seconds": elapsed, **view_meta, } return jsonify({"figure": pio.to_json(fig, validate=False), "meta": meta}) except Exception as exc: return jsonify({"error": str(exc)}), 400 def parse_args(): parser = argparse.ArgumentParser(description="Local interactive 3D RWLS sampler.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=5055) parser.add_argument("--debug", action="store_true") return parser.parse_args() def main(): args = parse_args() print(f"Open http://{args.host}:{args.port} in your browser") app.run(host=args.host, port=args.port, debug=args.debug) if __name__ == "__main__": main()