add new arroyo listener

[xiaoyachong]

1. add new arroyo listener according to https://app.asana.com/1/4262699445529/project/1210776618867217/task/1210805079681380?focus=true

2. add a results publisher:
   https://app.asana.com/1/4262699445529/project/1210776618867217/task/1210827537370723?focus=true

3. add websocket simulator for the new websocket listener
   https://app.asana.com/1/4262699445529/project/1210776618867217/task/1210827537370730?focus=true

[dylanmcreynolds]

Let's remove...api compatibility for the init is not really and issue here.

[xiaoyachong]

Thanks. I remove poll_pause_sec from the class parameter list.

[dylanmcreynolds]

I'm not sure I understand why we would want to process a single run in isolation. For testing?

[xiaoyachong]

I removed single_run from the class parameter list and also removed the _process_single_run method.

[dylanmcreynolds]

just so you know, these are likely to be pretty different when we get the real interface. "new_frame" will be an "event" that we have to do a fair amount of processing with

[xiaoyachong]

Thanks for the reminder. I will update them once we have the real interface. Let’s take a closer look tomorrow.

[dylanmcreynolds]

This raises an interesting architectural quesion.

Right now, we have Tiled WS sending events to one listener which has one publisher. Then, we'll be having multiple ZMQ subscribers listening to that publisher.

Our system could be made easier by eliminating the zmq altogether and having the LSEOperator and VizOperator both establishing their own web socket connects to tiled.

For now I guess it makes sense to keep it as is, since we'll be using hardware in Berkeley, and it means less data will come over the wire.

[xiaoyachong]

Sure, let’s leave it as is for now and address it in a future PR.

[dylanmcreynolds]

As a general practice, I would love to see type hints for all methods. I know that my code that you read doesn't have return types (it should) but, especially for framework code like this, I would really like to type hint everything.

[xiaoyachong]

@dylanmcreynolds Hi Dylan, I still have some questions about the newly added tiled_websocket.py file. I've listed them below:

Remaining Issues with new tiled_websocket.py

a. Uncertainty About When to Call publish_stop()
It's unclear whether and when publish_stop() should be triggered, since no clear "stop" signal can be inferred from the tiled_event_logs/xx/*.json files.

def publish_stop(self, num_frames: int = 0) -> None:
    """
    Publish a stop event to the operator indicating the end of a run.
    
    Args:
        num_frames: Number of frames processed in this run
    """
    stop_event = SASStop(num_frames=num_frames)
    self.send_to_operator(stop_event)
    logger.info(f"Published stop event: num_frames={num_frames}")

b. Placement of publish_event()
Currently, publish_event() is called in on_node_in_stream(), but it likely belongs at the end of on_event()

c. Incomplete publish_start() and publish_event() Implementations
The parameter lists are currently filled with placeholders. These should be populated with real values, but it's unclear where to extract them from—the corresponding .json files do not seem to contain sufficient information.

def publish_start(self, data: Dict[str, Any]) -> None:
    """
    Publish a start event to the operator.
    """
    run_id = data.get("key", "")

    start = SASStart(
        run_name=f"Run-{run_id}",     # TODO: Update with descriptive name if available
        run_id=run_id,                # Extracted from on_new_run_0001.json
        width=0,                      # Extracted from on_node_in_stream_0003.json
        height=0,                     # Extracted from on_node_in_stream_0003.json
        data_type="unknown",          # TODO: Update with actual data type if available
        tiled_url=""                  # TODO: Update with full URL if available
    )

    self.send_to_operator(start)

Question 1: Since width and height come from on_node_in_stream_0003.json, and run_id comes from on_new_run_0001.json, we might want to delay the call to publish_start() until we have all the metadata—perhaps move it to on_node_in_stream() if that's the earliest place we can access the image dimensions.

def publish_event(self, data: Dict[str, Any]) -> None:
    """
    Publish a frame event to the operator.
    """
    event = RawFrameEvent(
        image=None,                            # TODO: Add image data when available
        frame_number=data.get("sequence", 0),  # TODO: Consider using sequence - 1 if indexing starts at 1
        tiled_url=data.get("tiled_url", "")    # TODO: Provide full tiled URL when available
    )

    self.send_to_operator(event)

Question 2: It's unclear whether frame_number should use 0-based or 1-based indexing. The .json files appear to use 1-based indexing, but it's not confirmed whether the event expects 0-based or 1-based indexing.

Question3: Does tiled_url follow the format https://tiled.nsls2.bnl.gov/api/v1/metadata/smi/raw/28b16c3a-5ec4-4c83-8e2e-7e52df303914/primary/data/pil1M_image?slice=frame_number? The on_event_xxxx.json file only contains "data_uri": "file://localhost/nsls2/data/smi/proposals/2025-2/pass-317903/assets/pilatus2m-1/2025/07/17/40e32478-9601-4298-a3af_000004.tiff", which seems to be a local file path rather than a Tiled URL.

[dylanmcreynolds]

publish_stop

The tiled websocket service does not yet publish a stop event, so we have nothing to go on. This is fine for now, our application can get by with users refreshing the screen between scans.

publish_event

Good catch. Let's create a new PR?

incomplete

Yeah, we're going to have to make somethings up. It was hard to do any of this without the tiled websocket simulator.

Question 1

Makes sense.

Question 2

Good catch. I don't really know. I think we're go with the exact index that tiled gives us so that we can easily match.

Question 3

This is still a work in progress on the tiled websocket server, so we don't really know yet. We should find out Monday or Tuesday.