import sys,os,socket import argparse sys.path.insert(0, os.path.expanduser('/sf/cristallina/applications/slic/slic-package')) from slic.devices.timing.events import CTASequencer class IntRange: def __init__(self, imin=None, imax=None): self.imin = imin self.imax = imax def __call__(self, arg): try: value = int(arg) except ValueError: raise self.exception() if (self.imin is not None and value < self.imin) or (self.imax is not None and value > self.imax): raise self.exception() return value def exception(self): if self.imin is not None and self.imax is not None: return argparse.ArgumentTypeError(f"ON OFF ratio must be an integer in the range [{self.imin}, {self.imax}]") elif self.imin is not None: return argparse.ArgumentTypeError(f"ON OFF ratio must be an integer >= {self.imin}") elif self.imax is not None: return argparse.ArgumentTypeError(f"ON OFF ratio must be an integer <= {self.imax}") else: return argparse.ArgumentTypeError("ON OFF ratio must be an integer") class tuple_splitter: def __init__(self, x=None, y=None, ssz=False): self.x=x self.y=y self.ssz=ssz def __call__(self,arg): try: self.x = int(arg.split(',')[0]) self.y = int(arg.split(',')[1]) except ValueError: raise self.exception() if self.ssz and self.y != self.x-1: raise self.exception() return (self.x,self.y) def exception(self): if self.ssz and self.y != self.x-1: return argparse.ArgumentTypeError(f"for current motion scheme x,y should equal x, x-1 not {x}, {y}") else: if self.ssz: return argparse.ArgumentTypeError("ssz x and y must be an integer") else: return argparse.ArgumentTypeError("x and y must be an integer") class MotionSim: def __init__(self, args=None): try: self.cta=CTASequencer(args.cta_sequencer) self.clear_flag=args.clear self.check_sequence=args.check_sequence self.motion_mode=args.motion_mode self.repetitions=args.grid_size[0] self.cta_multiplier=args.grid_size[1] self.number_of_appertures=self.repetitions*self.cta_multiplier self.on_off_ratio=args.on_off_ratio+1 self.triggered=args.triggered self.ssz=args.ssz self.smv=args.smv self.twait=args.twait self.tmove=args.tmove except: self.clear=False self.check_sequence=False self.cta=CTASequencer("SAR-CCTA-ESC") self.motion_mode=4 self.repetitions=1 self.cta_multiplier=1 self.number_of_appertures=1 self.on_off_ratio=1 self.triggered=False raise self.exception() def exception(self): if args is not None: print(args) return argparse.ArgumentTypeError("problem with arg parse inputs") else: return argparse.ArgumentTypeError("no args given, using default values") def standard_motion(self): #need to implement on_off_ratio self.cta.seq[214]=[1,]+[0,]*(self.cta_multiplier-1) #start motion self.cta.seq[200]=[1,]*self.cta_multiplier #uncomment me for normal operation self.cta.seq[219]=[1,]*self.cta_multiplier #trigger detector if self.triggered: self.cta.seq[215]=[1,0]*(self.cta_multiplier//2) # shutter always open self.cta.seq[216]=[1,0,]*(self.cta_multiplier//2) # Label image light dark 1:1 #change back to 1,0 for normal on off measurements self.n_pulses_run = len(self.cta.seq[200])*self.repetitions self.cta.cfg.repetitions = self.repetitions self.upload() return def hit_and_return(self): if self.smv: transition_pulses=self.smv else: transition_pulses=self.ssz[0]-1 apperture_group_size = self.ssz[0] * self.ssz[1] number_of_apperture_groups = self.number_of_appertures // apperture_group_size self.repetitions = number_of_apperture_groups # number of repeitions is now number of times we repeat the loop xray_sequence = [0,] * apperture_group_size + [1,] * apperture_group_size + [0,] * (transition_pulses - 1) self.cta.seq[200] = xray_sequence self.cta.seq[214] = [1,] + [0,] * (apperture_group_size-1) + [0,] * apperture_group_size + [0,] * (transition_pulses - 1) #start motion if self.triggered: if self.on_off_ratio > apperture_group_size: self.on_off_ratio = apperture_group_size on_off_sequence = [1,] * (self.on_off_ratio - 1) + [0,] self.cta.seq[215] = on_off_sequence * (apperture_group_size//self.on_off_ratio) + [0,] * apperture_group_size + [0,] * (transition_pulses -1) # Trigger 1:1 repeat_sequence_laser_closed = [1, ] + [0,] * (apperture_group_size - 1) repeat_sequence_laser_open = [1,] + [0,] * (apperture_group_size - 1) tranisition_sequence_laser_shut = [0,] * (transition_pulses - 1) #close shutter after final point before starting move...? self.cta.seq[213] = repeat_sequence_laser_open + repeat_sequence_laser_closed + tranisition_sequence_laser_shut # Trigger 1:1 self.cta.seq[216] = [0,] * apperture_group_size + on_off_sequence * (apperture_group_size//self.on_off_ratio) + [0,] * (transition_pulses - 1) else: self.cta.seq[215] = [0] * apperture_group_size * 2 + [0,] * (transition_pulses - 1) # trigger (laser_shutter or droplet ejector) self.cta.seq[216] = [0] * apperture_group_size * 2 + [0,] * (transition_pulses -1) # image label (on or off)1 self.cta.seq[219] = xray_sequence # detector trigger self.cta.cfg.repetitions = self.repetitions self.upload() return def stop_and_go(self): if self.tmove==0: print("did you mean tmove=0 and not 10? setting wait_pulses to 0!") wait_pulses=0 else: wait_pulses=self.twait//self.tmove xray_sequence = [0,] * wait_pulses + [1,] self.cta.seq[200] = xray_sequence * self.cta_multiplier # x-ray_shutter self.cta.seq[214] = [1,] + [0,] * (len(xray_sequence * self.cta_multiplier) -1) #start motion if self.triggered: trigger_on = [1,] + [0,] * wait_pulses trigger_off = [0,] + [0,] * wait_pulses trigger_sequence = trigger_off + trigger_on * (self.on_off_ratio-1) image_on = [0,] * wait_pulses + [1,] image_label_sequence = [0,] * wait_pulses + [0,] + image_on * (self.on_off_ratio-1) self.cta.seq[215] = trigger_sequence * (self.cta_multiplier//self.on_off_ratio) # trigger (laser_shutter or droplet ejector) self.cta.seq[216] = image_label_sequence * (self.cta_multiplier//self.on_off_ratio) # image label (on or off) else: no_trigger_sequence = [0,] + [0,] * wait_pulses self.cta.seq[215] = no_trigger_sequence * self.cta_multiplier # x-ray_shutter # trigger (laser_shutter or droplet ejector) self.cta.seq[216] = no_trigger_sequence * self.cta_multiplier # x-ray_shutter # image label (on or off)1 self.cta.seq[219] = xray_sequence * self.cta_multiplier # detector trigger self.cta.cfg.repetitions = self.repetitions self.upload() return def upload(self): self.cta.seq.upload() return def clear(self): self.cta.seq[200]=[1,] self.upload() return def single_event(self, event:int): #self.cta.seq[200]=[0,]*99+[1,] self.cta.seq[event]=[0,]*99+[1,] self.upload() return def check(self): self.cta.seq.download() print(self.cta.seq) try: print(f"length of sequence: {len(self.cta.seq[200])}") except Exception as e: print(f"Exception {e}") return def main(args): ctaSeq = MotionSim(args) if args.clear: print("Clearing the CTA (has to have an event so setss [200] to [1]") ctaSeq.clear() return elif args.check_sequence: print("Checking the current CTA sequence") ctaSeq.check() return elif args.droplet: print("Clearing the CTA and priming a single event of [200] and [215] for aligning the droplet ejector") event=215 ctaSeq.single_event(event) elif args.shutter: print("Clearing the CTA and priming a single event of [200] and [213] for aligning the droplet ejector") event=213 ctaSeq.single_event(event) elif args.motion_mode == 1 or args.motion_mode == 2 or args.motion_mode == 3: print("probably doesnt use or doesn't need the cta. But would use a similar scheme to motion_mode 4") ctaSeq.standard_motion() event=200 elif args.motion_mode == 4: ctaSeq.standard_motion() event=200 elif args.motion_mode == 5: ctaSeq.stop_and_go() event=200 elif args.motion_mode == 6: ctaSeq.hit_and_return() event=200 else: print('nothing happened') return n_pulses_run = len(ctaSeq.cta.seq[event])*ctaSeq.repetitions bsdata_scalar = n_pulses_run/ctaSeq.number_of_appertures print(f"uploading sequence to cta of {n_pulses_run} pulses. Sequence length: {len(ctaSeq.cta.seq[200])} for one event. Repetitions: {ctaSeq.repetitions}.") print(f"BS data would be scaled by a factor of {bsdata_scalar}.") ctaSeq.upload() if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument( "-c", "--clear", help="replace current cta sequence", action='store_true', ) parser.add_argument( "-m", "--motion_mode", help="Mimic cta code generation for specific motion code, limit:1-6. Mode:1 point list, mode:2 unused, mode 3: inverse list, mode 4: DEFAULT (short code), motion 5: stop and go, motion 6: hit and return", type=IntRange(1,6), default=4 ) parser.add_argument( "-g", "--grid_size", help="size of grid, default 162,162", type=tuple_splitter(ssz=False), default='162,162' ) parser.add_argument( "-o", "--on_off_ratio", help="the number of on shots (i.e. laser triggers, drolet ejector) per single dark shot. i.e. 1 would give you a 1 on: 1 off ratio. limit:1-100, Note for hit and return the ", type=IntRange(1,100), default=1 ) parser.add_argument( "-w", "--twait", help="input the wait time in multiples of 10 i.e. 10 ms, 20 ms.", type=int, default=10 ) parser.add_argument( "-s", "--ssz", help="needed for motion mode 6, size fo the repeating grid i.e. 4x3, 8x7. Currently limited in real motion to x,x-1 due to smv requirement", type=tuple_splitter(ssz=True), default=None ) parser.add_argument( "--check_sequence", help="print current cta sequence", action='store_true', ) parser.add_argument( "-t", "--triggered", help="print current cta sequence", action='store_true', ) parser.add_argument( "--tmove", help="time between moves for motion mode 5 = stop and go, should be 10", type=int, default=10 ) parser.add_argument( "--smv", help="# of pulses to move between repeats in x and y i.e 3,3 would be 3 pulses to change in x and 3 pulses to change in y. Default value x-1. Currently must be same in x and y due to cta limitations.", type=int, default=None ) parser.add_argument( "--cta_sequencer", help="name of sequencer PV, default = SAR-CCTA-ESC", type=str, default="SAR-CCTA-ESC" ) parser.add_argument( "--droplet", help="for aligning droplet ejector, event 215 at 100 hz", action='store_true', ) parser.add_argument( "--shutter", help="for aligning laser during hit and return, event 213 at 100 hz", action='store_true', ) args = parser.parse_args() if args.motion_mode == 5 and args.twait is None: parser.error("--motion_mode = 5 requires twait to define wait time at each point.") if args.motion_mode == 6 and args.ssz is None: parser.error("--motion_mode = 6 requires ssz to define repeating grid.") main(args)