leonarski_fandClaude Opus 5.5
8e134c405d
For demonstrating and testing a collection followed while it is written: - jfjoch_stream2_replay replays a stored dataset as a DECTRIS detector's stream2 (ZeroMQ PUSH on the stream2 port): start message, every image at a given rate, end message. - fake_simplon.py stands in for the detector's SIMPLON control interface: answers the configuration and status requests the broker makes, accepts every command, reads back idle. It needs port 80. Not installed. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SVmAWnzCmRKAXVUCdc4iNi
90 lines
3.7 KiB
Python
90 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
# SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
|
|
# SPDX-License-Identifier: GPL-3.0-only
|
|
"""A stand-in for a DECTRIS detector's SIMPLON control interface, for running jfjoch_broker without a
|
|
detector (demonstrations and tests). It answers the configuration and status requests the broker makes,
|
|
accepts every command, and always reads back as idle. The images come from tools/jfjoch_stream2_replay,
|
|
which plays the detector's stream2.
|
|
|
|
fake_simplon.py --width 2068 --height 2164 --pixel-size 75e-6 [--port 80]
|
|
|
|
The broker reaches SIMPLON on port 80, so this has to be able to bind it (e.g. run it as root in a
|
|
container on the host network).
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
p.add_argument("--width", type=int, required=True)
|
|
p.add_argument("--height", type=int, required=True)
|
|
p.add_argument("--pixel-size", type=float, required=True, help="metres")
|
|
p.add_argument("--sensor-thickness", type=float, default=320e-6, help="metres")
|
|
p.add_argument("--sensor-material", default="Si")
|
|
p.add_argument("--bit-depth", type=int, default=32)
|
|
p.add_argument("--description", default="Replayed detector")
|
|
p.add_argument("--port", type=int, default=80)
|
|
a = p.parse_args()
|
|
|
|
mask = base64.b64encode(bytes(4 * a.width * a.height)).decode()
|
|
config = {
|
|
"description": {"value": a.description},
|
|
"detector_number": {"value": "replay-1"},
|
|
"x_pixel_size": {"value": a.pixel_size},
|
|
"y_pixel_size": {"value": a.pixel_size},
|
|
"sensor_material": {"value": a.sensor_material},
|
|
"sensor_thickness": {"value": a.sensor_thickness},
|
|
"bit_depth_image": {"value": a.bit_depth},
|
|
"x_pixels_in_detector": {"value": a.width},
|
|
"y_pixels_in_detector": {"value": a.height},
|
|
"frame_time": {"value": 0.01, "min": 0.0001},
|
|
"count_time": {"value": 0.01, "min": 0.0001},
|
|
"detector_readout_time": {"value": 0.0},
|
|
"threshold_energy": {"value": 4000.0, "min": 1000.0},
|
|
"photon_energy": {"value": 8000.0},
|
|
"nimages": {"value": 1},
|
|
"ntrigger": {"value": 1},
|
|
"pixel_mask": {"value": {"shape": [a.height, a.width], "type": "uint32", "data": mask}},
|
|
}
|
|
status = {"state": {"value": "idle"}, "humidity": {"value": 0.0}, "temperature": {"value": 25.0}}
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
def reply(self, body):
|
|
data = json.dumps(body).encode()
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
parts = self.path.strip("/").split("/") # module/api/version/task/key
|
|
key = parts[-1]
|
|
if len(parts) >= 5 and parts[3] == "status":
|
|
self.reply(status.get(key, {"value": 0}))
|
|
else:
|
|
self.reply(config.get(key, {"value": 0}))
|
|
|
|
def do_POST(self):
|
|
parts = self.path.strip("/").split("/")
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
body = self.rfile.read(length) if length else b""
|
|
if len(parts) >= 5 and parts[3] == "config" and body:
|
|
config.setdefault(parts[-1], {})["value"] = json.loads(body)["value"]
|
|
self.reply([])
|
|
|
|
do_PUT = do_POST
|
|
|
|
def log_message(self, fmt, *args):
|
|
pass
|
|
|
|
ThreadingHTTPServer(("0.0.0.0", a.port), Handler).serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|