"""A ZCAM E2 camera. Communicates with a TCP/IP-connected ZCAM E2 camera. Original author: Mark Zvilius, 20-July-2019 For: The Sexton Corp """ import sys import requests import time import E2Protocol as E2 ## Type alias for URL. Url = str class E2Camera: # Stream1 settings (for the video preview). # TODO: Figure out what 'bitrate' does. STREAM1_SETTINGS = {'width':'960', 'height':'540', 'bitrate':'400000', 'fps':'30'} ## # Arguments for OMXPlayer. # # TCP transport seems to be what works for the ZCam E2 cameras. This should # probably be investigated more carefully. UDP would be much more efficient. # # Presumably --live means that it displays each frame as it comes in, however # that should be checked. # # "--aidx -1" effectively turns off audio processing. # OMXPLAYER_ARGS = [ '--avdict', 'rts_transport:tcp', '--live', '--aidx', '-1' ] # Default timeout for camera communication (seconds). # Longest seems to be setting project_fps which can take about 0.7 secs. DEFAULT_TIMEOUT = 1.0 RECORD_TIMEOUT = 2 # record seems to take longer perhaps due to slaves # Status values. ERROR_CONNECTION = 41 # could not connect to camera ERROR_TIMEOUT = 42 # timeout waiting for reply ERROR_SESSION = 43 # failed to get camera session ERROR_CAMERA = 44 # camera in unexpected state def __init__(self, ip, id, master, no_lens=False): """Constructor. ip -- camera ip address id -- (string) unique camera id, e.g. '1', '2', '3', etc. master -- (bool) True if master camera, False if slave no_lens -- if camera has no lens (testing only) """ self.ip = ip self.id = id self.master = master self.no_lens = no_lens def start_camera(self, settings): """Initialize camera. Start with a ping to verify connectivity. Establish a session. Set various parameters appropriate for this application. Initialize dynamic settings (e.g. white balance). settings -- (dictionary) dynamic settings saved from last run returns -- (status,errmsg) status 0 on success errmsg on error """ # Ping to verify connectivity. (status,errmsg) = self.command_info() if status != 0: return (status,errmsg) # Get the session. (status,errmsg) = self.command_get_session() if status != 0: return (status,errmsg) # Initializations for the master camera. if self.master: (status,errmsg) = self.master_inits() if status != 0: return (status,errmsg) # Other camera initializations. (status,errmsg) = self.camera_inits() if status != 0: return (status,errmsg) # Restore the last-used settings. (status,errmsg) = self.camera_settings(settings) if status != 0: return (status,errmsg) # Setting up 'stream1' needs to happen after 'project_fps'. (status,errmsg) = self.command_stream_settings('stream1', E2Camera.STREAM1_SETTINGS) if status != 0: return (status,errmsg) return (0,'') def master_inits(self): """Intializations for the master camera. Disable union exposure (iso and shutter) and white balance. """ (status,errmsg) = self.command_set_union_ae('Disable') if status != 0: return (status,errmsg) return self.command_set_union_awb('Disable') def camera_inits(self): """Various camera initializations. Set camera to record mode. Turn off recording if it is on for some reason. WB set to Manual to allow setting directly in Kelvin. Configure autofocus. returns -- (status,errmsg) from the last command """ # Set camera to record mode. (status,errmsg) = self.command_record_mode('to_rec') if status != 0: return (status,errmsg) (status,errmsg) = self.command_record('stop') # Ignore error: camera will return -1 unless it was recording. # (Probably also returns an error for a slave camera.) # Check that camera is in record mode, not recording. (status,reply) = self.check_record_mode('rec') if status != 0: return (status,reply) # Check that union_ae and union_awb are disabled. (status,reply) = self.verify_unions_disabled() if (status != 0): return (status,reply) # Set white-balance mode to manual so it can be controlled. (status,errmsg) = self.command_set_wb_mode('Manual') if status != 0: return (status,errmsg) # Configure autofocus: no continuous, restore lens position. # (status,errmsg) = self.command_set_focus_mode('AF') # if status != 0: # return (status,errmsg) # (status,errmsg) = self.command_set_continuous_autofocus('Off') # if status != 0: # return (status,errmsg) # (status,errmsg) = self.command_set_restore_lens_pos('Enable') # if status != 0: # return (status,errmsg) return (0,'') def camera_settings(self, settings): """Dynamic camera settings. These are parameters that are changed by the user on the fly, and are persisted to a file and restored at initialization time. The keys and values are persisted using the camera command format, so they can be used directly by build_ctrl_command(). Examples: 'mwb' -- white balance 'iso' -- iso 'shutter_time' -- shutter speed 'video_encoder' -- video encoder 'iris' -- f-stop settings -- (dictionary) returns -- (status,reply) from the last command """ for (k,v) in settings.items(): url = self.build_ctrl_command('set', {k:v}) (status,reply) = E2Camera.http_request(url) if status != 0: # If camera has no lens (testing only), iris command will fail. if (k == 'iris') and self.no_lens: status = 0 continue break return self.make_errmsg((status,reply), '{0}={1} {2}'.format(k,v,reply)) def check_record_mode(self, expected_mode): """Check that the camera record mode is as expected. expected_mode -- 'rec', 'rec_ing', etc. returns -- (status,reply) or (status,errmsg) """ (status,reply) = self.command_record_mode('query') record_mode = reply['msg'] if (status != 0) or (record_mode == expected_mode): return (status,reply) else: # not the expected mode msg = 'record mode: {0} expected {1}'.format(record_mode, expected_mode) return self.make_errmsg((E2Camera.ERROR_CAMERA,reply), msg) def verify_unions_disabled(self): """Check that union_ae and union_awb are disabled. These keys are set on the master. It can take some time for them to propagate to the slave cameras: in testing I have seen as much as 7 seconds. So this method polls, with a 10 second timeout, for the values to read as 'Disable'. returns -- (status,errmsg) """ (status,errmsg) = self.wait_for_setting(E2.KEY_UNION_AE, E2.VAL_DISABLE) if status != 0: return (status,errmsg) return self.wait_for_setting(E2.KEY_UNION_AWB, E2.VAL_DISABLE) ## # Wait for stream ('stream1') to be idle. # # @param self Object pointer. # def wait_for_stream_idle(self): url = self.build_stream_settings_command('stream1', {'action': 'query'}) return self._wait_for_value(url, key='status', value='idle', timeout=20.0) ## # Wait for camera key to have the specified value. # # @param self Object pointer. # @param key Key to test. # @param val Value to test for. # @param timeout Timeout (seconds). # @return (status,errmsg) status=0 if key has value, status=ERROR_CAMERA if timeout # def wait_for_setting(self, key, val, timeout=10.0): url = self.build_ctrl_command('get', {'k': key}) return self._wait_for_value(url, key='value', value=val, timeout=timeout) ## # Wait for a camera setting to have a certain value. # # @param self Object pointer. # @param url HTTP query. # @param key Key to look for in response. # @param value Value that the key should have. # @param timeout Timeout (seconds). # @return (status,errmsg) status=0 if key has value, status=ERROR_CAMERA if timeout # def _wait_for_value(self, url, key, value, timeout): TICK = 0.1 # time between polls elapsed = 0.0 # elapsed time got_val = '' reply = {} for i in range(round(timeout/TICK)): (status,reply) = E2Camera.http_request(url) if status != 0: return (status,reply) got_val = reply[key] if got_val == value: if i != 0: print(f'{self.ip}: got {value} after {elapsed:.1f}') return (0,'') if i==0: print(f'{self.ip}: wait {value} got {got_val}') time.sleep(TICK) elapsed += TICK msg = f'{got_val} after {timeout} secs waiting for {value}' return self.make_errmsg((E2Camera.ERROR_CAMERA,reply), msg) def command_info(self): """Send camera-info command, receive response. Useful as a ping. returns -- (status,reply) if success (status,errmsg) if error """ url = self.build_info_command() r = E2Camera.http_request(url) return self.make_errmsg(r, 'info {0}'.format( r[1] )) def command_get_session(self): """Get camera session. returns -- (status,errmsg) status 409 means could not get session """ url = self.build_session_command() (status,reply) = E2Camera.http_request(url) if status == 409: return self.make_errmsg((E2Camera.ERROR_SESSION,reply), 'could not get session: {0}'.format(reply)) else: return self.make_errmsg((status,reply), 'get_sesssion %s' % (reply)) def command_quit_session(self): """Quit camera session. returns -- (status,errmsg) """ url = self.build_session_command('quit') r = E2Camera.http_request(url) return self.make_errmsg(r, 'quit_session {0}'.format( r[1] )) def command_set_iso(self, val): """Set ISO value. val -- (string) iso setting returns -- (status,reply) from http_request() """ url = self.build_iso_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, 'iso {0} {1}'.format(val, r[1])) def command_set_wb_mode(self, mode): """Set white balance mode: auto/manual. mode -- (string) wb mode ('Auto' or 'Manual') returns -- (status,errmsg) """ url = self.build_wb_mode_command(mode) r = E2Camera.http_request(url) return self.make_errmsg(r, 'wb_mode {0} {1}'.format(mode, r[1])) def command_set_wb(self, val): """Set white balance. val -- (string) wb value as degrees kelvin returns -- (status,errmsg) """ url = self.build_wb_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, 'wb {0} {1}'.format(val, r[1])) def command_set_shutter(self, val): """Set shutter speed. val -- (string) shutter speed returns -- (status,errmsg) """ url = self.build_shutter_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, 'shutter {0} {1}'.format(val, r[1])) def command_set_encoder(self, val): """Set video encoder. val -- (string) video encoder name returns -- (status,errmsg) """ url = self.build_encoder_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, 'encoder {0} {1}'.format(val, r[1])) def command_set_fstop(self, val): """Set f-stop (iris). val -- (string) f-stop value returns -- (status,errmsg) """ url = self.build_iris_command(val) # If camera has no lens (testing only), iris command will fail. r = E2Camera.http_request(url) if self.no_lens: return (0,'') return self.make_errmsg(r, 'fstop {0} {1}'.format(val, r[1])) def command_set_union_ae(self, val): """Set union exposure: when enabled, shutter and iso are controlled by master. Should be set only on the master camera; slaves will return error. val -- (string) Enable/Disable returns -- (status,errmsg) """ url = self.build_union_ae_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, f'union_ae {val} {r[1]}') def command_set_union_awb(self, val): """Set union white balance: when enabled, white balance is controlled by master. Should be set only on the master camera; slaves will return error. val -- (string) Enable/Disable returns -- (status,errmsg) """ url = self.build_union_awb_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, f'union_awb {val} {r[1]}') def command_stream_settings(self, index, settings={}): """Set stream settings. index -- 'stream0' or 'stream1' settings -- e.g. width, height, bitrate, fps returns -- (status,reply) from http_request() """ url = self.build_stream_settings_command(index, settings) r = E2Camera.http_request(url) return self.make_errmsg(r, 'stream_settings {0} {1}'.format(index, r[1])) def command_record_mode(self, action): """Control camera recording mode. action -- 'to_rec', 'to_pb' returns -- (status,reply) or (status,errmsg) """ url = self.build_record_mode_command(action) r = E2Camera.http_request(url) return self.make_errmsg(r, 'record_mode {0} {1}'.format(action, r[1])) def command_record(self, action): """Control recording. action -- 'start', 'stop', or 'remain' returns -- (status,reply) from http_request() (status,errmsg) on error """ url = self.build_record_command(action) r = E2Camera.http_request(url, timeout=E2Camera.RECORD_TIMEOUT) return self.make_errmsg(r, 'record {0} {1}'.format(action, r[1])) ## # Set focus mode. # # @param self Object pointer. # @param mode Manual/autofocus. # @return (status,reply) from http_request() or (status,errmsg) on error # def command_set_focus_mode(self, mode: str) -> (int,str): url = self.build_set_focus_command(mode) r = E2Camera.http_request(url) return self.make_errmsg(r, f'focus {mode} {r[1]}') ## # Set continuous-autofocus on/off. # # @param self Object pointer. # @param val On/off. # @return (status,reply) from http_request() or (status,errmsg) on error # def command_set_continuous_autofocus(self, val: str) -> (int,str): url = self.build_set_caf_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, f'caf {val} {r[1]}') ## # Set restore-lens-position enable/disable. # # @param self Object pointer. # @param val Enable/disable. # @return (status,reply) from http_request() or (status,errmsg) on error # def command_set_restore_lens_pos(self, val: str) -> (int,str): url = self.build_set_restore_lens_pos_command(val) r = E2Camera.http_request(url) return self.make_errmsg(r, f'restore_lens_pos {val} {r[1]}') ## # Set lens-focus-position. # # @param self Object pointer. # @param val Lens focus position (values are lens dependent). # @return (status,reply) from http_request() or (status,errmsg) on error # def command_set_lens_focus_pos(self, pos: int) -> (int,str): url = self.build_set_lens_focus_pos_command(pos) r = E2Camera.http_request(url) return self.make_errmsg(r, f'restore_lens_pos {pos} {r[1]}') def get_temperature(self): """Get camera temperature in degC. returns -- (status,result) success: status=0 and result=temperature error: status!=0 and result=errmsg """ url = self.build_get_temperature() r = E2Camera.http_request(url) # TODO: Finish here... def build_info_command(self): """Build camera-info command. returns -- url string """ return self.url_prefix() + 'info' def build_session_command(self, action=None): """Build camera session command. action -- None to get sesssion, 'quit' to quit session. """ params = {} if action: params['action'] = action return self.build_ctrl_command('session', params) def build_iso_command(self, val): """Build set-iso command. val -- ISO setting. """ return self.build_ctrl_command('set', {E2.KEY_ISO: val}) def build_wb_mode_command(self, mode): """Build set-white-balance-mode command. mode -- white-balance mode ('Auto' or 'Manual') """ return self.build_ctrl_command('set', {E2.KEY_WBMODE: mode}) def build_wb_command(self, val): """Build set-white-balance command. val -- white-balance value """ return self.build_ctrl_command('set', {E2.KEY_WB: val}) def build_shutter_command(self, val): """Build set-shutter-speed command. val -- shutter speed value """ return self.build_ctrl_command('set', {E2.KEY_SHUTTER: val}) def build_encoder_command(self, val): """Build set-encoder command. val -- video encoder name """ return self.build_ctrl_command('set', {E2.KEY_ENCODER: val}) def build_iris_command(self, val): """Build iris command. val -- iris value """ return self.build_ctrl_command('set', {E2.KEY_IRIS: val}) def build_union_ae_command(self, val): """Build union_ae command. val -- Enable/Disable """ return self.build_ctrl_command('set', {E2.KEY_UNION_AE: val}) def build_union_awb_command(self, val): """Build union_awb command. val -- Enable/Disable """ return self.build_ctrl_command('set', {E2.KEY_UNION_AWB: val}) ## # Build set focus command. # # @param self Object pointer. # @param mode Manual/autofocus. # @return URL # def build_set_focus_command(self, mode: str) -> Url: return self.build_ctrl_command('set', {E2.KEY_FOCUS: mode}) ## # Build set continuous-autofocus command. # # @param self Object pointer. # @param val On/off. # @return URL # def build_set_caf_command(self, val: str) -> Url: return self.build_ctrl_command('set', {E2.KEY_CAF: val}) ## # Build set restore-lens-position command. # # @param self Object pointer. # @param val Enable/disable. # @return URL # def build_set_restore_lens_pos_command(self, val: str) -> Url: return self.build_ctrl_command('set', {E2.KEY_RESTORE_LENS: val}) ## # Build set set-lens-focus-position command. # # @param self Object pointer. # @param pos Position (which is lens dependent). # @return URL # def build_set_lens_focus_pos_command(self, pos: int) -> Url: return self.build_ctrl_command('set', {E2.KEY_LENS_FOCUS_POS: pos}) def build_stream_settings_command(self, index, settings): """Build set-stream-settings command. index -- (string) stream index settings -- (dictionary) stream settings """ params = {'index': index, **settings} return self.build_ctrl_command('stream_setting', params) def build_record_mode_command(self, action): """Build record-mode command. action -- (string) """ return self.build_ctrl_command('mode', {'action': action}) def build_record_command(self, action): """Build record command. action -- (string) 'start', 'stop', etc """ return self.build_ctrl_command('rec', {'action': action}) def build_ctrl_command(self, command, params={}): """Build a "/ctrl" command. Format: /ctrl/[?=[&=]...] command -- (string) params -- (dictionary) option-value pairs """ url = f'{self.url_prefix()}ctrl/{command}' query_char = '?' for (k,v) in params.items(): url = url + f'{query_char}{k}={v}' query_char = '&' return url def url_prefix(self): """Return prefix for URL.""" return f'http://{self.ip}/' @staticmethod def http_request(url, payload=None, timeout=DEFAULT_TIMEOUT): """Issue request and get reply. url -- (string) payload -- (string or None) timeout -- comm timeout in seconds (number) returns -- (status,reply) status: (integer) 0 success, nonzero ERROR_TIMEOUT or status_code reply: (dictionary) json portion of reply, or error message """ try: r = requests.get(url, params=payload, timeout=timeout) except requests.exceptions.ConnectionError: return (E2Camera.ERROR_CONNECTION, 'connection error: {0} ({1} secs)'.format(url, timeout)) except requests.exceptions.Timeout: return (E2Camera.ERROR_TIMEOUT, 'timeout: {0} ({1} secs)'.format(url, timeout)) if r.status_code != 200: # not HTTP STATUS_OK return (r.status_code, 'http status {0}'.format(r.status_code)) status = 0 reply = r.json() if 'code' in reply: status = reply['code'] return (status,reply) def make_errmsg(self, sr, msg): """Make human-readable error message. sr -- (status, reply) tuple msg -- partial error message returns -- (status,reply) if success (status,errmsg) if error """ status = sr[0] if status == 0: return sr else: return (status, 'cam{0}: {1}'.format(self.id, msg)) if __name__ == '__main__': import E2Camera_tests DEFAULT_CAMERA_IP = '10.6.10.162' if (len(sys.argv) > 1): ip = sys.argv[1] else: ip = DEFAULT_CAMERA_IP cam = E2Camera(ip, '1', master=True) E2Camera_tests.run_unit_tests(cam) sys.exit()