lihato/np_app
1
0
Fork 0
Code Issues Pull requests Projects Releases 2 Packages 1 Wiki Activity Actions
np_app/service/server.py

2518 lines
92 KiB
Python
Raw Blame History

import json
import mimetypes
import multiprocessing
import os
import queue
import re
import threading
import time
import traceback
from decimal import Decimal
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
import subprocess
import tempfile
from urllib.parse import parse_qs, urlencode, urlparse
from uuid import uuid4
import numpy
from nupack import (
Complex,
Complementarity,
Domain,
Diversity,
EnergyMatch,
DesignOptions,
Library,
Match,
Model,
Pattern,
Similarity,
SSM,
SetSpec,
Strand,
TargetComplex,
TargetStrand,
TargetTube,
Tube,
Weights,
Window,
complex_analysis,
complex_design,
tube_analysis,
tube_design,
)
from nupack import config as nupack_config
from account import AccountStore, OIDCAuth
from split_strand_svg import render_split_strands_svg
try:
import redis
except ImportError:
redis = None
ROOT = Path(__file__).resolve().parent
INDEX_PATH = ROOT / "index.html"
FAVICON_PATH = ROOT / "favicon.svg"
GUIDE_PATH = ROOT / "design-guide.html"
HOST = os.environ.get("NP_HOST", "0.0.0.0")
PORT = int(os.environ.get("NP_PORT", "18765"))
RNA_PLOT_CMD = os.environ.get("RNA_PLOT_CMD", "RNAplot")
ENABLE_RNAPLOT = os.environ.get("ENABLE_RNAPLOT", "1") != "0"
STRUCTURE_PLOT_MODE = os.environ.get("NP_STRUCTURE_PLOT_MODE", "auto").strip().lower()
RUN_MODE = os.environ.get("NP_RUN_MODE", "server")
REDIS_URL = os.environ.get("NP_REDIS_URL", "").strip()
JOB_QUEUE_KEY = os.environ.get("NP_JOB_QUEUE_KEY", "np_replica:jobs")
JOB_RUNNING_KEY = os.environ.get("NP_JOB_RUNNING_KEY", "np_replica:jobs:running")
WORKER_CONCURRENCY = int(os.environ.get("NP_WORKER_CONCURRENCY", "2"))
PER_JOB_THREAD_LIMIT = int(os.environ.get("NP_PER_JOB_THREAD_LIMIT", "1"))
NUPACK_CACHE_GB = float(os.environ.get("NP_NUPACK_CACHE_GB", "2.0"))
ACCOUNT_DB_PATH = os.environ.get("NP_ACCOUNT_DB_PATH", "/data/np-replica.sqlite3")
UNIT_SCALE = {
"M": 1.0,
"mM": 1e-3,
"uM": 1e-6,
"nM": 1e-9,
"pM": 1e-12,
}
IUPAC_CODES = "ACGTUWSMKRYBDHVN"
IUPAC_CONSTRAINT_TOKEN = re.compile(rf"[{IUPAC_CODES}](?:\d+)?")
VALID_COMPUTE = {"pfunc", "pairs", "mfe", "sample", "subopt", "ensemble_size"}
CANCEL_REQUESTED_STATUS = "cancel_requested"
CANCELED_STATUS = "canceled"
TERMINAL_JOB_STATUSES = {"success", "error", CANCELED_STATUS}
JOB_STORE = {}
JOB_LOCK = threading.Lock()
JOB_TTL_SECONDS = int(os.environ.get("NP_JOB_TTL_SECONDS", "3600"))
JOB_HEARTBEAT_SECONDS = max(1, int(os.environ.get("NP_JOB_HEARTBEAT_SECONDS", "30")))
JOB_MAX_COUNT = int(os.environ.get("NP_JOB_MAX_COUNT", "64"))
SHARE_STORE = {}
SHARE_LOCK = threading.Lock()
SHARE_MAX_COUNT = int(os.environ.get("NP_SHARE_MAX_COUNT", "100"))
SHARE_MAX_BYTES = int(os.environ.get("NP_SHARE_MAX_BYTES", str(12 * 1024 * 1024)))
SHARE_KEY_PREFIX = os.environ.get("NP_SHARE_KEY_PREFIX", "np_replica:shares")
SHARE_INDEX_KEY = f"{SHARE_KEY_PREFIX}:index"
REDIS_CLIENT = None
ACCOUNT_STORE = AccountStore(ACCOUNT_DB_PATH)
OIDC_AUTH = None
def json_bytes(payload, status=HTTPStatus.OK):
return (
status,
"application/json; charset=utf-8",
json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8"),
)
def log_event(message):
print(f"[np-replica] {time.strftime('%Y-%m-%d %H:%M:%S')} {message}", flush=True)
def apply_thread_limits():
# Keep native math libraries and NUPACK's own executor from using all CPU cores inside one job.
thread_limit_int = max(1, PER_JOB_THREAD_LIMIT)
thread_limit = str(thread_limit_int)
for key in (
"OMP_NUM_THREADS",
"OPENBLAS_NUM_THREADS",
"MKL_NUM_THREADS",
"NUMEXPR_NUM_THREADS",
"VECLIB_MAXIMUM_THREADS",
"GOTO_NUM_THREADS",
):
os.environ[key] = thread_limit
nupack_config.threads = thread_limit_int
nupack_config.cache = max(0.1, NUPACK_CACHE_GB)
def html_bytes(path):
body = path.read_bytes()
mime, _ = mimetypes.guess_type(str(path))
return HTTPStatus.OK, mime or "text/html; charset=utf-8", body
def file_etag(path):
stat = path.stat()
return f'W/"{stat.st_mtime_ns:x}-{stat.st_size:x}"'
def redis_enabled():
return bool(REDIS_URL and redis is not None)
def redis_client():
global REDIS_CLIENT
if not redis_enabled():
return None
if REDIS_CLIENT is None:
REDIS_CLIENT = redis.Redis.from_url(REDIS_URL, decode_responses=True)
return REDIS_CLIENT
OIDC_AUTH = OIDCAuth(redis_client)
def queue_size():
if not redis_enabled():
with JOB_LOCK:
return sum(1 for job in JOB_STORE.values() if job.get("status") in {"queued", "running"})
try:
return int(redis_client().llen(JOB_QUEUE_KEY))
except Exception:
return None
def prune_redis_running_jobs():
if not redis_enabled():
return 0
client = redis_client()
removed = 0
try:
for job_id in client.smembers(JOB_RUNNING_KEY):
raw = client.get(job_key(job_id))
if raw is None:
client.srem(JOB_RUNNING_KEY, job_id)
removed += 1
continue
try:
if json.loads(raw).get("status") != "running":
client.srem(JOB_RUNNING_KEY, job_id)
removed += 1
except Exception:
client.srem(JOB_RUNNING_KEY, job_id)
removed += 1
except Exception:
return removed
return removed
def job_stats():
stats = {
"queued": 0,
"running": 0,
"success": 0,
"error": 0,
}
if redis_enabled():
prune_redis_running_jobs()
client = redis_client()
stats["queued"] = int(client.llen(JOB_QUEUE_KEY))
stats["running"] = int(client.scard(JOB_RUNNING_KEY))
return stats
with JOB_LOCK:
prune_jobs()
for job in JOB_STORE.values():
status = job.get("status")
if status in stats:
stats[status] += 1
return stats
def job_key(job_id):
return f"np_replica:job:{job_id}"
def share_key(share_id):
return f"{SHARE_KEY_PREFIX}:item:{share_id}"
def normalize_sequence(sequence):
return "".join(sequence.upper().split())
def normalize_design_sequence(sequence):
return re.sub(r"\s+", "", str(sequence or "").upper())
def is_valid_iupac_constraint(sequence):
seq = normalize_design_sequence(sequence)
if not seq:
return False
index = 0
while index < len(seq):
match = IUPAC_CONSTRAINT_TOKEN.match(seq, index)
if not match:
return False
index = match.end()
return True
def ensure_unit_interval_limits(lower, upper, label):
if not (0 <= lower <= upper <= 1):
raise ValueError(f"{label} limits must satisfy 0 <= lower <= upper <= 1.")
def unit_to_molar(value, unit):
if unit not in UNIT_SCALE:
raise ValueError(f"Unsupported concentration unit: {unit}")
return float(value) * UNIT_SCALE[unit]
def build_model(model_input):
return Model(
material=model_input.get("material", "rna"),
ensemble=model_input.get("ensemble", "stacking"),
celsius=float(model_input.get("celsius", 37.0)),
sodium=float(model_input.get("sodium", 1.0)),
magnesium=float(model_input.get("magnesium", 0.0)),
)
def build_model_summary(model_input):
return {
"material": model_input.get("material", "rna"),
"ensemble": model_input.get("ensemble", "stacking"),
"celsius": float(model_input.get("celsius", 37.0)),
"sodium": float(model_input.get("sodium", 1.0)),
"magnesium": float(model_input.get("magnesium", 0.0)),
}
def parse_compute(payload):
compute = payload.get("compute") or ["pfunc", "mfe"]
compute = [item for item in compute if item in VALID_COMPUTE]
if not compute:
raise ValueError("At least one compute option is required.")
return compute
def parse_options(payload):
raw = payload.get("options") or {}
options = {
"num_sample": int(raw.get("num_sample", 20)),
"energy_gap": float(raw.get("energy_gap", 1.0)),
"sparsity_fraction": float(raw.get("sparsity_fraction", 1.0)),
"sparsity_threshold": float(raw.get("sparsity_threshold", 0.0)),
"single_mfe": bool(raw.get("single_mfe", False)),
"result_limit": int(raw.get("result_limit", 25)),
"pairs_preview_size": int(raw.get("pairs_preview_size", 24)),
}
if options["num_sample"] < 0 or options["num_sample"] > 1000:
raise ValueError("num_sample must be between 0 and 1000.")
if options["energy_gap"] < 0:
raise ValueError("energy_gap must be non-negative.")
if not 0 <= options["sparsity_fraction"] <= 1:
raise ValueError("sparsity_fraction must be between 0 and 1.")
if options["sparsity_threshold"] < 0:
raise ValueError("sparsity_threshold must be non-negative.")
if options["result_limit"] < 1:
raise ValueError("result_limit must be at least 1.")
if options["pairs_preview_size"] < 8 or options["pairs_preview_size"] > 128:
raise ValueError("pairs_preview_size must be between 8 and 128.")
return options
def parse_design_options(payload):
raw = payload.get("design") or {}
options = {
"trials": int(raw.get("trials", 1)),
"result_limit": int(raw.get("result_limit", 25)),
"off_target_max_size": int(raw.get("off_target_max_size", payload.get("tube", {}).get("max_size", 2))),
"stop_condition": float(raw.get("stop_condition", 0.02)),
"seed": int(raw.get("seed", 0)),
"wobble_mutations": bool(raw.get("wobble_mutations", False)),
"max_time_seconds": int(raw.get("max_time_seconds", 0)),
}
if options["trials"] < 1 or options["trials"] > 8:
raise ValueError("design trials must be between 1 and 8.")
if options["result_limit"] < 1:
raise ValueError("design result_limit must be at least 1.")
if options["off_target_max_size"] < 1 or options["off_target_max_size"] > 8:
raise ValueError("design off_target_max_size must be between 1 and 8.")
if not 0 < options["stop_condition"] < 1:
raise ValueError("design stop_condition must be between 0 and 1.")
if options["max_time_seconds"] < 0:
raise ValueError("design max_time_seconds must be non-negative.")
return options
def build_strands(strand_payload):
if not strand_payload:
raise ValueError("At least one strand is required.")
strand_map = {}
ordered = []
for row in strand_payload:
name = (row.get("name") or "").strip()
sequence = normalize_sequence(row.get("sequence") or "")
if not name:
raise ValueError("Every strand needs a name.")
if not sequence:
raise ValueError(f"Strand {name} is missing a sequence.")
if name in strand_map:
raise ValueError(f"Duplicate strand name: {name}")
if not re.fullmatch(r"[ACGTUWSMKRYBDHVN]+", sequence):
raise ValueError(f"Strand {name} contains unsupported characters.")
concentration = float(row.get("concentration", 0))
if concentration < 0:
raise ValueError(f"Strand {name} concentration must be non-negative.")
strand = Strand(sequence, name=name)
strand_map[name] = strand
ordered.append(
{
"name": name,
"sequence": sequence,
"concentration": concentration,
"unit": row.get("unit", "uM"),
"object": strand,
}
)
return strand_map, ordered
def build_design_domains(domain_payload):
domain_map = {}
ordered = []
for row in domain_payload:
name = (row.get("name") or "").strip()
sequence_constraint = normalize_design_sequence(row.get("sequence") or "")
if not name:
raise ValueError("Every design domain needs a name.")
if not sequence_constraint:
raise ValueError(f"Design domain {name} is missing a sequence constraint.")
if name in domain_map:
raise ValueError(f"Duplicate domain name: {name}")
if not is_valid_iupac_constraint(sequence_constraint):
raise ValueError(f"Design domain {name} contains unsupported constraint characters.")
domain = Domain(sequence_constraint, name=name)
domain_map[name] = domain
ordered.append(
{
"name": name,
"constraint": sequence_constraint,
"object": domain,
}
)
return domain_map, ordered
def parse_domain_composition(text, domain_map):
tokens = [token.strip() for token in re.split(r"[\s,]+", str(text or "").strip()) if token.strip()]
if not tokens:
raise ValueError("Domain composition cannot be empty.")
domains = []
for token in tokens:
complement = False
domain_name = token
if token.startswith("~"):
complement = True
domain_name = token[1:]
elif token.endswith("*"):
complement = True
domain_name = token[:-1]
if not domain_name:
raise ValueError(f"Invalid domain token: {token}")
if domain_name not in domain_map:
raise ValueError(f"Unknown domain name in strand composition: {domain_name}")
domain = domain_map[domain_name]
domains.append(~domain if complement else domain)
return domains
def build_design_strands(strand_payload, domain_map=None):
if not strand_payload:
raise ValueError("At least one design strand is required.")
domain_map = domain_map or {}
use_domain_composition = bool(domain_map)
target_strand_map = {}
ordered = []
for row in strand_payload:
name = (row.get("name") or "").strip()
raw_definition = str(row.get("sequence") or "").strip()
sequence_constraint = normalize_design_sequence(raw_definition)
if not name:
raise ValueError("Every design strand needs a name.")
if not raw_definition:
raise ValueError(f"Design strand {name} is missing a strand definition.")
if name in target_strand_map:
raise ValueError(f"Duplicate strand name: {name}")
if use_domain_composition:
try:
strand_domains = parse_domain_composition(raw_definition, domain_map)
except ValueError:
if (" " in raw_definition) or ("," in raw_definition) or ("~" in raw_definition) or ("*" in raw_definition):
raise
if not is_valid_iupac_constraint(sequence_constraint):
raise
inline_domain = Domain(sequence_constraint, name=name)
strand_domains = [inline_domain]
target_strand = TargetStrand(strand_domains, name=name)
constraint_kind = "sequence_constraint"
constraint_value = sequence_constraint
else:
inline_domain = None
target_strand = TargetStrand(strand_domains, name=name)
constraint_kind = "domain_composition"
constraint_value = raw_definition
else:
if not is_valid_iupac_constraint(sequence_constraint):
raise ValueError(f"Design strand {name} contains unsupported constraint characters.")
inline_domain = Domain(sequence_constraint, name=name)
strand_domains = [inline_domain]
target_strand = TargetStrand(strand_domains, name=name)
constraint_kind = "sequence_constraint"
constraint_value = sequence_constraint
target_strand_map[name] = target_strand
ordered.append(
{
"name": name,
"definition": raw_definition,
"constraint": constraint_value,
"constraint_kind": constraint_kind,
"object": target_strand,
"domains": strand_domains,
"domain": inline_domain,
}
)
return target_strand_map, ordered
def resolve_scope_reference(expr, domain_map, strand_map, *, allow_global=False):
text = str(expr or "").strip()
if not text:
if allow_global:
return None
raise ValueError("Constraint scope cannot be empty.")
if text in strand_map:
return strand_map[text]
return parse_domain_composition(text, domain_map)
def resolve_domain_scope(expr, domain_map):
text = str(expr or "").strip()
if not text:
raise ValueError("Domain scope cannot be empty.")
return parse_domain_composition(text, domain_map)
def resolve_target_complex_scope(expr, target_complex_map):
text = str(expr or "").strip()
if not text:
raise ValueError("Target complex scope cannot be empty.")
tokens = [token.strip() for token in re.split(r"[\s,]+", text) if token.strip()]
if not tokens:
raise ValueError("Target complex scope cannot be empty.")
missing = [token for token in tokens if token not in target_complex_map]
if missing:
raise ValueError(f"Unknown target complex name(s): {', '.join(missing)}")
return [target_complex_map[token] for token in tokens]
def parse_sequence_list(text):
if isinstance(text, list):
values = text
else:
normalized = str(text or "").replace("|", "\n").replace(";", "\n")
values = normalized.splitlines()
output = [item.strip() for item in values if item and item.strip()]
if not output:
raise ValueError("Constraint sequence list cannot be empty.")
for seq in output:
if not re.fullmatch(rf"[{IUPAC_CODES}]+", seq.upper()):
raise ValueError(f"Invalid sequence source: {seq!r}")
return output
def parse_pattern_list(text):
if isinstance(text, list):
values = text
else:
values = re.split(r"[\n,]+", str(text or ""))
output = [item.strip().upper() for item in values if item and item.strip()]
if not output:
raise ValueError("Pattern constraint requires at least one pattern.")
for pattern in output:
if not is_valid_iupac_constraint(pattern):
raise ValueError(f"Invalid pattern: {pattern!r}")
return output
def parse_catalog_list(text):
raw_lines = [line.strip() for line in str(text or "").splitlines() if line.strip()]
if not raw_lines:
raise ValueError("Library constraint requires at least one catalog row.")
catalog = []
for line in raw_lines:
library = [item.strip().upper() for item in re.split(r"[,|]+", line) if item.strip()]
if not library:
raise ValueError("Library constraint contains an empty catalog row.")
for seq in library:
if not re.fullmatch(rf"[{IUPAC_CODES}]+", seq):
raise ValueError(f"Invalid library sequence: {seq!r}")
catalog.append(library)
return catalog
def build_hard_constraints(payload, domain_map, strand_map):
constraints = []
for index, row in enumerate(payload.get("hard_constraints") or [], start=1):
constraint_type = (row.get("type") or "").strip().lower()
if not constraint_type:
continue
try:
def require_text(field, label):
value = str(row.get(field) or "").strip()
if not value:
raise ValueError(f"{label} is required.")
return value
if constraint_type == "match":
constraints.append(
Match(
resolve_scope_reference(require_text("left", "Match left scope"), domain_map, strand_map),
resolve_scope_reference(require_text("right", "Match right scope"), domain_map, strand_map),
)
)
elif constraint_type == "complementarity":
constraints.append(
Complementarity(
resolve_scope_reference(require_text("left", "Complementarity left scope"), domain_map, strand_map),
resolve_scope_reference(require_text("right", "Complementarity right scope"), domain_map, strand_map),
wobble_mutations=bool(row.get("wobble_mutations", False)),
)
)
elif constraint_type == "diversity":
word = int(row.get("word", 4))
types = int(row.get("types", 2))
if word < 1:
raise ValueError("Diversity word must be >= 1.")
if types < 1 or types > 4:
raise ValueError("Diversity types must be between 1 and 4.")
kwargs = {
"word": word,
"types": types,
}
scope = resolve_scope_reference(row.get("scope"), domain_map, strand_map, allow_global=True)
if scope is not None:
kwargs["scope"] = scope
constraints.append(Diversity(**kwargs))
elif constraint_type == "similarity":
lower = float(row.get("min_fraction", 0.0))
upper = float(row.get("max_fraction", 1.0))
ensure_unit_interval_limits(lower, upper, "Similarity")
reference = normalize_design_sequence(row.get("reference") or "")
if not is_valid_iupac_constraint(reference):
raise ValueError("Similarity reference must be a valid IUPAC constraint.")
constraints.append(
Similarity(
resolve_scope_reference(require_text("scope", "Similarity scope"), domain_map, strand_map),
reference,
limits=[lower, upper],
)
)
elif constraint_type == "window":
constraints.append(
Window(
resolve_scope_reference(require_text("scope", "Window scope"), domain_map, strand_map),
sources=parse_sequence_list(row.get("sources") or ""),
)
)
elif constraint_type == "library":
constraints.append(
Library(
resolve_scope_reference(require_text("scope", "Library scope"), domain_map, strand_map),
catalog=parse_catalog_list(row.get("catalog") or ""),
)
)
elif constraint_type == "pattern":
kwargs = {"patterns": parse_pattern_list(row.get("patterns") or "")}
scope = resolve_scope_reference(row.get("scope"), domain_map, strand_map, allow_global=True)
if scope is not None:
kwargs["scope"] = scope
constraints.append(Pattern(**kwargs))
else:
raise ValueError(f"Unsupported hard constraint type: {constraint_type}")
except Exception as exc:
raise ValueError(f"Invalid hard constraint #{index}: {exc}") from exc
return constraints
def build_soft_constraints(payload, domain_map, strand_map, target_complex_map):
constraints = []
for index, row in enumerate(payload.get("soft_constraints") or [], start=1):
constraint_type = (row.get("type") or "").strip().lower()
if not constraint_type:
continue
try:
def require_text(field, label):
value = str(row.get(field) or "").strip()
if not value:
raise ValueError(f"{label} is required.")
return value
if constraint_type == "pattern":
weight = float(row.get("weight", 1.0))
if weight < 0:
raise ValueError("Pattern weight must be non-negative.")
kwargs = {
"patterns": parse_pattern_list(row.get("patterns") or ""),
"weight": weight,
}
scope = resolve_scope_reference(row.get("scope"), domain_map, strand_map, allow_global=True)
if scope is not None:
kwargs["scope"] = scope
constraints.append(Pattern(**kwargs))
elif constraint_type == "similarity":
lower = float(row.get("min_fraction", 0.0))
upper = float(row.get("max_fraction", 1.0))
ensure_unit_interval_limits(lower, upper, "Similarity")
reference = normalize_design_sequence(row.get("reference") or "")
if not is_valid_iupac_constraint(reference):
raise ValueError("Similarity reference must be a valid IUPAC constraint.")
weight = float(row.get("weight", 1.0))
if weight < 0:
raise ValueError("Similarity weight must be non-negative.")
constraints.append(
Similarity(
resolve_scope_reference(require_text("scope", "Similarity scope"), domain_map, strand_map),
reference,
limits=[lower, upper],
weight=weight,
)
)
elif constraint_type == "ssm":
word = int(row.get("word", 4))
if word < 1:
raise ValueError("SSM word must be >= 1.")
weight = float(row.get("weight", 1.0))
if weight < 0:
raise ValueError("SSM weight must be non-negative.")
kwargs = {
"word": word,
"weight": weight,
}
scope = resolve_target_complex_scope(row.get("scope"), target_complex_map) if row.get("scope") else None
if scope is not None:
kwargs["scope"] = scope
constraints.append(SSM(**kwargs))
elif constraint_type == "energy_match":
weight = float(row.get("weight", 1.0))
if weight < 0:
raise ValueError("EnergyMatch weight must be non-negative.")
kwargs = {
"domains": resolve_domain_scope(require_text("scope", "EnergyMatch domain scope"), domain_map),
"weight": weight,
}
if row.get("energy_ref") not in {None, ""}:
kwargs["energy_ref"] = float(row.get("energy_ref"))
constraints.append(EnergyMatch(**kwargs))
else:
raise ValueError(f"Unsupported soft constraint type: {constraint_type}")
except Exception as exc:
raise ValueError(f"Invalid soft constraint #{index}: {exc}") from exc
return constraints
def parse_design_complexes(payload, target_strand_map):
target_rows = payload.get("design_complexes") or payload.get("design_targets") or []
if not target_rows:
raise ValueError("At least one design target complex is required.")
targets = []
target_complex_map = {}
for idx, row in enumerate(target_rows, start=1):
name = (row.get("name") or f"target_{idx}").strip() or f"target_{idx}"
if name in target_complex_map:
raise ValueError(f"Duplicate design target complex name: {name}")
strand_text = (row.get("strands") or "").strip()
structure = (row.get("structure") or "").strip()
if not strand_text:
raise ValueError(f"Design target complex {name} is missing strand membership.")
if not structure:
raise ValueError(f"Design target complex {name} is missing a target structure.")
tokens = [token.strip() for token in strand_text.split("+") if token.strip()]
if not tokens:
raise ValueError(f"Design target complex {name} has an invalid strand list.")
missing = [token for token in tokens if token not in target_strand_map]
if missing:
raise ValueError(
f"Unknown strand name(s) in design target complex {name}: {', '.join(missing)}"
)
target_complex = TargetComplex(
[target_strand_map[token] for token in tokens],
structure,
name=name,
)
target_payload = {
"name": name,
"strands": tokens,
"structure": structure,
"object": target_complex,
}
targets.append(target_payload)
target_complex_map[name] = target_complex
return targets, target_complex_map
def parse_design_tubes(payload, target_rows, target_complex_map, default_max_size):
tube_rows = payload.get("design_tubes") or []
if not tube_rows:
legacy_targets = payload.get("design_targets") or []
if legacy_targets:
on_targets = []
for row in legacy_targets:
if row.get("concentration") in {None, ""}:
continue
on_targets.append(
{
"complex": row.get("name"),
"concentration": row.get("concentration"),
"unit": row.get("unit", "uM"),
}
)
if on_targets:
tube_rows = [{
"name": ((payload.get("tube") or {}).get("name") or "design_tube"),
"max_size": default_max_size,
"on_targets": on_targets,
}]
if not tube_rows:
raise ValueError("Tube design requires at least one target tube.")
tubes = []
ordered_rows = []
tube_name_set = set()
for index, row in enumerate(tube_rows, start=1):
tube_name = (row.get("name") or f"tube_{index}").strip() or f"tube_{index}"
if tube_name in tube_name_set:
raise ValueError(f"Duplicate target tube name: {tube_name}")
tube_name_set.add(tube_name)
max_size = int(row.get("max_size", default_max_size))
if max_size < 1 or max_size > 8:
raise ValueError(f"Target tube {tube_name} max_size must be between 1 and 8.")
on_targets_input = row.get("on_targets") or []
if not on_targets_input:
raise ValueError(f"Target tube {tube_name} must include at least one on-target complex.")
on_targets = {}
serialized_on_targets = []
seen_complex_names = set()
for entry in on_targets_input:
complex_name = (entry.get("complex") or "").strip()
if complex_name not in target_complex_map:
raise ValueError(f"Unknown target complex {complex_name} in tube {tube_name}.")
if complex_name in seen_complex_names:
raise ValueError(f"Duplicate on-target complex {complex_name} in tube {tube_name}.")
seen_complex_names.add(complex_name)
concentration = float(entry.get("concentration", 0))
unit = entry.get("unit", "uM")
if concentration <= 0:
raise ValueError(f"Target tube {tube_name} requires positive target concentrations.")
concentration_M = unit_to_molar(concentration, unit)
complex_obj = target_complex_map[complex_name]
on_targets[complex_obj] = concentration_M
serialized_on_targets.append(
{
"complex": complex_name,
"concentration": concentration,
"unit": unit,
"target_concentration_M": concentration_M,
}
)
tube = TargetTube(
on_targets=on_targets,
off_targets=SetSpec(max_size=max_size),
name=tube_name,
)
ordered_rows.append(
{
"name": tube_name,
"max_size": max_size,
"on_targets": serialized_on_targets,
"object": tube,
}
)
tubes.append(tube)
return ordered_rows, tubes
def validate_design_object_names(design_domains, design_strands, target_rows, tube_rows):
name_map = {}
for kind, rows in (
("domain", design_domains),
("strand", design_strands),
("target complex", target_rows),
("target tube", tube_rows),
):
for row in rows:
name = (row.get("name") or "").strip()
if not name:
continue
prev = name_map.get(name)
if prev and prev != kind:
raise ValueError(
f"Object name {name!r} is reused across {prev} and {kind}. "
"Design object names must be unique in a job."
)
name_map[name] = kind
def build_defect_weights(payload, design_mode, design_domains, design_strands, target_rows, tube_rows):
rows = payload.get("defect_weights") or []
if not rows:
return None
domain_index = {item["name"]: item["object"] for item in design_domains}
strand_index = {item["name"]: item["object"] for item in design_strands}
complex_index = {item["name"]: item["object"] for item in target_rows}
tube_index = {item["name"]: item["object"] for item in tube_rows}
weight_base = [item["object"] for item in tube_rows] if design_mode == "tube" else [item["object"] for item in target_rows]
weights = Weights(weight_base)
for index, row in enumerate(rows, start=1):
kind = (row.get("kind") or "").strip().lower()
value = float(row.get("weight", 1.0))
if value < 0:
raise ValueError(f"Invalid defect weight #{index}: weight must be non-negative")
name = (row.get("name") or "").strip()
try:
if kind == "global":
weights[:] = value
elif kind == "domain":
weights[domain_index[name]] = value
elif kind == "strand":
weights[:, strand_index[name]] = value
elif kind == "complex":
weights[:, :, complex_index[name]] = value
elif kind == "tube":
if design_mode != "tube":
raise ValueError("tube-level defect weights are only valid in tube design mode")
weights[:, :, :, tube_index[name]] = value
else:
raise ValueError(f"Unsupported defect weight kind: {kind}")
except KeyError as exc:
raise ValueError(f"Unknown entity for defect weight #{index}: {name}") from exc
except Exception as exc:
raise ValueError(f"Invalid defect weight #{index}: {exc}") from exc
return weights
def parse_complex_lines(text, strand_map):
complexes = []
raw_lines = [line.strip() for line in text.splitlines() if line.strip()]
if not raw_lines:
raise ValueError("Complex mode requires at least one complex definition.")
for idx, line in enumerate(raw_lines, start=1):
tokens = [token.strip() for token in line.split("+") if token.strip()]
if not tokens:
raise ValueError(f"Invalid complex definition on line {idx}: {line}")
missing = [token for token in tokens if token not in strand_map]
if missing:
raise ValueError(
f"Unknown strand name(s) in complex line {idx}: {', '.join(missing)}"
)
strands = [strand_map[token] for token in tokens]
complexes.append(Complex(strands, name=f"complex_{idx}"))
return complexes
def stringify_complex(complex_obj):
return " + ".join(strand.name for strand in complex_obj.strands)
def flatten_sequence(complex_obj):
return "".join(str(strand) for strand in complex_obj.strands)
def sanitize_structure_for_rnaplot(structure):
return str(structure).replace("+", "")
def safe_name(name):
return re.sub(r"[^A-Za-z0-9_.-]+", "_", name).strip("_") or "structure"
def render_rnaplot_svg(complex_obj, structure, suffix):
if not ENABLE_RNAPLOT:
return None
seq_name = safe_name(f"{stringify_complex(complex_obj)}_{suffix}")
sequence = flatten_sequence(complex_obj)
structure_text = sanitize_structure_for_rnaplot(structure)
with tempfile.TemporaryDirectory(prefix="rnaplot-") as tmpdir:
tmp_path = Path(tmpdir)
input_path = tmp_path / f"{seq_name}.seq"
output_path = tmp_path / f"{seq_name}_ss.svg"
input_path.write_text(
f">{seq_name}\n{sequence}\n{structure_text}\n",
encoding="utf-8",
newline="\n",
)
process = subprocess.run(
[RNA_PLOT_CMD, "-f", "svg", input_path.name],
cwd=tmpdir,
capture_output=True,
text=True,
timeout=20,
)
if process.returncode != 0:
raise RuntimeError(process.stderr.strip() or "RNAplot failed.")
if not output_path.exists():
raise RuntimeError("RNAplot finished without producing an SVG file.")
return output_path.read_text(encoding="utf-8")
def render_structure_svg(complex_obj, structure, suffix):
structure_text = str(structure)
multistrand = len(complex_obj.strands) > 1 and "+" in structure_text
should_try_split = STRUCTURE_PLOT_MODE == "split" or (
STRUCTURE_PLOT_MODE != "rnaplot" and multistrand
)
split_error = None
if should_try_split:
try:
return render_split_strands_svg(
[str(strand) for strand in complex_obj.strands],
structure_text,
title=safe_name(f"{stringify_complex(complex_obj)}_{suffix}"),
)
except Exception as exc:
split_error = exc
if STRUCTURE_PLOT_MODE == "split":
raise RuntimeError(f"Split-strand layout failed: {exc}") from exc
if STRUCTURE_PLOT_MODE == "split":
if split_error is not None:
raise RuntimeError(f"Split-strand layout failed: {split_error}") from split_error
return None
try:
return render_rnaplot_svg(complex_obj, structure, suffix)
except Exception as exc:
if split_error is not None:
raise RuntimeError(
f"Split-strand layout failed: {split_error}; RNAplot fallback failed: {exc}"
) from exc
raise
def serialize_structures(complex_obj, items, plot_kind=None):
output = []
for index, item in enumerate(items or []):
row = {
"structure": str(item.structure),
"energy": round(float(item.energy), 6),
"stack_energy": round(float(item.stack_energy), 6),
}
if plot_kind and index == 0:
try:
row["rnaplot_svg"] = render_structure_svg(complex_obj, item.structure, plot_kind)
except Exception as exc:
row["rnaplot_error"] = str(exc)
output.append(row)
return output
def serialize_samples(items):
return [str(item) for item in (items or [])]
def serialize_pairs(pair_matrix, preview_limit=24):
if pair_matrix is None:
return None
dense = numpy.asarray(pair_matrix.to_array(), dtype=float)
preview = dense[:preview_limit, :preview_limit]
return {
"shape": list(dense.shape),
"preview": numpy.round(preview, 6).tolist(),
"preview_size": int(preview_limit),
"preview_truncated": bool(dense.shape[0] > preview_limit),
}
def parse_structure_pairs(structure):
stack = []
pairs = []
compact_index = -1
index_map = []
for char in str(structure):
if char == "+":
continue
compact_index += 1
index_map.append(compact_index)
if char == "(":
stack.append(compact_index)
elif char == ")":
if not stack:
raise ValueError("Unbalanced structure: missing opening bracket.")
left = stack.pop()
pairs.append((left, compact_index))
if stack:
raise ValueError("Unbalanced structure: missing closing bracket.")
pairs.sort()
return pairs, compact_index + 1
def build_structure_probabilities(structure, pair_matrix):
if pair_matrix is None:
return None
dense = numpy.asarray(pair_matrix.to_array(), dtype=float)
mfe_pairs, structure_length = parse_structure_pairs(structure)
if dense.shape[0] != dense.shape[1]:
return None
if structure_length != dense.shape[0]:
return None
pair_partner = {}
for left, right in mfe_pairs:
pair_partner[left] = right
pair_partner[right] = left
residue_probabilities = []
for index in range(structure_length):
partner = pair_partner.get(index)
if partner is None:
probability = float(dense[index, index])
else:
probability = float(dense[index, partner])
residue_probabilities.append(round(probability, 6))
pair_probabilities = [
{
"i": left + 1,
"j": right + 1,
"probability": round(float(dense[left, right]), 6),
}
for left, right in mfe_pairs
]
return {
"length": structure_length,
"residue_probabilities": residue_probabilities,
"pair_probabilities": pair_probabilities,
}
def serialize_complex_result(complex_obj, data, pairs_preview_size=24):
payload = {
"name": getattr(complex_obj, "name", None),
"display": stringify_complex(complex_obj),
"strand_names": [strand.name for strand in complex_obj.strands],
"sequence": flatten_sequence(complex_obj),
"strand_lengths": [len(str(strand)) for strand in complex_obj.strands],
}
if data.pfunc is not None:
payload["pfunc"] = format(data.pfunc, "g") if isinstance(data.pfunc, Decimal) else str(data.pfunc)
if data.free_energy is not None:
payload["free_energy_kcal_mol"] = round(float(data.free_energy), 6)
if data.ensemble_size is not None:
payload["ensemble_size"] = int(data.ensemble_size)
if data.mfe_stack is not None:
payload["mfe_stack_kcal_mol"] = round(float(data.mfe_stack), 6)
if data.mfe is not None:
payload["mfe"] = serialize_structures(complex_obj, data.mfe, plot_kind="mfe")
if data.subopt is not None:
payload["subopt"] = serialize_structures(complex_obj, data.subopt)
if data.sample is not None:
payload["sample"] = serialize_samples(data.sample)
if data.pairs is not None:
payload["pairs"] = serialize_pairs(data.pairs, preview_limit=pairs_preview_size)
if payload.get("mfe"):
payload["mfe"][0]["pair_probability_annotations"] = build_structure_probabilities(
payload["mfe"][0]["structure"], data.pairs
)
return payload
def attach_pairs_result(payload, pair_data, pairs_preview_size=24):
if pair_data is None or pair_data.pairs is None:
return payload
payload["pairs"] = serialize_pairs(pair_data.pairs, preview_limit=pairs_preview_size)
if payload.get("mfe"):
payload["mfe"][0]["pair_probability_annotations"] = build_structure_probabilities(
payload["mfe"][0]["structure"], pair_data.pairs
)
return payload
def serialize_target_defect_rows(df, limit=None):
if df is None:
return []
rows = []
for record in df.to_dict("records"):
row = {}
for key, value in record.items():
if key in {"tube", "complex"}:
continue
if isinstance(value, numpy.floating):
row[key] = float(value)
else:
row[key] = value
rows.append(row)
if limit is not None:
rows = rows[:limit]
return rows
def serialize_design_result(
design_result,
design_mode,
model_summary,
design_options,
target_rows,
tube_rows,
ordered_domains,
ordered_strands,
):
designed_domains = []
designed_domain_map = getattr(design_result, "domains", {}) or {}
for item in ordered_domains:
designed_domain = designed_domain_map.get(item["object"])
designed_domains.append(
{
"name": item["name"],
"constraint": item["constraint"],
"sequence": str(designed_domain) if designed_domain is not None else None,
"length": len(str(designed_domain)) if designed_domain is not None else None,
}
)
designed_strands = []
for item in ordered_strands:
target_strand = item["object"]
analysis_strand = design_result.to_analysis[target_strand]
designed_strands.append(
{
"name": item["name"],
"constraint": item["constraint"],
"constraint_kind": item.get("constraint_kind", "sequence_constraint"),
"definition": item.get("definition", item["constraint"]),
"sequence": str(analysis_strand),
"length": len(str(analysis_strand)),
}
)
target_complexes = []
for target in target_rows:
target_complex = target["object"]
analysis_complex = design_result.to_analysis[target_complex]
target_complexes.append(
{
"name": target["name"],
"display": stringify_complex(analysis_complex),
"strand_names": list(target["strands"]),
"structure": target["structure"],
"sequence": flatten_sequence(analysis_complex),
"target_concentration_M": target.get("target_concentration_M"),
}
)
concentration_rows = []
concentration_table = getattr(getattr(design_result, "concentrations", None), "table", None)
if design_mode == "tube" and concentration_table is not None:
for record in concentration_table.to_dict("records"):
concentration_rows.append(
{
"complex_name": record["complex_name"],
"concentration_M": float(record["concentration"]),
"target_concentration_M": float(record["target_concentration"]),
"nucleotides": int(record["nucleotides"]),
}
)
concentration_rows.sort(key=lambda row: row["concentration_M"], reverse=True)
return {
"workflow": "design",
"mode": design_mode,
"model": model_summary,
"compute": ["design"],
"options": design_options,
"strands": designed_strands,
"complexes": target_complexes,
"total_complex_count": len(target_complexes),
"displayed_complex_count": len(target_complexes),
"tube": {
"name": tube_rows[0]["name"] if tube_rows else "design_tube",
"max_size": tube_rows[0]["max_size"] if tube_rows else design_options["off_target_max_size"],
"complex_concentrations": concentration_rows[: design_options["result_limit"]],
"total_complex_concentrations": len(concentration_rows),
} if design_mode == "tube" else None,
"design": {
"ensemble_defect": float(design_result.ensemble_defect),
"domains": designed_domains,
"stats": {
key: float(value) if isinstance(value, (int, float, numpy.floating)) else value
for key, value in design_result.stats.items()
},
"job_options": {
"f_stop": design_options["stop_condition"],
"seed": design_options["seed"],
"wobble_mutations": design_options["wobble_mutations"],
"max_time_seconds": design_options["max_time_seconds"],
},
"target_tubes": [
{
"name": row["name"],
"max_size": row["max_size"],
"on_targets": row["on_targets"],
}
for row in tube_rows
],
"targets": target_complexes,
"defects": {
"tubes": serialize_target_defect_rows(getattr(design_result.defects, "tubes", None)),
"complexes": serialize_target_defect_rows(getattr(design_result.defects, "complexes", None)),
"tube_complexes": serialize_target_defect_rows(getattr(design_result.defects, "tube_complexes", None)),
},
"concentrations": concentration_rows[: design_options["result_limit"]],
},
}
def sort_concentrations(complex_concentrations):
rows = []
for complex_obj, value in complex_concentrations.items():
rows.append(
{
"display": stringify_complex(complex_obj),
"strand_names": [strand.name for strand in complex_obj.strands],
"concentration_M": float(value),
}
)
rows.sort(key=lambda row: row["concentration_M"], reverse=True)
return rows
def run_job_payload(payload):
workflow = payload.get("workflow", "analysis")
mode = payload.get("mode", "tube")
model_input = payload.get("model") or {}
model = build_model(model_input)
model_summary = build_model_summary(model_input)
if workflow == "design":
design_options = parse_design_options(payload)
design_job_option_kwargs = {
"f_stop": design_options["stop_condition"],
"seed": design_options["seed"],
"wobble_mutations": design_options["wobble_mutations"],
}
if design_options["max_time_seconds"] > 0:
design_job_option_kwargs["max_time"] = design_options["max_time_seconds"]
design_job_options = DesignOptions(**design_job_option_kwargs)
design_domain_map, design_domains = build_design_domains(payload.get("design_domains") or [])
target_strand_map, design_strands = build_design_strands(
payload.get("strands") or [],
domain_map=design_domain_map,
)
target_rows, target_complex_map = parse_design_complexes(payload, target_strand_map)
hard_constraints = build_hard_constraints(payload, design_domain_map, target_strand_map)
soft_constraints = build_soft_constraints(
payload,
design_domain_map,
target_strand_map,
target_complex_map,
)
tube_rows = []
design_tubes = []
if mode == "tube":
tube_rows, design_tubes = parse_design_tubes(
payload,
target_rows,
target_complex_map,
design_options["off_target_max_size"],
)
validate_design_object_names(design_domains, design_strands, target_rows, tube_rows)
defect_weights = build_defect_weights(
payload,
mode,
design_domains,
design_strands,
target_rows,
tube_rows,
)
if mode == "tube":
tube_name = ((payload.get("tube") or {}).get("name") or "design_tube").strip() or "design_tube"
design_job = tube_design(
tubes=design_tubes,
model=model,
options=design_job_options,
hard_constraints=hard_constraints,
soft_constraints=soft_constraints,
defect_weights=defect_weights,
)
elif mode == "complex":
design_job = complex_design(
complexes=[row["object"] for row in target_rows],
model=model,
options=design_job_options,
hard_constraints=hard_constraints,
soft_constraints=soft_constraints,
defect_weights=defect_weights,
)
else:
raise ValueError(f"Unsupported design mode: {mode}")
results = design_job.run(trials=design_options["trials"])
best_result = min(results, key=lambda item: float(item.ensemble_defect))
return serialize_design_result(
best_result,
mode,
model_summary,
design_options,
target_rows,
tube_rows,
design_domains,
design_strands,
)
if workflow != "analysis":
raise ValueError(f"Unsupported workflow: {workflow}")
compute = parse_compute(payload)
options = parse_options(payload)
nupack_options = {
key: value
for key, value in options.items()
if key not in {"result_limit", "pairs_preview_size"}
}
strand_map, strands = build_strands(payload.get("strands") or [])
if mode == "tube":
wants_pairs = "pairs" in compute
tube_compute = [item for item in compute if item != "pairs"]
if not tube_compute:
tube_compute = ["pfunc"]
tube_cfg = payload.get("tube") or {}
max_size = int(tube_cfg.get("max_size", 2))
if max_size < 1 or max_size > 8:
raise ValueError("max_size must be between 1 and 8.")
strand_concentrations = {
item["object"]: unit_to_molar(item["concentration"], item["unit"])
for item in strands
}
tube = Tube(
strands=strand_concentrations,
complexes=SetSpec(max_size=max_size),
name=(tube_cfg.get("name") or "tube1").strip() or "tube1",
)
result = tube_analysis([tube], model=model, compute=tube_compute, options=nupack_options)
concentration_rows = sort_concentrations(result[tube].complex_concentrations)
total_complex_count = len(concentration_rows)
result_limit = options["result_limit"]
displayed_concentration_rows = concentration_rows[:result_limit]
concentration_by_display = {
row["display"]: row["concentration_M"] for row in displayed_concentration_rows
}
pair_data_by_display = {}
if wants_pairs:
displayed_complex_objects = [
complex_obj
for complex_obj in result.complexes
if stringify_complex(complex_obj) in concentration_by_display
]
if displayed_complex_objects:
pair_result = complex_analysis(
displayed_complex_objects,
model=model,
compute=["pairs"],
options=nupack_options,
)
pair_data_by_display = {
stringify_complex(complex_obj): data
for complex_obj, data in pair_result.complexes.items()
}
complexes = []
for complex_obj, data in result.complexes.items():
display_name = stringify_complex(complex_obj)
if display_name not in concentration_by_display:
continue
row = serialize_complex_result(
complex_obj,
data,
pairs_preview_size=options["pairs_preview_size"],
)
if wants_pairs:
attach_pairs_result(
row,
pair_data_by_display.get(display_name),
pairs_preview_size=options["pairs_preview_size"],
)
row["concentration_M"] = concentration_by_display[row["display"]]
complexes.append(row)
complexes.sort(key=lambda row: row["concentration_M"], reverse=True)
return {
"workflow": "analysis",
"mode": "tube",
"model": model_summary,
"compute": compute,
"options": options,
"strands": [
{
"name": item["name"],
"sequence": item["sequence"],
"concentration": item["concentration"],
"unit": item["unit"],
"concentration_M": unit_to_molar(item["concentration"], item["unit"]),
}
for item in strands
],
"tube": {
"name": tube.name,
"max_size": max_size,
"complex_concentrations": displayed_concentration_rows,
"total_complex_concentrations": total_complex_count,
},
"complexes": complexes,
"total_complex_count": total_complex_count,
"displayed_complex_count": len(complexes),
}
if mode == "complex":
complexes = parse_complex_lines(payload.get("complexes_text", ""), strand_map)
result = complex_analysis(complexes, model=model, compute=compute, options=nupack_options)
rows = [
serialize_complex_result(
complex_obj,
data,
pairs_preview_size=options["pairs_preview_size"],
)
for complex_obj, data in result.complexes.items()
]
rows.sort(key=lambda row: row["display"])
total_complex_count = len(rows)
rows = rows[: options["result_limit"]]
return {
"workflow": "analysis",
"mode": "complex",
"model": model_summary,
"compute": compute,
"options": options,
"strands": [
{
"name": item["name"],
"sequence": item["sequence"],
}
for item in strands
],
"complexes": rows,
"total_complex_count": total_complex_count,
"displayed_complex_count": len(rows),
}
raise ValueError(f"Unsupported analysis mode: {mode}")
def prune_jobs(now=None):
now = now or time.time()
expired_ids = []
for job_id, job in JOB_STORE.items():
updated_at = job.get("updated_at", job.get("created_at", now))
if now - updated_at > JOB_TTL_SECONDS:
expired_ids.append(job_id)
for job_id in expired_ids:
JOB_STORE.pop(job_id, None)
if len(JOB_STORE) > JOB_MAX_COUNT:
keep_ids = sorted(
JOB_STORE,
key=lambda job_id: JOB_STORE[job_id].get("created_at", 0),
reverse=True,
)[:JOB_MAX_COUNT]
keep_set = set(keep_ids)
for job_id in list(JOB_STORE):
if job_id not in keep_set:
JOB_STORE.pop(job_id, None)
def set_job_data(job):
if redis_enabled():
client = redis_client()
key = job_key(job["job_id"])
encoded = json.dumps(job, ensure_ascii=False)
if job.get("status") in TERMINAL_JOB_STATUSES:
client.setex(key, JOB_TTL_SECONDS, encoded)
else:
# Active jobs may legitimately outlive the result-retention TTL.
client.set(key, encoded)
return
with JOB_LOCK:
prune_jobs(job.get("updated_at"))
JOB_STORE[job["job_id"]] = dict(job)
def get_job_data(job_id, include_payload=False):
if redis_enabled():
raw = redis_client().get(job_key(job_id))
if raw is None:
return None
job = json.loads(raw)
if not include_payload:
job.pop("payload", None)
return job
with JOB_LOCK:
prune_jobs()
job = JOB_STORE.get(job_id)
if job is None:
return None
output = dict(job)
if not include_payload:
output.pop("payload", None)
return output
def should_keep_cancel_state(current, updates):
current_status = current.get("status") if current else None
next_status = updates.get("status")
if current_status in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
return next_status not in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}
return False
def update_job_data(job_id, **updates):
if redis_enabled():
client = redis_client()
key = job_key(job_id)
while True:
pipe = client.pipeline()
try:
pipe.watch(key)
raw = pipe.get(key)
current = json.loads(raw) if raw is not None else {"job_id": job_id, "created_at": time.time()}
if should_keep_cancel_state(current, updates):
pipe.unwatch()
return current
current.update(updates)
current["updated_at"] = time.time()
pipe.multi()
encoded = json.dumps(current, ensure_ascii=False)
if current.get("status") in TERMINAL_JOB_STATUSES:
pipe.setex(key, JOB_TTL_SECONDS, encoded)
else:
pipe.set(key, encoded)
pipe.execute()
return current
except redis.WatchError:
continue
finally:
pipe.reset()
with JOB_LOCK:
current = JOB_STORE.get(job_id)
if current is None:
current = {"job_id": job_id, "created_at": time.time()}
if should_keep_cancel_state(current, updates):
return dict(current)
current.update(updates)
current["updated_at"] = time.time()
prune_jobs(current["updated_at"])
JOB_STORE[job_id] = current
return dict(current)
def create_job(payload, owner):
job_id = uuid4().hex
now = time.time()
job = {
"job_id": job_id,
"user_id": owner["user_id"],
"status": "queued",
"error": None,
"result": None,
"created_at": now,
"updated_at": now,
"payload": payload,
}
ACCOUNT_STORE.create_job(job_id, owner, payload, status="queued", created_at=now)
set_job_data(job)
log_event(f"accepted job_id={job_id} user_id={owner['user_id']} mode={payload.get('mode', 'tube')}")
if redis_enabled():
redis_client().lpush(JOB_QUEUE_KEY, job_id)
else:
thread = threading.Thread(target=_run_job, args=(job_id, payload), daemon=True)
thread.start()
return job_id
def _job_process_entry(payload, result_queue):
apply_thread_limits()
try:
result_queue.put({"status": "success", "result": run_job_payload(payload)})
except BaseException as exc:
result_queue.put(
{
"status": "error",
"error": {
"message": str(exc),
"traceback": traceback.format_exc(),
},
}
)
def terminate_process(process):
if not process.is_alive():
return
if hasattr(process, "kill"):
process.kill()
else:
process.terminate()
process.join(timeout=2)
if process.is_alive():
process.terminate()
process.join(timeout=2)
def _run_job(job_id, payload):
started_at = time.time()
last_heartbeat_at = started_at
current = get_job_data(job_id)
if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
elapsed = round(time.time() - started_at, 3)
update_job_data(job_id, status=CANCELED_STATUS, payload=None, elapsed_seconds=elapsed)
ACCOUNT_STORE.update_job(job_id, CANCELED_STATUS, error={"message": "Job canceled by user."}, elapsed_seconds=elapsed)
return
log_event(f"running job_id={job_id}")
update_job_data(job_id, status="running")
ACCOUNT_STORE.update_job(job_id, "running")
if redis_enabled():
redis_client().sadd(JOB_RUNNING_KEY, job_id)
ctx = multiprocessing.get_context("spawn")
result_queue = ctx.Queue(maxsize=1)
process = ctx.Process(target=_job_process_entry, args=(payload, result_queue), daemon=True)
message = None
try:
process.start()
while process.is_alive():
current = get_job_data(job_id)
if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
terminate_process(process)
elapsed = round(time.time() - started_at, 3)
update_job_data(
job_id,
status=CANCELED_STATUS,
error={"message": "Job canceled by user."},
result=None,
payload=None,
elapsed_seconds=elapsed,
)
ACCOUNT_STORE.update_job(
job_id, CANCELED_STATUS, error={"message": "Job canceled by user."}, elapsed_seconds=elapsed
)
log_event(f"canceled job_id={job_id} elapsed={elapsed}s")
return
now = time.time()
if now - last_heartbeat_at >= JOB_HEARTBEAT_SECONDS:
update_job_data(job_id, heartbeat_at=now)
last_heartbeat_at = now
try:
message = result_queue.get_nowait()
break
except queue.Empty:
time.sleep(0.25)
process.join(timeout=2)
if message is None:
try:
message = result_queue.get_nowait()
except queue.Empty:
message = None
current = get_job_data(job_id)
if current and current.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
elapsed = round(time.time() - started_at, 3)
update_job_data(
job_id,
status=CANCELED_STATUS,
error={"message": "Job canceled by user."},
result=None,
payload=None,
elapsed_seconds=elapsed,
)
ACCOUNT_STORE.update_job(
job_id, CANCELED_STATUS, error={"message": "Job canceled by user."}, elapsed_seconds=elapsed
)
log_event(f"canceled job_id={job_id} elapsed={elapsed}s")
return
elapsed = round(time.time() - started_at, 3)
if message and message.get("status") == "success":
update_job_data(job_id, status="success", result=message["result"], payload=None, elapsed_seconds=elapsed)
ACCOUNT_STORE.update_job(job_id, "success", result=message["result"], elapsed_seconds=elapsed)
log_event(f"success job_id={job_id} elapsed={elapsed}s")
elif message and message.get("status") == "error":
error = message.get("error") or {"message": "Job failed."}
update_job_data(
job_id,
status="error",
error=error,
payload=None,
elapsed_seconds=elapsed,
)
ACCOUNT_STORE.update_job(job_id, "error", error=error, elapsed_seconds=elapsed)
log_event(f"error job_id={job_id} elapsed={elapsed}s message={message.get('error', {}).get('message')}")
else:
error = {"message": f"Job process exited with code {process.exitcode}."}
update_job_data(
job_id,
status="error",
error=error,
payload=None,
elapsed_seconds=elapsed,
)
ACCOUNT_STORE.update_job(job_id, "error", error=error, elapsed_seconds=elapsed)
log_event(f"error job_id={job_id} elapsed={elapsed}s exitcode={process.exitcode}")
finally:
terminate_process(process)
result_queue.close()
result_queue.join_thread()
if redis_enabled():
redis_client().srem(JOB_RUNNING_KEY, job_id)
def cancel_job(job_id):
current = get_job_data(job_id, include_payload=True)
if current is None:
return None
status = current.get("status")
if status in TERMINAL_JOB_STATUSES:
output = dict(current)
output.pop("payload", None)
return output
if redis_enabled() and status == "queued":
redis_client().lrem(JOB_QUEUE_KEY, 0, job_id)
if status == "queued":
updated = update_job_data(
job_id,
status=CANCELED_STATUS,
error={"message": "Job canceled by user."},
result=None,
payload=None,
)
ACCOUNT_STORE.update_job(job_id, CANCELED_STATUS, error={"message": "Job canceled by user."})
else:
updated = update_job_data(
job_id,
status=CANCEL_REQUESTED_STATUS,
error={"message": "Cancellation requested."},
)
ACCOUNT_STORE.update_job(job_id, CANCEL_REQUESTED_STATUS, error={"message": "Cancellation requested."})
updated.pop("payload", None)
return updated
def get_job(job_id):
return get_job_data(job_id, include_payload=False)
def prune_shares(now=None):
if SHARE_MAX_COUNT < 1:
return
if redis_enabled():
client = redis_client()
extra = client.zcard(SHARE_INDEX_KEY) - SHARE_MAX_COUNT
if extra <= 0:
return
stale_ids = client.zrange(SHARE_INDEX_KEY, 0, extra - 1)
if stale_ids:
client.delete(*(share_key(share_id) for share_id in stale_ids))
client.zrem(SHARE_INDEX_KEY, *stale_ids)
return
keep_ids = sorted(
SHARE_STORE,
key=lambda share_id: SHARE_STORE[share_id].get("created_at", 0),
reverse=True,
)[:SHARE_MAX_COUNT]
keep_set = set(keep_ids)
for share_id in list(SHARE_STORE):
if share_id not in keep_set:
SHARE_STORE.pop(share_id, None)
def create_share(record):
payload = record.get("payload")
result = record.get("result")
if not isinstance(payload, dict):
raise ValueError("Share payload must contain an input payload object.")
if not isinstance(result, dict):
raise ValueError("Share payload must contain a result object.")
now = time.time()
share_id = uuid4().hex[:16]
item = {
"id": share_id,
"created_at": now,
"created_at_iso": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
"payload": payload,
"result": result,
"result_summary": record.get("result_summary") or {},
}
if redis_enabled():
client = redis_client()
client.set(share_key(share_id), json.dumps(item, ensure_ascii=False))
client.zadd(SHARE_INDEX_KEY, {share_id: now})
prune_shares(now)
return item
with SHARE_LOCK:
SHARE_STORE[share_id] = item
prune_shares(now)
return dict(item)
def get_share(share_id):
share_id = str(share_id or "").strip()
if not re.fullmatch(r"[0-9a-fA-F]{8,64}", share_id):
return None
if redis_enabled():
raw = redis_client().get(share_key(share_id))
return json.loads(raw) if raw else None
with SHARE_LOCK:
item = SHARE_STORE.get(share_id)
return dict(item) if item else None
def recover_interrupted_jobs():
client = redis_client()
recovered = 0
canceled = 0
for job_id in client.smembers(JOB_RUNNING_KEY):
raw = client.get(job_key(job_id))
if raw is None:
client.srem(JOB_RUNNING_KEY, job_id)
continue
job = json.loads(raw)
status = job.get("status")
if status == "running":
client.lrem(JOB_QUEUE_KEY, 0, job_id)
update_job_data(
job_id,
status="queued",
recovered_at=time.time(),
recovery_count=int(job.get("recovery_count", 0)) + 1,
)
client.lpush(JOB_QUEUE_KEY, job_id)
ACCOUNT_STORE.update_job(job_id, "queued")
recovered += 1
elif status == CANCEL_REQUESTED_STATUS:
update_job_data(
job_id,
status=CANCELED_STATUS,
error={"message": "Job canceled while the worker was restarting."},
result=None,
payload=None,
)
ACCOUNT_STORE.update_job(
job_id,
CANCELED_STATUS,
error={"message": "Job canceled while the worker was restarting."},
)
canceled += 1
client.srem(JOB_RUNNING_KEY, job_id)
if recovered or canceled:
log_event(f"recovered jobs queued={recovered} canceled={canceled}")
def run_worker_loop():
if not redis_enabled():
raise RuntimeError("Worker mode requires NP_REDIS_URL and the redis package.")
client = redis_client()
recover_interrupted_jobs()
worker_count = max(1, WORKER_CONCURRENCY)
log_event(
f"Starting worker loop on Redis queue {JOB_QUEUE_KEY} "
f"(concurrency={worker_count}, per_job_threads={max(1, PER_JOB_THREAD_LIMIT)})"
)
with ThreadPoolExecutor(max_workers=worker_count, thread_name_prefix="np-job") as executor:
inflight = set()
while True:
finished = {future for future in inflight if future.done()}
if finished:
inflight.difference_update(finished)
for future in finished:
future.result()
if len(inflight) >= worker_count:
done, _ = wait(inflight, return_when=FIRST_COMPLETED, timeout=1)
inflight.difference_update(done)
for future in done:
future.result()
continue
item = client.brpop(JOB_QUEUE_KEY, timeout=2)
if not item:
continue
_, job_id = item
raw = client.get(job_key(job_id))
if raw is None:
continue
job = json.loads(raw)
if job.get("status") in {CANCEL_REQUESTED_STATUS, CANCELED_STATUS}:
continue
payload = job.get("payload")
if payload is None:
continue
inflight.add(executor.submit(_run_job, job_id, payload))
EXAMPLE_PAYLOAD = {
"workflow": "analysis",
"mode": "tube",
"model": {
"material": "rna",
"ensemble": "stacking",
"celsius": 37,
"sodium": 1.0,
"magnesium": 0.0,
},
"compute": ["pfunc", "mfe", "pairs"],
"options": {
"num_sample": 20,
"energy_gap": 1.0,
"sparsity_fraction": 1.0,
"sparsity_threshold": 0.0,
"single_mfe": False,
"result_limit": 25,
"pairs_preview_size": 24,
},
"strands": [
{"name": "A", "sequence": "AGUCUAGGAU", "concentration": 1.0, "unit": "uM"},
{"name": "B", "sequence": "UUAACCCACG", "concentration": 2.0, "unit": "uM"},
],
"tube": {"name": "tube1", "max_size": 2},
"complexes_text": "A\nB\nA+B",
"design": {
"trials": 1,
"result_limit": 25,
"off_target_max_size": 2,
"stop_condition": 0.02,
"seed": 0,
"wobble_mutations": False,
"max_time_seconds": 0,
},
"design_domains": [
{"name": "a", "sequence": "N10"},
{"name": "b", "sequence": "N10"},
],
"design_complexes": [
{
"name": "AB_target",
"strands": "A+B",
"structure": "(10+)10",
}
],
"design_tubes": [
{
"name": "Tube 1",
"max_size": 2,
"on_targets": [
{"complex": "AB_target", "concentration": 1.0, "unit": "uM"},
],
}
],
"hard_constraints": [],
"soft_constraints": [],
"defect_weights": [],
}
# Use A=a and B=~a so the full-duplex target "(10+)10" is base-pair consistent.
DESIGN_TUBE_EXAMPLE_PAYLOAD = {
"workflow": "design",
"mode": "tube",
"model": {
"material": "rna",
"ensemble": "stacking",
"celsius": 37,
"sodium": 1.0,
"magnesium": 0.0,
},
"compute": ["pfunc", "mfe", "pairs"],
"options": {
"num_sample": 20,
"energy_gap": 1.0,
"sparsity_fraction": 1.0,
"sparsity_threshold": 0.0,
"single_mfe": False,
"result_limit": 25,
"pairs_preview_size": 24,
},
"strands": [
{"name": "A", "sequence": "a", "concentration": 1.0, "unit": "uM"},
{"name": "B", "sequence": "~a", "concentration": 1.0, "unit": "uM"},
],
"tube": {"name": "Design Tube", "max_size": 2},
"complexes_text": "A+B",
"design": {
"trials": 1,
"result_limit": 25,
"off_target_max_size": 2,
"stop_condition": 0.05,
"seed": 1,
"wobble_mutations": False,
"max_time_seconds": 0,
},
"design_domains": [
{"name": "a", "sequence": "N10"},
],
"design_complexes": [
{
"name": "AB_target",
"strands": "A+B",
"structure": "(10+)10",
}
],
"design_tubes": [
{
"name": "Design Tube",
"max_size": 2,
"on_targets": [
{"complex": "AB_target", "concentration": 1.0, "unit": "uM"},
],
}
],
"hard_constraints": [],
"soft_constraints": [],
"defect_weights": [],
}
DESIGN_COMPLEX_EXAMPLE_PAYLOAD = {
**DESIGN_TUBE_EXAMPLE_PAYLOAD,
"mode": "complex",
"tube": {"name": "Design Tube", "max_size": 1},
"design": {
**DESIGN_TUBE_EXAMPLE_PAYLOAD["design"],
"off_target_max_size": 1,
},
"design_tubes": [],
}
def get_example_payload(query):
params = parse_qs(query)
workflow = (params.get("workflow") or ["analysis"])[0]
mode = (params.get("mode") or ["tube"])[0]
if workflow == "design":
if mode == "complex":
return DESIGN_COMPLEX_EXAMPLE_PAYLOAD
return DESIGN_TUBE_EXAMPLE_PAYLOAD
return EXAMPLE_PAYLOAD
class AppHandler(BaseHTTPRequestHandler):
server_version = "NPReplica/0.1"
def _session(self):
if not hasattr(self, "_cached_session"):
self._cached_session = OIDC_AUTH.current_session(self.headers)
if self._cached_session:
ACCOUNT_STORE.upsert_user(self._cached_session["user"])
return self._cached_session
def _require_user(self):
session = self._session()
if session is None:
self._respond(*json_bytes({"error": "Authentication required", "login_url": "/auth/login"}, status=HTTPStatus.UNAUTHORIZED))
return None
return session["user"]
def _read_json(self, max_bytes=16 * 1024 * 1024):
length = int(self.headers.get("Content-Length", "0"))
if length < 0 or length > max_bytes:
raise ValueError(f"Request payload is too large. Limit is {max_bytes} bytes.")
return json.loads(self.rfile.read(length).decode("utf-8"))
def _redirect(self, location, *, cookie=None):
headers = {"Location": location}
if cookie:
headers["Set-Cookie"] = cookie
self._respond(HTTPStatus.FOUND, "text/plain; charset=utf-8", b"Redirecting", extra_headers=headers)
def _require_page_user(self, next_path):
if self._session() is not None:
return True
self._redirect(f"/auth/login?{urlencode({'next': next_path})}")
return False
def _owned_job(self, user, job_id, include_content=True):
if ACCOUNT_STORE.owner_id(job_id) != user["user_id"]:
return None
live = get_job(job_id)
if live is not None:
live.pop("user_id", None)
return live
return ACCOUNT_STORE.get_job(user["user_id"], job_id, include_content=include_content)
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/auth/login":
next_path = (parse_qs(parsed.query).get("next") or ["/"])[0]
try:
self._redirect(OIDC_AUTH.begin_login(next_path))
except Exception as exc:
self._respond(*json_bytes({"error": f"Unable to start login: {exc}"}, status=HTTPStatus.BAD_GATEWAY))
return
if parsed.path == "/auth/callback":
try:
session_id, user, next_path = OIDC_AUTH.complete_login(parse_qs(parsed.query))
ACCOUNT_STORE.upsert_user(user)
self._redirect(next_path, cookie=OIDC_AUTH.cookie_header(session_id))
except Exception as exc:
log_event(f"OIDC callback failed: {exc}")
self._respond(*json_bytes({"error": f"Login failed: {exc}"}, status=HTTPStatus.BAD_REQUEST))
return
if parsed.path == "/auth/logout":
session = self._session()
try:
location = OIDC_AUTH.logout_url(session)
except Exception:
location = "/"
OIDC_AUTH.delete_session(session)
self._redirect(location, cookie=OIDC_AUTH.clear_cookie_header())
return
share_page = re.fullmatch(r"/share/([A-Za-z0-9_-]{8,128})", parsed.path)
if share_page:
self._respond_file(INDEX_PATH, cache_control="public, max-age=60, stale-while-revalidate=86400")
return
legacy_share = (parse_qs(parsed.query).get("share") or [""])[0]
if parsed.path == "/" and re.fullmatch(r"[A-Za-z0-9_-]{8,128}", legacy_share):
self._respond_file(INDEX_PATH, cache_control="public, max-age=60, stale-while-revalidate=86400")
return
if parsed.path == "/":
if not self._require_page_user(self.path):
return
self._respond_file(INDEX_PATH, cache_control="private, no-cache")
return
if parsed.path in {"/favicon.svg", "/favicon.ico"}:
self._respond_file(FAVICON_PATH, cache_control="public, max-age=86400")
return
if parsed.path == "/design-guide.html":
if not self._require_page_user(self.path):
return
self._respond_file(GUIDE_PATH, cache_control="public, max-age=3600")
return
if parsed.path == "/health":
stats = job_stats()
self._respond(
*json_bytes(
{
"status": "ok",
"job_backend": "redis" if redis_enabled() else "memory",
"run_mode": RUN_MODE,
"queue_depth": queue_size(),
"jobs_running": stats["running"],
"jobs_queued": stats["queued"],
"job_ttl_seconds": JOB_TTL_SECONDS,
"worker_concurrency": max(1, WORKER_CONCURRENCY),
"per_job_thread_limit": max(1, PER_JOB_THREAD_LIMIT),
"nupack_threads": int(getattr(nupack_config, "threads", PER_JOB_THREAD_LIMIT)),
"nupack_cache_gb": float(getattr(nupack_config, "cache", NUPACK_CACHE_GB)),
}
)
)
return
if parsed.path == "/api/me":
user = self._require_user()
if user is None:
return
self._respond(*json_bytes({"status": "success", "user": user, "usage": ACCOUNT_STORE.usage(user["user_id"])}))
return
if parsed.path == "/api/history":
user = self._require_user()
if user is None:
return
params = {key: values[0] for key, values in parse_qs(parsed.query).items() if values}
self._respond(*json_bytes({"status": "success", **ACCOUNT_STORE.list_jobs(user["user_id"], params)}))
return
if parsed.path.startswith("/api/history/"):
user = self._require_user()
if user is None:
return
job_id = parsed.path.rsplit("/", 1)[-1]
job = ACCOUNT_STORE.get_job(user["user_id"], job_id, include_content=True)
if job is None:
self._respond(*json_bytes({"error": "History record not found"}, status=HTTPStatus.NOT_FOUND))
return
self._respond(*json_bytes({"status": "success", "item": job}))
return
if parsed.path == "/api/account/shares":
user = self._require_user()
if user is None:
return
self._respond(*json_bytes({"status": "success", "items": ACCOUNT_STORE.list_shares(user["user_id"])}))
return
if parsed.path == "/api/example":
user = self._require_user()
if user is None:
return
self._respond(*json_bytes(get_example_payload(parsed.query)))
return
if parsed.path.startswith("/api/jobs/"):
user = self._require_user()
if user is None:
return
job_id = parsed.path.rsplit("/", 1)[-1]
job = self._owned_job(user, job_id)
if job is None:
self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
return
self._respond(*json_bytes(job))
return
if parsed.path.startswith("/api/shares/"):
share_id = parsed.path.rsplit("/", 1)[-1]
share = ACCOUNT_STORE.resolve_share(share_id) or get_share(share_id)
if share is None:
self._respond(*json_bytes({"error": "Share not found"}, status=HTTPStatus.NOT_FOUND))
return
self._respond(*json_bytes({"status": "success", "share": share}))
return
self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
def do_POST(self):
parsed = urlparse(self.path)
user = self._require_user()
if user is None:
return
try:
if parsed.path.startswith("/api/jobs/") and parsed.path.endswith("/cancel"):
job_id = parsed.path.split("/")[-2]
if ACCOUNT_STORE.owner_id(job_id) != user["user_id"]:
self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
return
job = cancel_job(job_id)
if job is None:
self._respond(*json_bytes({"error": "Job not found"}, status=HTTPStatus.NOT_FOUND))
return
job.pop("user_id", None)
self._respond(*json_bytes({"status": "success", "job": job}))
return
if parsed.path.startswith("/api/jobs/") and parsed.path.endswith("/shares"):
job_id = parsed.path.split("/")[-2]
body = self._read_json()
expires_in = body.get("expires_in")
if expires_in not in {None, "", 0}:
expires_in = int(expires_in)
if expires_in < 60:
raise ValueError("Share duration must be at least 60 seconds.")
else:
expires_in = None
share = ACCOUNT_STORE.create_share(user["user_id"], job_id, expires_in)
self._respond(*json_bytes({"status": "success", "share": share, "url": f"/share/{share['share_id']}"}, status=HTTPStatus.CREATED))
return
if parsed.path == "/api/jobs":
payload = self._read_json()
job_id = create_job(payload, user)
self._respond(*json_bytes({"status": "accepted", "job_id": job_id}, status=HTTPStatus.ACCEPTED))
return
if parsed.path == "/api/history/import":
body = self._read_json(max_bytes=64 * 1024 * 1024)
entries = body.get("history") if isinstance(body, dict) else None
if not isinstance(entries, list):
raise ValueError("History import requires a history array.")
imported = ACCOUNT_STORE.import_history(user, entries)
self._respond(*json_bytes({"status": "success", "imported": imported}))
return
if parsed.path == "/api/shares":
body = self._read_json()
job_id = str(body.get("job_id") or "")
if not job_id:
raise ValueError("Cloud shares require a job_id.")
share = ACCOUNT_STORE.create_share(user["user_id"], job_id, body.get("expires_in"))
self._respond(*json_bytes({"status": "success", "share_id": share["share_id"], "url": f"/share/{share['share_id']}"}, status=HTTPStatus.CREATED))
return
if parsed.path == "/api/analyze":
payload = self._read_json()
job_id = uuid4().hex
started = time.time()
ACCOUNT_STORE.create_job(job_id, user, payload, status="running", created_at=started)
try:
result = run_job_payload(payload)
except Exception as exc:
elapsed = round(time.time() - started, 3)
ACCOUNT_STORE.update_job(job_id, "error", error={"message": str(exc), "traceback": traceback.format_exc()}, elapsed_seconds=elapsed)
raise
elapsed = round(time.time() - started, 3)
ACCOUNT_STORE.update_job(job_id, "success", result=result, elapsed_seconds=elapsed)
self._respond(*json_bytes({"status": "success", "job_id": job_id, "result": result}))
return
self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
except Exception as exc:
self._respond(
*json_bytes(
{
"status": "error",
"error": str(exc),
"traceback": traceback.format_exc(),
},
status=HTTPStatus.BAD_REQUEST,
)
)
def do_PATCH(self):
parsed = urlparse(self.path)
user = self._require_user()
if user is None:
return
try:
if parsed.path.startswith("/api/account/shares/"):
share_id = parsed.path.rsplit("/", 1)[-1]
body = self._read_json()
expires_in = body["expires_in"] if "expires_in" in body else "unchanged"
share = ACCOUNT_STORE.update_share(
user["user_id"], share_id, active=body.get("active"), expires_in=expires_in
)
if share is None:
self._respond(*json_bytes({"error": "Share not found"}, status=HTTPStatus.NOT_FOUND))
return
self._respond(*json_bytes({"status": "success", "share": share}))
return
self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
except Exception as exc:
self._respond(*json_bytes({"status": "error", "error": str(exc)}, status=HTTPStatus.BAD_REQUEST))
def do_DELETE(self):
parsed = urlparse(self.path)
user = self._require_user()
if user is None:
return
try:
if parsed.path.startswith("/api/history/"):
job_id = parsed.path.rsplit("/", 1)[-1]
if not ACCOUNT_STORE.delete_job(user["user_id"], job_id):
self._respond(*json_bytes({"error": "History record not found"}, status=HTTPStatus.NOT_FOUND))
return
self._respond(*json_bytes({"status": "success"}))
return
if parsed.path.startswith("/api/account/shares/"):
share_id = parsed.path.rsplit("/", 1)[-1]
share = ACCOUNT_STORE.update_share(user["user_id"], share_id, active=False)
if share is None:
self._respond(*json_bytes({"error": "Share not found"}, status=HTTPStatus.NOT_FOUND))
return
self._respond(*json_bytes({"status": "success", "share": share}))
return
self._respond(*json_bytes({"error": "Not found"}, status=HTTPStatus.NOT_FOUND))
except Exception as exc:
self._respond(*json_bytes({"status": "error", "error": str(exc)}, status=HTTPStatus.BAD_REQUEST))
def log_message(self, format_, *args):
print(f"{self.address_string()} - {format_ % args}")
def _respond_file(self, path, cache_control):
etag = file_etag(path)
if self.headers.get("If-None-Match") == etag:
self.send_response(HTTPStatus.NOT_MODIFIED)
self.send_header("ETag", etag)
self.send_header("Cache-Control", cache_control)
self.end_headers()
return
status, content_type, body = html_bytes(path)
self._respond(status, content_type, body, cache_control=cache_control, extra_headers={"ETag": etag})
def _respond(self, status, content_type, body, cache_control=None, extra_headers=None):
self.send_response(status)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", cache_control or "no-store, max-age=0")
if cache_control is None:
self.send_header("Pragma", "no-cache")
self.send_header("Expires", "0")
for key, value in (extra_headers or {}).items():
self.send_header(key, value)
self.end_headers()
self.wfile.write(body)
def main():
apply_thread_limits()
ACCOUNT_STORE.initialize()
if RUN_MODE == "worker":
run_worker_loop()
return
server = ThreadingHTTPServer((HOST, PORT), AppHandler)
log_event(
f"Serving NP replica on http://{HOST}:{PORT} "
f"(jobs via {'redis' if redis_enabled() else 'memory'}, "
f"per_job_threads={max(1, PER_JOB_THREAD_LIMIT)})"
)
server.serve_forever()
if __name__ == "__main__":
main()
Reference in a new issue View git blame Copy permalink