fix: delete pixelmap json, make it a method
This commit was merged in pull request #10.
This commit is contained in:
-524309
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,26 @@
|
||||
"""Utilities for building Timepix pixel-map payloads in Python."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import PixelMap
|
||||
|
||||
DEFAULT_TIMEPIX_CHIPS = 8
|
||||
DEFAULT_PIXELS_PER_CHIP = 256 * 256
|
||||
|
||||
|
||||
def create_single_energy_per_chip_pixel_map(
|
||||
num_chips: int = DEFAULT_TIMEPIX_CHIPS,
|
||||
pixels_per_chip: int = DEFAULT_PIXELS_PER_CHIP,
|
||||
) -> PixelMap:
|
||||
"""Create a pixel map where each chip maps fully to its own energy bin."""
|
||||
if num_chips <= 0:
|
||||
raise ValueError("num_chips must be a positive integer.")
|
||||
if pixels_per_chip <= 0:
|
||||
raise ValueError("pixels_per_chip must be a positive integer.")
|
||||
|
||||
return PixelMap(
|
||||
chips=[
|
||||
[{"i": pixel_index, "p": [chip_index], "f": [1.0]} for pixel_index in range(pixels_per_chip)]
|
||||
for chip_index in range(num_chips)
|
||||
]
|
||||
)
|
||||
@@ -34,17 +34,16 @@ from ophyd_devices.devices.areadetector.plugins import HDF5Plugin_V35, ImagePlug
|
||||
from ophyd_devices.interfaces.base_classes.psi_device_base import PSIDeviceBase
|
||||
from typeguard import typechecked
|
||||
|
||||
import superxas_bec.devices.timepix.default_pixel_maps as _default_pixel_maps
|
||||
from superxas_bec.devices.timepix.pixel_map_utils import create_single_energy_per_chip_pixel_map
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_backend import TimepixFlyBackend
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_client import TimePixFlyStatus
|
||||
from superxas_bec.devices.timepix.timepix_fly_client.timepix_fly_interface import (
|
||||
OtherConfigModel,
|
||||
PixelMap,
|
||||
)
|
||||
from superxas_bec.devices.timepix.utils import AndStatusWithList
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from bec_lib.messages import DevicePreviewMessage, ScanStatusMessage
|
||||
from bec_lib.messages import ScanStatusMessage
|
||||
|
||||
|
||||
logger = bec_logger.logger
|
||||
@@ -178,10 +177,6 @@ class TimePixControl(ADBase):
|
||||
hdf = Cpt(HDF5Plugin_Timepix, "HDF1:")
|
||||
|
||||
|
||||
DEFAULT_PIXEL_MAP = os.path.join(
|
||||
os.path.dirname(_default_pixel_maps.__file__), "timepix_8_chips_single_energy_per_chip.json"
|
||||
)
|
||||
|
||||
DETECTOR_SHAPE = (512, 1024) # Shape of the TimePix detector
|
||||
|
||||
|
||||
@@ -544,16 +539,14 @@ class Timepix(PSIDeviceBase, TimePixControl):
|
||||
"""Get the current pixel map of the TimePix detector."""
|
||||
if self._pixel_map is None:
|
||||
try:
|
||||
pixel_map = load_pixel_map_from_json(DEFAULT_PIXEL_MAP)
|
||||
pixel_map = create_single_energy_per_chip_pixel_map()
|
||||
self._pixel_map = pixel_map
|
||||
# pylint: disable=broad-except
|
||||
# pylint: disable=raise-missing-from
|
||||
except Exception:
|
||||
content = traceback.format_exc()
|
||||
logger.error(f"Failed to load default pixel map: {content}")
|
||||
raise ValueError(
|
||||
f"Failed to load default pixel map from {DEFAULT_PIXEL_MAP}: {content}"
|
||||
)
|
||||
raise ValueError(f"Failed to generate default pixel map: {content}")
|
||||
return self._pixel_map
|
||||
|
||||
@pixel_map.setter
|
||||
@@ -599,10 +592,10 @@ class Timepix(PSIDeviceBase, TimePixControl):
|
||||
set default values on signals, please use on_connected instead.
|
||||
"""
|
||||
start_time = time.time()
|
||||
logger.info(f"Loading default pixel map for TimePix detector {self.name}...")
|
||||
self.set_pixel_map_from_json_file(DEFAULT_PIXEL_MAP)
|
||||
logger.info(f"Generating default pixel map for TimePix detector {self.name}...")
|
||||
self.pixel_map = create_single_energy_per_chip_pixel_map()
|
||||
logger.info(
|
||||
f"Default pixel map for TimePix detector {self.name} loaded after {time.time() - start_time:.3f} seconds."
|
||||
f"Default pixel map for TimePix detector {self.name} generated after {time.time() - start_time:.3f} seconds."
|
||||
)
|
||||
|
||||
def _enable_xes_settings(self, enabled: bool) -> None:
|
||||
|
||||
@@ -1,18 +0,0 @@
|
||||
#!/usr/bin/env python
|
||||
|
||||
"""Client using the asyncio API."""
|
||||
|
||||
import asyncio
|
||||
from websockets.asyncio.client import connect
|
||||
|
||||
|
||||
async def hello():
|
||||
async with connect("ws://localhost:8452/ws") as websocket:
|
||||
await websocket.send("Hello world!")
|
||||
while True:
|
||||
message = await websocket.recv()
|
||||
print(message)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(hello())
|
||||
Reference in New Issue
Block a user