"""Manage group of cameras that are set up as a master and slaves. Original author: Mark Zvilius, 20-July-2019 For: The Sexton Corp """ import E2Protocol as E2 import E2Configs # asssumes E2Configs.load() has already been called from E2Camera import E2Camera from typing import Dict from typing import Any IpType = str class CameraGroup: """A group of cameras set up as a master and slaves.""" def __init__(self, ips, no_lens=False): """Constructor.""" # # ips -- (list) camera ip addresses, [0] is master # no_lens -- if camera has no lens (testing only) # Exceptions: none # self.ips = ips self.cameras = [] # list of camera objects self.recording = False # this is forced in start_cameras() self.no_lens = no_lens def start_cameras(self): """Create camera objects, connect and initialize them.""" # # Exceptions: # ConfigError -- ips[] list is empty # ConnectionError -- if camera connection falied # if len(self.ips) == 0: raise CameraGroup.ConfigError('no cameras are configured') id = 1 master = True # first camera is master for ip in self.ips: cam_str = f'camera {id} ({ip})' print('connect', cam_str) cam = E2Camera(ip, str(id), master=master, no_lens=self.no_lens) settings = self.build_camera_settings(ip) (status,errmsg) = cam.start_camera(settings) if status != 0: raise CameraGroup.ConnectionError(f'{cam_str}: {errmsg}') self.cameras.append(cam) id = id+1 master = False # all other cameras are slaves print() ## # Build camera settings. # # Start with the settings persisted to the camera_settings file. # If there is a focus position config, append that to the settings. # # @param self Object pointer. # @param ip IP address of camera. # @return Dictionary of camera settings. # def build_camera_settings(self, ip: IpType) -> Dict[str,Any]: settings = E2Configs.camera_settings if ip in E2Configs.camera_focus: pos = E2Configs.camera_focus[ip] settings[E2.KEY_LENS_FOCUS_POS] = pos print(f'{ip} focus {pos}') return settings def validate_index(self, idx): """Verify that idx is a valid index into the list of cameras.""" if (idx >= len(self.cameras)) or (idx < 0): raise Exception(f'camera index {idx} invalid (# cams={len(self.cameras)})') def wrap_index_decrement(self, idx): """Decrement a camera index, with wrapping.""" self.validate_index(idx) next_idx = idx+1 if next_idx == len(self.cameras): next_idx = 0 return next_idx def wrap_index_increment(self, idx): """Increment a camera index, with wrapping.""" self.validate_index(idx) next_idx = idx-1 if next_idx < 0: next_idx = len(self.cameras)-1 return next_idx # # TODO: need to write send_to_all_cameras(xxx) .. needs a mapping to # E2Camera.command_xxx() # TODO: need a better way of handling errors .. should try setting in # all cameras even if err # def change_wb(self, new_value): """Change white-balance setting.""" status = 0 errmsg = '' for c in self.cameras: (status,errmsg) = c.command_set_wb(new_value) if status != 0: break return (status,errmsg) def change_iso(self, new_value): """Change iso setting.""" status = 0 errmsg = '' for c in self.cameras: (status,errmsg) = c.command_set_iso(new_value) if status != 0: break return (status,errmsg) def change_shutter(self, new_value): """Change shutter setting.""" status = 0 errmsg = '' for c in self.cameras: (status,errmsg) = c.command_set_shutter(new_value) if status != 0: break return (status,errmsg) def change_encoder(self, new_value): """Change video-encoder setting.""" status = 0 errmsg = '' for c in self.cameras: (status,errmsg) = c.command_set_encoder(new_value) if status != 0: break return (status,errmsg) def change_fstop(self, new_value): """Change f-stop (iris) setting.""" status = 0 errmsg = '' for c in self.cameras: (status,errmsg) = c.command_set_fstop(new_value) if status != 0: break return (status,errmsg) def command_record_toggle(self): """Recording start/stop.""" if self.recording: return self.stop_recording() else: return self.start_recording() def start_recording(self): """Start cameras recording.""" # If no cameras, simulate start recording. if len(self.cameras) == 0: self.recording = True print('start recording') return (0,'') self.recording = False # # TODO: do a "remain" on all cameras to find minimum recording time, # and report. # # Send start to the master. (status,errmsg) = self.cameras[0].command_record('start') if status != 0: return (status,errmsg) # Poll status on all cameras. (status,errmsg) = self.poll_record_status('rec_ing') if status != 0: return (status,errmsg) self.recording = True print('start recording') return (status,errmsg) def stop_recording(self): """Stop cameras recording.""" # State will be not-recording no matter what. self.recording = False print('stop recording') if len(self.cameras) == 0: return (0,'') # Send stop to the master. (status,errmsg) = self.cameras[0].command_record('stop') if status != 0: return (status,errmsg) # Poll status on all cameras. return self.poll_record_status('rec') def poll_record_status(self, expected_status): """Poll all cameras for expected recording status.""" status = 0 errmsg = '' for c in self.cameras: (status,errmsg) = c.check_record_mode(expected_status) if status != 0: break return (status,errmsg) def get_temperature(self): """Get camera temperature. returns -- (status,result) success: status=0 and result=temperature error: status!=0 and result=errmsg """ # TODO: currently only gets master camera temperature. # TODO: probably need to get maximum of all cameras. if len(self.cameras) == 0: return (0,99) else: return self.cameras[0].get_temperature() class Error(Exception): """Base class for exceptions in CameraGroup.""" pass class ConfigError(Error): """Exception for configuration error.""" def __init__(self, message): """Constructor.""" self.message = message class ConnectionError(Error): """Exception for connection error.""" def __init__(self, message): """Constructor.""" self.message = message