File size: 6,228 Bytes
9ae68e3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | # semantics.py
import math
import re
STAN_INT_MAX = 2_147_483_647
FIELDS = {
"observation_id", "sample_id", "gene_id", "condition",
"timepoint", "replicate", "batch", "cell_type", "count",
"missing_reason", "exposure", "normalized_expression",
"normalization", "measurement_quality",
}
def _require(ok, message):
if not ok:
raise ValueError(message)
def _identifier(value):
return isinstance(value, str) and bool(value.strip())
def _finite_number(value):
if type(value) not in (int, float):
return False
try:
return math.isfinite(value)
except OverflowError:
return False
def validate_dataset(dataset, conditions):
"""Validate without modifying, imputing, or coercing input.
Null metadata stays null. Native adapters must explicitly exclude
records missing required model metadata, preserving an exclusion map.
Normalization is derived; counts are observed, unless synthetic.
Source-byte hashes belong in the pipeline artifact manifest.
"""
_require(
isinstance(conditions, (list, tuple))
and len(conditions) == 2
and all(_identifier(x) for x in conditions)
and conditions[0] != conditions[1],
"config requires two distinct explicit conditions",
)
_require(isinstance(dataset, dict), "dataset must be an object")
_require(
{"dataset_id", "synthetic", "observations"} <= dataset.keys(),
"missing dataset fields",
)
_require(_identifier(dataset["dataset_id"]), "invalid dataset_id")
_require(type(dataset["synthetic"]) is bool, "synthetic must be boolean")
_require(isinstance(dataset["observations"], list), "observations must be a list")
seen = set()
for index, row in enumerate(dataset["observations"]):
prefix = f"observations[{index}]"
_require(isinstance(row, dict), f"{prefix}: expected object")
_require(FIELDS <= row.keys(), f"{prefix}: missing required fields")
for field in ("observation_id", "sample_id", "gene_id"):
_require(_identifier(row[field]), f"{prefix}: invalid {field}")
identity = row["observation_id"]
_require(identity not in seen, f"{prefix}: duplicate observation_id")
seen.add(identity)
condition = row["condition"]
_require(
condition is None
or (isinstance(condition, str) and condition in conditions),
f"{prefix}: unknown condition",
)
for field in ("replicate", "batch", "cell_type"):
_require(
row[field] is None or _identifier(row[field]),
f"{prefix}: {field} must be null or nonempty string",
)
_require(
row["timepoint"] is None or _finite_number(row["timepoint"]),
f"{prefix}: timepoint must be null or finite number",
)
count, reason = row["count"], row["missing_reason"]
if count is None:
_require(_identifier(reason), f"{prefix}: missing count requires reason")
else:
_require(reason is None, f"{prefix}: present count cannot have missing reason")
_require(
isinstance(count, str)
and re.fullmatch(r"(?:0|[1-9][0-9]*)", count) is not None,
f"{prefix}: count must be canonical unsigned decimal string",
)
limit = str(STAN_INT_MAX)
_require(
len(count) < len(limit)
or (len(count) == len(limit) and count <= limit),
f"{prefix}: count exceeds Stan integer range",
)
exposure = row["exposure"]
_require(
_finite_number(exposure) and exposure > 0,
f"{prefix}: exposure must be positive and finite",
)
for field in ("normalized_expression", "measurement_quality"):
_require(
row[field] is None or _finite_number(row[field]),
f"{prefix}: {field} must be null or finite number",
)
normalization = row["normalization"]
if row["normalized_expression"] is not None:
_require(normalization is not None, f"{prefix}: normalization provenance required")
if normalization is not None:
_require(
isinstance(normalization, dict)
and {"method", "source_ids"} <= normalization.keys(),
f"{prefix}: invalid normalization provenance",
)
_require(_identifier(normalization["method"]), f"{prefix}: missing normalization method")
sources = normalization["source_ids"]
_require(
isinstance(sources, list)
and bool(sources)
and all(_identifier(s) for s in sources),
f"{prefix}: normalization source_ids required",
)
_require(len(set(sources)) == len(sources), f"{prefix}: duplicate source_ids")
return dataset
# Fixed PTM: input alphabet {0,1}; tape alphabet {0,1,_}.
# Output: final parity bit in the first formerly blank cell.
# q_accept and q_reject halt; malformed encodings reject before a transition.
DELTA = {
("q_even", "0"): ("q_even", "0", 1),
("q_even", "1"): ("q_odd", "1", 1),
("q_odd", "0"): ("q_odd", "0", 1),
("q_odd", "1"): ("q_even", "1", 1),
("q_even", "_"): ("q_accept", "0", 0),
("q_odd", "_"): ("q_accept", "1", 0),
}
def ptm_parity(bits):
if not isinstance(bits, str) or any(bit not in "01" for bit in bits):
return {
"status": "rejected", "state": "q_reject",
"reason": "invalid input encoding", "transitions": 0,
}
tape = list(bits) + ["_"]
state, head, transitions = "q_even", 0, 0
while state != "q_accept":
state, write, move = DELTA[(state, tape[head])]
tape[head] = write
head += move
transitions += 1
return {
"status": "accepted",
"state": state,
"output": tape[head],
"output_cell": head,
"encoded_input_bits": len(bits),
"transitions": transitions,
"total_tape_cells": len(tape),
"auxiliary_tape_cells": 1,
}
|