#!/usr/bin/env python3 # # Copyright 2024-2025 FieldLine Industries, Inc. # # Proprietary and Confidential. # import argparse import logging import numpy as np import serial import struct import time from queue import Queue, Empty from serial.tools.list_ports import comports import threading import socket import sys import re import signal from time import sleep, perf_counter import csv # To set this do the following: # 1.) Clear the version in APP_VERSION # 2.) run: # ./fli_scalar_api_set_ver.sh APP_VERSION="v0.2.0-gb1c482c" # TODO: put into class object like SM200RegisterMap WAVEGEN_SHAPE_MAX = 7 WAVEGEN_INDEX_MAX = 28 ZFS_WAVEGEN_CLK_HZ = 25000 class BidirectionalStrIntMap(object): ''' Dictionary with bidirectional str<->int lookup. ''' def __init__(self): self._dict = {} def __getitem__(self, key): if isinstance(key, int): if key not in self._dict.values(): raise IndexError(key) return list(self._dict.keys())[list(self._dict.values()).index(key)] elif isinstance(key, str): if key not in self._dict: raise KeyError(key) return self._dict[key] else: raise TypeError(key) def __str__(self): s = f'{type(self).__name__}: {{ \n' for k, v in self._dict.items(): s += f' {k}: {v}\n' s += '}' return s class SM200StreamMap(BidirectionalStrIntMap): ''' SM200 data stream name to id mapper. ''' def __init__(self): self._dict = { "dds_output_freq": 18, "mag_mz": 23, "logic_module_state": 35, "pps_clock_count": 61, "raw_pps_output": 67, } class SM200RegisterMap(BidirectionalStrIntMap): ''' SM200 register name to id mapper. ''' def __init__(self): self._dict = { "sync_config": 0x00, "pcb_id": 0x02, "read_reg": 0x03, "scratch": 0x04, "schedule_frequency": 0x17, # 23 "checksum_and_state_mon": 0x43, # 67 "uart_rate": 0x44, # 68 "logic_module_ctrl": 0x4d, # 77 "led_ctrl": 0x6e, # 110 } LOGIC_MODULE_START_CMD = 0x1F LOGIC_MODULE_STOP_CMD = 0x00 class SM200StateMap(BidirectionalStrIntMap): ''' SM200 state name to id mapper. ''' def __init__(self): self._dict = { "off": 0, "laser_check": 1, "warm_up": 2, "optical_scan": 3, "heat_stabilize": 4, "magnetic_scan": 5, "magnetic_lock": 6, } class SM200StreamDecoder: def __init__(self): self.data = b'' self.esc_char = False self.valid_packet = False self.sensor_name = "" def set_name(self, sensor_name): self.sensor_name = sensor_name def get_name(self): return self.sensor_name def add_char(self, in_char): ret = dict() timestamp = None raw_cmd = [] if not self.esc_char and in_char == 0x1B: self.esc_char = True elif self.esc_char: self.esc_char = False self.data += bytes([in_char]) elif in_char == 0x0A: self.valid_packet = True self.data = b'' elif in_char == 0x0D: if len(self.data) < 7: logging.error(f"Invalid data, not enough bytes len(self.data): {len(self.data)}") else: timestamp = int.from_bytes(self.data[0:2], byteorder='big') ret = dict() self.data = self.data[2:] while len(self.data): datatype = int.from_bytes(self.data[0:1], byteorder='big') try: dataval = struct.unpack(">i", self.data[1:5])[0] except struct.error: logging.error("unpack requires a buffer of 4 bytes, continuing") pass if datatype == 3: raw_reg = (dataval >> 16) & 0xFFFF raw_val = dataval & 0xFFFF raw_cmd.append((raw_reg, raw_val)) else: ret[datatype] = dataval self.data = self.data[5:] self.valid_packet = False elif self.valid_packet: self.data += bytes([in_char]) elif self.data != b'': logging.warning(f"partial data read detected! self.data: {self.data}") return timestamp, ret, raw_cmd class PFN200RegisterMap(BidirectionalStrIntMap): ''' PFN200 register name to id mapper. ''' def __init__(self): self._dict = { "pfn_core": 0x00, "read_reg": 0x02, "timeout": 0x04, "schedule_frequency": 0x06, "version": 0x08, "unused_0": 0x0A, "serial_number_lower": 0x0C, "serial_number_upper": 0x0E, "pfn_coil_zero": 0x10, "pfn_coil_one": 0x12, "pfn_coil_two": 0x14, "unused_1": 0x16, } class PFN200StreamMap(BidirectionalStrIntMap): ''' PFN200 data stream name to id mapper. ''' def __init__(self): self._dict = { "dac_0": 0, "dac_1": 1, "dac_2": 2, "read": 3, "pps_clock_count": 4, } class FliScalarApi(object): '''Python object for interacting w/ the FLI Scalar Magnetometer (SM200)''' _valid_device_list = [(0x403, 0x6010), (0x403, 0x6001), (0x10C4, 0xEA60), (0x1A86, 0x7523), (0x4b4, 0x5)] def __init__(self, callback=None, type="scalar"): super().__init__() self._connected_devices = {} self._id_counter = 0 self._logger = logging.getLogger('fli-scalar-api') self._serial_threads = {} self._freq = 1000 self._data_callback = callback self._read_callback = None # needed for FLIRecorder recording callback self._raw_cmd_callback = None self._lock = threading.Lock() self._last_timestamp = None self.type = type # 'scalar' or 'pfn' self.stream_map = SM200StreamMap() self.reg_map = SM200RegisterMap() self.state_map = SM200StateMap() self.pfn_reg_map = PFN200RegisterMap() self.pfn_stream_map = PFN200StreamMap() # Determines if we allow synchronus reads in _serial_loop and _net_recv_loop # Needed to support calls to connect_to_device() when a _raw_cmd_callback is registered self.sync_read = True def _gen_device_name(self, device): device_name = device['name'] name = device_name device_name_parts = device_name.split("/") if "/dev/" in name: # *nix but only tested on linux # self._logger.info(f"device_name_parts {device_name_parts}") if len(device_name_parts) > 1: if ":" in name: name = device_name_parts[2].split(":")[0] else: name = device_name_parts[2] elif "COM" in name: # Windows device_name_parts = device_name.split(":") # logging.info(f"device_name_parts {device_name_parts}") if len(device_name_parts) > 0: if ":" in name: name = device_name_parts[0] logging.info(f"_gen_device_name({device_name}) = {name}") return name def _gen_name(self, port): return port.device + ":" + port.description def _get_vid_pid_interface(self, port): '''Takes comport, returns (vid, pid, interface) from comport''' vid = 0 pid = 0 interface = '' if "/dev/" in port.device: # *nix but only tested on linux # logging.info("found linux platform") vid = port.vid pid = port.pid if port.location is not None: interface = port.location else: interface = '' return (vid, pid, interface) elif "COM" in port.device: # Windows` # logging.info("found Win platform") hwid = port.hwid # logging.info(f"hwid: {hwid}") hwid_split = hwid.split("\\") # logging.info(f"hwid_split: {hwid_split}") vid_pid_str = '' if len(hwid_split) > 1: vid_pid_str = hwid_split[1] # logging.info(f"vid_pid_str: {vid_pid_str}") vid_pid_interface_split = vid_pid_str.split("&") # logging.info(f"vid_pid_interface_split: {vid_pid_interface_split}") if len(vid_pid_interface_split) > 2: vid = vid_pid_interface_split[0][4:] pid = vid_pid_interface_split[1][4:] interface = vid_pid_interface_split[2][3:] else: vid = "0" pid = "0" # logging.info(f"vid: {vid} pid: {pid} interface: {interface}") return (int(vid, 16), int(pid, 16), interface) else: logging.error(f"failed to match {port.device} w/ 'COM' or '/dev/' unknown platform") return (vid, pid, interface) def _is_usable_device(self, port): SM300_VID = 0x4b4 SM300_PID = 0x0005 check_for_stm_uart = False port_is_stm_sm300 = False (vid, pid, location) = self._get_vid_pid_interface(port) try: if port.name.startswith('scalar'): return True # logging.info(f"vid: {vid} pid: {pid} location: {location}") if vid == SM300_VID and pid == SM300_PID: check_for_stm_uart = True if "/dev" in port.device: # *nix but only tested on linux if check_for_stm_uart: port_is_stm_sm300 = re.search(":1.2", location) if not port_is_stm_sm300: try: # logging.info(f"vid {hex(vid)} pid {hex(pid)}") return (vid, pid) in self._valid_device_list except Exception as e: # logging.info(f"Failed to check port: {port} with exception {e}") return False elif "COM" in port.device: # Windows if check_for_stm_uart: port_is_stm_sm300 = re.search("02", location) if not port_is_stm_sm300: try: # logging.info(f"vid {hex(vid)} pid {hex(pid)}") return (vid, pid) in self._valid_device_list except Exception as e: # logging.info(f"Failed to check port: {port} with exception {e}") return False else: self._logger.info(f"Failed to check port: {port}") return False except Exception as e: self._logger.info(f"Failed to check port: {port} with exception {e}") return False def _int_to_hex_str(self, val, length=2): ret_str = b"" val_str = str.encode(hex(val)[2:]) if len(val_str) > length: raise ValueError(val_str) for c in range(0, length - len(val_str)): ret_str += b"0" ret_str += val_str return ret_str # This sends both commands to set addr of stream to read and sends # onetime read command on stream #3. def _send_one_time_read(self, name, addr, timeout): data_str = b"@" data_str += self._int_to_hex_str(3) data_str += self._int_to_hex_str(addr, length=4) ret = self._add_to_wr_queue(name, data_str) if ret == BrokenPipeError: return ret ret = self._one_time_read(name, addr) if ret == BrokenPipeError: return ret # Used for the PFN (which has converted to proper 32-bit registers) # This sends both commands to set addr of stream to read and sends # onetime read command on stream #3. def _send_one_time_read_32(self, name, addr, timeout): data_str = b"@" data_str += self._int_to_hex_str(3) data_str += self._int_to_hex_str(addr, length=4) ret = self._add_to_wr_queue(name, data_str) if ret == BrokenPipeError: return ret data_str_up = b"@" data_str_up += self._int_to_hex_str(2) data_str_up += self._int_to_hex_str(0, length=4) ret = self._add_to_wr_queue(name, data_str_up) if ret == BrokenPipeError: return ret ret = self._one_time_read(name, addr) if ret == BrokenPipeError: return ret # Since the 'rd_queue' won't get populated due to the raw_callback being registered # avoid waiting for the timeout here and just return None to unblock the call if self._raw_cmd_callback is not None: return None try: data_to_return = self._connected_devices[name]['rd_queue'].get(block=True, timeout=timeout) except Empty: return None return data_to_return[0][1] def _net_send(self, device): name = device['data'].get_name() # self._logger.info(f"Start _net_send for device: {device}") self.reset_schedule(name) try: while self._connected_devices[name]['keep_running']: try: data_str = device['wr_queue'].get(timeout=0.01) # self._logger.info(f"WRITE {data_str}") device['port'].sendall(data_str) except Empty: pass except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _net_send and handled, continuing") try: self._connected_devices[name]["wr_queue"].queue.clear() except Exception as e: self._logger.exception(e) self._logger.error("exception taken doing queue.clear() in _net_send and handled, continuing") self._logger.info(f"_net_send exiting for {name}") try: device['port'].shutdown(socket.SHUT_RDWR) except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _net_send and handled, continuing") try: device['port'].close() except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _net_send and handled, continuing") self._logger.info(f"exiting _net_send for {name}") def _net_recv(self, device): name = device['data'].get_name() # self._logger.info(f"Start _net_recv for device: {name}") self.reset_schedule(name) original_name = name try: while device['keep_running']: try: in_chars = device['port'].recv(1024) except socket.error as msg: self._logger.error(f"ERROR device: {device['name']} err: {msg}") device['keep_running'] = False continue if in_chars: # This logic is needed because we start this thread with a name of host:port, as an example, but we # require the 6-digit sensor SN as the name so we need to update the internal name to match if name == original_name: name = device['data'].get_name() for i in range(0, len(in_chars)): timestamp, new_data, raw_cmd = device['data'].add_char(in_chars[i]) if new_data: data_to_send = {'timestamp': timestamp, 'devices': {}} data_to_send['devices'][device['data'].get_name()] = new_data if device['recording']: self._save_data(name, data_to_send) if self._data_callback is not None: self._data_callback(name, data_to_send) if self._read_callback is not None: self._read_callback(name, data_to_send) if raw_cmd: if self._raw_cmd_callback is not None: self._raw_cmd_callback(name, raw_cmd) if self.sync_read: device['rd_queue'].put(raw_cmd) else: self._logger.error(f"ERROR device: {device['name']} ") device['keep_running'] = False continue except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _net_recv and handled, continuing") self._logger.info(f"_net_recv exiting for name {name}") try: device['port'].shutdown(socket.SHUT_RDWR) except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _net_recv and handled, continuing") try: device['port'].close() except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _net_recv and handled, continuing") self._logger.info(f"exiting _net_recv for {name}") def _serial_loop(self, device): name = device['data'].get_name() # self._logger.info(f"Start serial loop for device: {name}") self.reset_schedule(name) original_name = name try: while device['keep_running']: try: while True: data_str = device['wr_queue'].get(block=False) device['port'].write(data_str) except Empty: pass try: in_chars = device['port'].read(1024) except serial.SerialException: self._logger.error(f"SerialException device: {device['name']}") device['keep_running'] = False continue if in_chars: # This logic is needed because we start this thread with a name of 'ttyUSB0', as an example, but we # require the 6-digit sensor SN as the name so we need to update the internal name to match if name == original_name: name = device['data'].get_name() for i in range(0, len(in_chars)): timestamp, new_data, raw_cmd = device['data'].add_char(in_chars[i]) if new_data: data_to_send = {'timestamp': timestamp, 'devices': {}} data_to_send['devices'][device['data'].get_name()] = new_data # self._logger.info(f"data_to_send: {data_to_send}") # self._logger.debug(f"new_data {new_data}") if device['recording']: self._save_data(name, data_to_send) # else: # self._logger.error(f"device['recording'] is False") if self._data_callback is not None: self._data_callback(name, data_to_send) # self.data_available_sig.emit(data_to_send) # else: # self._logger.error(f"callback is {self._data_callback}... skipping") if self._read_callback is not None: self._read_callback(name, data_to_send) # else: # self._logger.debug("new_data is 'None', skipping new_data") if raw_cmd: if self._raw_cmd_callback is not None: self._raw_cmd_callback(name, raw_cmd) if self.sync_read: device['rd_queue'].put(raw_cmd) # self.raw_cmd_resp_sig.emit(raw_cmd) else: time.sleep(0.001) except Exception as e: self._logger.exception(e) self._logger.error("exception taken in _serial_loop and handled, continuing") # self._logger.info(f"serial port exiting for {name} on device {device}") try: device["wr_queue"].queue.clear() except Exception as e: self._logger.exception(e) self._logger.error("exception taken doing queue.clear() in _serial_loop and handled, continuing") device['port'].close() def _save_data(self, name, data_to_save): '''Takes name & data_to_save, saves the data for later use''' # Get the list of "devices" from data_to_save devices = data_to_save['devices'] timestamp = data_to_save['timestamp'] streams = devices[name] # Get all the channel ID's from the list of tuples for use in the below loop channel_list_ids = [lis[0] for lis in self._connected_devices[name]['channels']] # self._logger.debug("channel_list_ids:", channel_list_ids) with self._lock: for channel_id in channel_list_ids: if channel_id in streams: break else: return self._data['time'].append(time.time()) for channel_id in channel_list_ids: if self.type == "scalar": if self.stream_map[channel_id] in self._data: if channel_id in streams: self._data[self.stream_map[channel_id]].append(streams[channel_id]) # Save sensor_state for use in start_sensor() if channel_id == 35: self._connected_devices[name]['sensor_state'] = streams[channel_id] else: self._data[self.stream_map[channel_id]].append(float("NaN")) # self._logger.error(f"Sample received in _save_data not in channel_list: inserting NaN channel_id: {channel_id}") elif self.type == "pfn": if self.pfn_stream_map[channel_id] in self._data: if channel_id in streams: self._data[self.pfn_stream_map[channel_id]].append(streams[channel_id]) else: self._data[self.pfn_stream_map[channel_id]].append(float("NaN")) # self._logger.error(f"Sample received in _save_data not in channel_list: inserting NaN channel_id: {channel_id}") # Each timestamp tick represents one sample. # We need to populate the data for any samples we missed. # TODO: look into (timestamp != 0) and clean that up... It will prevent the timestamp overflow logic from working if self._last_timestamp is not None and timestamp != self._last_timestamp+1 and timestamp != 0: # Handle if timestamp overflows if self._last_timestamp > timestamp: time_delta = (65535 - self._last_timestamp) - timestamp - 1 else: time_delta = timestamp - self._last_timestamp - 1 if time_delta > 1: self._logger.error(f"timestamp: {timestamp} last_timestamp: {self._last_timestamp} time_delta: {time_delta} range(time_delta): {range(time_delta)}") # For each missed sample we populate a time and NaN sample for each channel for i in range(time_delta): # self._logger.error("sample missed...") self._data['time'].append(time.time()) for channel_id in channel_list_ids: if self.type == "scalar": if self.stream_map[channel_id] in self._data: self._data[self.stream_map[channel_id]].append(float("NaN")) elif self.type == "pfn": if self.pfn_stream_map[channel_id] in self._data: self._data[self.pfn_stream_map[channel_id]].append(float("NaN")) self._last_timestamp = timestamp def _start_data_stream(self, name, stream_channel_list): '''Takes name - which sensor to work on Takes stream_channel_list list of streams to enable for that sensor Returns BrokenPipeError if remote connection is closed''' self.reset_schedule(name) self._data = {'time': []} self._last_timestamp = None cur_chan_list = [] # Ensure we only add new channels to the channel list for chan, freq in self._connected_devices[name]['channels']: cur_chan_list.append(chan) for channel in stream_channel_list: # self._logger.warning(f"channel: {channel} self._connected_devices[name]['channels'] {self._connected_devices[name]['channels']}") if channel in cur_chan_list: # self._logger.warning(f"channel: {channel} already in self._connected_devices[name]['channels'] {self._connected_devices[name]['channels']} streaming skipping this channel") continue data_str = b"#" channel_id = '' # Convert from int to string for the FLIRecorder if isinstance(channel, str): if self.type == "scalar": channel_id = self.stream_map[channel] elif self.type == "pfn": channel_id = self.pfn_stream_map[channel] else: channel_id = channel ch_str = str.encode(hex(channel_id)[2:]) while len(ch_str) < 2: ch_str = b"0" + ch_str data_str += ch_str freq_val = self._calc_freq_val(self._freq) freq_str = str.encode(hex(freq_val)[2:]) while len(freq_str) < 4: freq_str = b"0" + freq_str data_str += freq_str self._connected_devices[name]['channels'].append((channel_id, self._freq)) ret = self._add_to_wr_queue(name, data_str) if ret == BrokenPipeError: return ret self._data[channel] = [] self._send_sync(name) def _reset_data(self, name): self._data = {'time': []} self._last_timestamp = None if self.type == "scalar": for c, channel in enumerate(self._connected_devices[name]['channels']): self._data[self.stream_map[channel[0]]] = [] elif self.type == "pfn": for c, channel in enumerate(self._connected_devices[name]['channels']): self._data[self.pfn_stream_map[channel[0]]] = [] def _convert_to_base36(self, val): ret = "" char_list = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ" while len(ret) < 6: ret = char_list[val % 36] + ret val //= 36 return ret def _calc_freq_val(self, freq): if freq == 0: return 0 else: return int(25000 / freq) def _send_sync(self, name): '''Takes in name, send the sync stream command to the sensor''' if self.type == "scalar": self._connected_devices[name]['wr_queue'].put(b"@000001") elif self.type == "pfn": self._connected_devices[name]['wr_queue'].put(b"@010000") self._connected_devices[name]['wr_queue'].put(b"@000004") def _add_to_wr_queue(self, name, data_str): # Only add to the wr_queue if the remote side has not dropped the connection # otherwise we will get stuck waiting here to drain the wr_queue when there is no-one # on the other side to recieve the data if self._connected_devices[name]['keep_running'] == True: self._connected_devices[name]['wr_queue'].put(data_str) else: return BrokenPipeError def _sync_to_pps(self): '''Loop through the connected_device_list and sync all devices to the same PPS edge''' sync_config_addr = self.reg_map['sync_config'] for device in self.get_connected_device_list(): self.sync_read = True cv_csr = self.read_register(device, sync_config_addr) if cv_csr is None: self._logger.error("failed to send the sync to PPS because read_regsister returned 'None'") self.sync_read = False return # Set the Reset PPS Count bit in the Cyclone CV CSR cv_csr = cv_csr | (1 << 9) # self._logger.info(f"writing {cv_csr} to addr: {sync_config_addr}") self.write_register(device, sync_config_addr, cv_csr) self.sync_read = False def set_baud_rate(self, name, value): '''Takes name & value and sets baud to value Returns BrokenPipeError if remote connection is closed''' return self.send_command(name, 68, value) def is_connected(self): '''Returns True if connected to a device False otherwises''' return len(self._connected_devices) > 0 def register_callback(self, callback, type="data"): '''Function signature for the callback: my_callback(name, data): name - the device name example: "ttyACM0" data - a dictionary: Sample # {'timestamp': 363, 'devices': {'Dev0': [39: 37153257, 40: -32832852, 41: 69986109]}} Deconstructing the 'devices' key: Keys are channel ID's and values are samples device [channel id: sample, channel id: sample, channel id: sample] {'Dev0': [39: 37153257, 40: -32832852, 41: 69986109]} ''' if type == "data": self._data_callback = callback elif type == "read": self._read_callback = callback elif type == "raw": self._raw_cmd_callback = callback else: self._logger.error(f"type error: {type} not recognized") # self._logger.info(f"setting {type} callback") def get_human_readable_device_list(self): '''Returns a list of human readable devices''' ret = [] ports = comports(True) for p in ports: name = self._gen_name(p) ret.append(name) return ret def get_chassis_name_to_id(self): ''' TODO: document this function...''' ret = {} for n, d in self._connected_devices.items(): ret[n] = d['id'] return ret def get_connected_device_list(self): '''Returns a list of connected devices''' return list(self._connected_devices.keys()) def get_connected_device_list_thread_name(self): '''Returns a list of connected devices using the 'thread_name' or original name passed into connect_to_device/connect_to_net_device. Needed when connecting or disconnecting from the FLIRecorder''' serial_list = [] for sensor_sn in self._connected_devices.keys(): serial_list.append(self._connected_devices[sensor_sn]['thread_name']) return serial_list def get_connected_human_readable_device_list(self): '''Return a list of human readable device list''' ret = [] for name, port in self._connected_devices.items(): # logging.info(f"name: {name} port: {port}") if name in self._connected_devices: # logging.info(f"get_connected_human_readable_device_list checking device: {name}") if self._connected_devices[name]['transport'] == "serial": if self._connected_devices[name]['sensor_serial_num'] != '': ret.append(self._connected_devices[name]['sensor_serial_num']) elif self._connected_devices[name]['transport'] == "net": if self._connected_devices[name]['sensor_serial_num'] != '': ret.append(self._connected_devices[name]['sensor_serial_num']) # logging.info(f"get_connected_human_readable_device_list connected dev list: {ret}") return ret def reset_schedule(self, name): sensor_sn = name # self._logger.info(f"reset schedule for {name} serial number: {sensor_sn}") try: '''Takes in name, Disable all streams''' if sensor_sn is not None: del self._connected_devices[sensor_sn]['channels'][:] if self.type == "scalar" and sensor_sn is not None: self._connected_devices[sensor_sn]['wr_queue'].put(b"@000002") elif self.type == "pfn": self._connected_devices[sensor_sn]['wr_queue'].put(b"@010000") self._connected_devices[sensor_sn]['wr_queue'].put(b"@000002") except: self._logger.warning(f"failed to reset schedule for {name} serial number: {sensor_sn}") pass def wait_to_drain_write_queue(self, name): '''Takes in name, blocks until the wr_queue is empty''' self._logger.debug("waiting for wr_queue to be empty") while not self._connected_devices[name]['wr_queue'].empty(): time.sleep(0.01) self._logger.debug("done waiting for wr_queue to be empty") time.sleep(0.01) def connect_to_net_device(self, ip_addr, port): '''Takes in ip_addr and port, returns True if connected, False otherwise''' host_port = ip_addr + ":" + str(port) # self._logger.info(f"Connecting to {host_port} with {ip_addr}:{port}") try: net = socket.socket() net.connect((ip_addr, port)) self._connected_devices[host_port] = { 'thread_name': host_port, 'name': host_port, 'port': net, 'id': self._id_counter, 'data': SM200StreamDecoder(), 'channels': [], 'wr_queue': Queue(), 'rd_queue': Queue(), 'sensor_state': 0, 'keep_running': True, 'recording': False, 'transport': 'net', 'sensor_serial_num': '', 'scalar_serial_num': '', 'cv_fw_version': '0.0.0', } self._connected_devices[host_port]['data'].set_name(self._gen_device_name(self._connected_devices[host_port])) self._id_counter += 1 self._serial_threads[host_port+'_net_recv'] = threading.Thread(target=self._net_recv, args=(self._connected_devices[host_port],)) self._serial_threads[host_port+'_net_recv'].start() self._serial_threads[host_port+'_net_send'] = threading.Thread(target=self._net_send, args=(self._connected_devices[host_port],)) self._serial_threads[host_port+'_net_send'].start() # Get the sensor serial number, scalar electronics serial number and Cyclone V Firmware Version sensor_sn = self.get_sn_sensor_base36(host_port) if sensor_sn is None or (sensor_sn is not None and sensor_sn == '000000'): self._logger.error(f"Failed to read sensor serial number for sensor {host_port}") self._connected_devices[host_port]['thread_name'] = 'none' self._connected_devices[host_port]['sensor_serial_num'] = '' self.disconnect_from_device(host_port) raise socket.timeout self._connected_devices[host_port]['sensor_serial_num'] = sensor_sn self._connected_devices[host_port]['scalar_serial_num'] = self.get_sn_scal_elec_base36(host_port) self._connected_devices[host_port]['cv_fw_version'] = self.get_fw_version_str(host_port) self._connected_devices[host_port]['data'].set_name(host_port) # rename the main key in _connected_devices to sensor serial number from host:port self._connected_devices[sensor_sn] = self._connected_devices.pop(host_port) self._logger.info(f"remapping addr:port {host_port} to serial number {sensor_sn}") except Exception as ex: self._logger.error(f"Failed to connect to {ip_addr}:{port} ex: {ex}") return False return True def connect_to_device(self, name, baudrate): '''Takes in name and baudrate, returns True if connected, False otherwise''' # self._logger.info(f"Connecting to name: '{name}' with baudrate: '{baudrate}'") try: ser = serial.Serial(name, baudrate=baudrate, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, bytesize=serial.EIGHTBITS, timeout=0) # Clear any remaining data from kernel buffers, before we open this serial port ser.reset_input_buffer() self._connected_devices[name] = { 'thread_name': name, 'name': name, 'port': ser, 'id': self._id_counter, 'data': SM200StreamDecoder(), 'channels': [], 'wr_queue': Queue(), 'rd_queue': Queue(), 'sensor_state': 0, 'keep_running': True, 'recording': False, 'transport': 'serial', 'sensor_serial_num': '', 'scalar_serial_num': '', 'cv_fw_version': '0.0.0', } self._connected_devices[name]['data'].set_name(self._gen_device_name(self._connected_devices[name])) self._id_counter += 1 self._serial_threads[name] = threading.Thread( target=self._serial_loop, args=(self._connected_devices[name],)) self._serial_threads[name].start() # Get the sensor serial number, scalar electronics serial number and Cyclone V Firmware Version sensor_sn = self.get_sn_sensor_base36(name) if sensor_sn is None or (sensor_sn is not None and sensor_sn == '000000'): self._logger.error(f"Failed to read sensor serial number for sensor {name}") self._connected_devices[name]['thread_name'] = 'none' self._connected_devices[name]['sensor_serial_num'] = '' self.disconnect_from_device(name) raise serial.SerialTimeoutException self._connected_devices[name]['sensor_serial_num'] = sensor_sn self._connected_devices[name]['scalar_serial_num'] = self.get_sn_scal_elec_base36(name) self._connected_devices[name]['cv_fw_version'] = self.get_fw_version_str(name) self._connected_devices[name]['data'].set_name(sensor_sn) # rename the main key in _connected_devices to sensor serial number from '/dev/ttyUSBX' self._connected_devices[sensor_sn] = self._connected_devices.pop(name) self._logger.info(f"remapping device {name} to serial number {sensor_sn}") except Exception as ex: self._logger.error(f"Failed to connect to {name} ex: {ex}") return False return True def stop(self): '''Disconnect from all connected devices''' for name in list(self._connected_devices): self.disconnect_from_device(name) def disconnect_from_device(self, name): '''Takes in name, disconnect from that device''' if name in self._connected_devices: self.reset_schedule(name) # in this case, if disable_streams fails due to a broken pipe we will # run the below code to cleanup our instance and halt the running threads self.disable_streams(name) self.wait_to_drain_write_queue(name) self._connected_devices[name]['keep_running'] = False self._connected_devices[name]['sensor_state'] = 0 if self._connected_devices[name]['transport'] == 'net': thread_name = self._connected_devices[name]['thread_name'] if thread_name != 'none': self._serial_threads[thread_name+"_net_send"].join() self._serial_threads[thread_name+"_net_recv"].join() del self._serial_threads[thread_name+"_net_send"] del self._serial_threads[thread_name+"_net_recv"] del self._connected_devices[name] elif self._connected_devices[name]['transport'] == 'serial': thread_name = self._connected_devices[name]['thread_name'] if thread_name != 'none': self._serial_threads[thread_name].join() del self._serial_threads[thread_name] del self._connected_devices[name] return True else: self._logger.warning(f"name {name} not found in _connected_devices, skipping") return False def get_sensor_state(self, name): '''Takes in name, returns sensor state''' return self._connected_devices[name]['sensor_state'] def get_freq(self): '''Return the frequency the devices are configured to stream at''' return self._freq def set_freq(self, freq): '''Set the streaming samplerate, in Hz''' self._freq = freq for name in self.get_connected_device_list(): if self.type == "scalar": addr = self.reg_map['schedule_frequency'] ret = self.send_command(name=name, addr=addr, data=self._calc_freq_val(freq)) if ret == BrokenPipeError: return ret elif self.type == "pfn": addr = self.pfn_reg_map['schedule_frequency'] ret = self.write_register_32(name, addr, self._calc_freq_val(freq)) if ret == BrokenPipeError: return ret for c, channel in enumerate(self._connected_devices[name]['channels']): # self._logger.debug("'channels'[0]: ", self._connected_devices[name]['channels'][c]) self._connected_devices[name]['channels'][c] = (channel[0], freq) def get_channels(self, name): '''Takes in name, returns list of enabled channel names''' # self._logger.info(f"get_channels: name {name}") return [("%s-%d" % (self._connected_devices[name]['data'].get_name(), c[0]), c[1]) for c in self._connected_devices[name]['channels']] def get_all_channels(self): '''Returns list of all enabled channel names''' ret = [] for name in self._connected_devices.keys(): if name is not None: ret.extend(self.get_channels(name)) return ret def get_channel_config(self, name): '''Takes in name, returns dict of channel configs''' return self._connected_devices[name]['channels'] def _one_time_read(self, name, addr): '''Takes name and addr, returns value from register''' data_str = b"#" data_str += self._int_to_hex_str(3) data_str += b"FFFF" return self._add_to_wr_queue(name, data_str) # Send a command starting with @ to indicate write to a user register def send_command(self, name, addr, data): '''Takes name addr & data, sends data to addr''' data_str = b"@" data_str += self._int_to_hex_str(addr) data_str += self._int_to_hex_str(data, length=4) return self._add_to_wr_queue(name, data_str) def write_register(self, name, addr, value): '''Takes name, addr, value, writes value to addr Returns BrokenPipeError if remote connection is closed''' return self.send_command(name, addr, value) def read_register(self, name, addr, timeout=1): '''Takes name addr & timeout Returns - register value read if succeses None if read failed BrokenPipeError if remote connection is closed''' self._send_one_time_read(name, addr, timeout) # Since the 'rd_queue' won't get populated due to the raw_callback being registered # avoid waiting for the timeout here and just return None to unblock the call if self._raw_cmd_callback is not None and self.sync_read == False: return None try: data_to_return = self._connected_devices[name]['rd_queue'].get(block=True, timeout=timeout) except Empty: return None return data_to_return[0][1] def write_register_32(self, name, addr, value): '''Takes name, addr, value, writes 32-bit value to addr Returns BrokenPipeError if remote connection is closed''' lower_val = value & 0xFFFF upper_val = ((value >> 16) & 0xFFFF) upper_ret = self.write_register(name, addr+1, upper_val) if upper_ret == BrokenPipeError: return upper_ret lower_ret = self.write_register(name, addr, lower_val) return lower_ret def read_register_32(self, name, addr, timeout=1): '''Takes name addr & timeout Returns - 32-bit register value read if succeses None if read failed BrokenPipeError if remote connection is closed''' lower_bits = self._send_one_time_read_32(name, addr, timeout) if lower_bits == BrokenPipeError or lower_bits is None: return lower_bits # self._logger.info(f"read_register_32: lower_bits: {lower_bits}") upper_bits = self._send_one_time_read_32(name, addr+1, timeout) if upper_bits == BrokenPipeError or upper_bits is None: return upper_bits reg = (((upper_bits << 16) & 0xFFFF_0000) + lower_bits) # self._logger.info(f"read_register_32: upper_bits: {((upper_bits << 16) & 0xFFFF_0000)}") return reg # Return: True if sensor started succesfully (or non blocking call) # False if sensor failed to start def start_sensor(self, name, blocking=False, timeout=0, stream_list=()): '''Takes name, starts the sensor Optional keyword arguments: blocking - Make this API blocking. timeout - timeout to wait when blocking=True. stream_list - additional streams to enable when blocking on stream 35 Returns - True if sensor started succesfully (or non blocking call) False if sensor failed to start in alloted timeout BrokenPipeError if the remote connection is closed Dictionary of time spent in each sensor state ''' self.record(name, True) ret = self.write_register(name, self.reg_map['logic_module_ctrl'], LOGIC_MODULE_START_CMD) if ret == BrokenPipeError: return ret sleep_time = 0.001 sensor_state_dict = {2: "NaN", 3: "NaN", 4: "NaN", 5: "NaN", 6: "NaN", } if blocking is True and timeout != 0: time_to_wait = 0.0 stream_channel_list = ["logic_module_state"] if len(stream_list) > 3: self._logger.error("provided stream_list is greater than 3, bailing") return False, sensor_state_dict for stream in stream_list: stream_channel_list.append(stream) ret = self._start_data_stream(name, stream_channel_list) if ret == BrokenPipeError: return ret time.sleep(0.1) sensor_state = self.get_sensor_state(name) while sensor_state != self.state_map['magnetic_lock'] and self._connected_devices[name]['keep_running']: sensor_state = self.get_sensor_state(name) for key in sensor_state_dict: if sensor_state_dict[key] == "NaN" and key == sensor_state: sensor_state_dict[key] = time.time() # self._logger.debug(f"time: {sensor_state_dict[key]} sensor_state: {sensor_state}") break time.sleep(sleep_time) if time_to_wait > timeout: self._logger.error(f"sensor_state at timeout: {str(sensor_state)}") self._logger.error("Timeout reached, sensor never reached expected state, returning") return False, sensor_state_dict time_to_wait += sleep_time for key in sensor_state_dict: if key != 6 and sensor_state_dict[key] != "NaN" and sensor_state_dict[key+1] != "NaN": sensor_state_dict[key] = sensor_state_dict[key+1] - sensor_state_dict[key] return True, sensor_state_dict return True, sensor_state_dict def stop_sensor(self, name): '''Takes name, stops the sensor''' return self.write_register(name, self.reg_map['logic_module_ctrl'], LOGIC_MODULE_STOP_CMD) def enable_streams(self, name, list_of_streams): '''Takes name & list_of_streams, enables those streams''' ret = self.set_freq(self._freq) if ret == BrokenPipeError: return ret ret = self._start_data_stream(name, list_of_streams) if ret == BrokenPipeError: return ret self.record(name, True) def disable_streams(self, name): '''Takes name, disable streams for that sensor''' self.record(name, False) for name in self._connected_devices: if self.type == "scalar": sync_config_addr = self.reg_map['sync_config'] ret = self.send_command(name, addr=sync_config_addr, data=0x02) if ret == BrokenPipeError: return ret elif self.type == "pfn": pfn_csr_addr = self.pfn_reg_map['pfn_core'] ret = self.write_register_32(name, pfn_csr_addr, 0x10) if ret == BrokenPipeError: return ret def record(self, name, record=True): '''Takes name, Start or stop recording based on record Optional keyword arguments: record - Enable recording if True, disable recording if False ''' with self._lock: if not self._connected_devices[name]['recording'] and record: # TODO: Move this to _serial_loop() before calling _save_data() # so it can be decoupled from record() self._connected_devices[name]['sensor_state'] = 0 self._reset_data(name) self._connected_devices[name]['recording'] = record def get_data(self, name): '''Takes name Returns - recorded data as a numpy array, on success BrokenPipeError, on failure''' with self._lock: data = {} for ch in self._data: if self._connected_devices[name]['keep_running'] == True: data[ch] = np.array(self._data[ch]) else: return BrokenPipeError self._reset_data(name) return data def get_sn_sensor_base36(self, name): '''Takes name, returns base36 serial number of the Sensor on success, None on failure Checks internal data structure first and if valid just use that value. ''' if self._connected_devices[name]['sensor_serial_num'] != '': return self._connected_devices[name]['sensor_serial_num'] else: serial_num_int = self.get_sn_sensor_base10(name) if serial_num_int is not None: self._connected_devices[name]['sensor_serial_num'] = self._convert_to_base36((serial_num_int >> 8) & 0xffffffff) return self._convert_to_base36((serial_num_int >> 8) & 0xffffffff) return None def get_sn_sensor_base10(self, name): '''Takes name, Returns - base10 serial number of the Sensor on success, None on failure BrokenPipeError if remote connection is closed''' sn_reg = 0x7C first_word = self.read_register(name, sn_reg) sn_reg = sn_reg + 1 if first_word is None: return None if first_word == BrokenPipeError: return first_word second_word = self.read_register(name, sn_reg) sn_reg = sn_reg + 1 if second_word is None: return None if second_word == BrokenPipeError: return second_word third_word = self.read_register(name, sn_reg) sn_reg = sn_reg + 1 if third_word is None: return None if third_word == BrokenPipeError: return third_word forth_word = self.read_register(name, sn_reg) if forth_word is None: return None if forth_word == BrokenPipeError: return forth_word serial_num = ((forth_word << 48) & 0xffff_0000_0000_0000) serial_num = serial_num | ((third_word << 32) & 0x0000_ffff_0000_0000) serial_num = serial_num | ((second_word << 16) & 0x0000_0000_ffff_0000) serial_num = serial_num | ((first_word) & 0x0000_0000_0000_ffff) return serial_num def get_sn_scal_elec_base36(self, name, type="scalar"): '''Takes name and type, returns base36 serial number of the Scalar Electronics on success, None on failure Checks internal data structure first and if valid just use that value. ''' if self._connected_devices[name]['scalar_serial_num'] != '': return self._connected_devices[name]['scalar_serial_num'] else: serial_num_int = self.get_sn_scal_elec_base10(name, type) if serial_num_int is not None: self._connected_devices[name]['scalar_serial_num'] = self._convert_to_base36((serial_num_int >> 8) & 0xffffffff) return self._convert_to_base36((serial_num_int >> 8) & 0xffffffff) return None def get_sn_scal_elec_base10(self, name, type=""): '''Takes name and optional type, returns base10 serial number of the Scalar Electonics on success, None on failure''' if type == "scalar": sn_reg = 0x82 else: sn_reg = 0x7c first_word = self.read_register(name, sn_reg) sn_reg = sn_reg + 1 if first_word is None: return None if first_word == BrokenPipeError: return first_word second_word = self.read_register(name, sn_reg) sn_reg = sn_reg + 1 if second_word is None: return None if second_word == BrokenPipeError: return second_word third_word = self.read_register(name, sn_reg) sn_reg = sn_reg + 1 if third_word is None: return None if third_word == BrokenPipeError: return third_word forth_word = self.read_register(name, sn_reg) if forth_word is None: return None if forth_word == BrokenPipeError: return forth_word serial_num = ((forth_word << 48) & 0xffff_0000_0000_0000) serial_num = serial_num | ((third_word << 32) & 0x0000_ffff_0000_0000) serial_num = serial_num | ((second_word << 16) & 0x0000_0000_ffff_0000) serial_num = serial_num | ((first_word) & 0x0000_0000_0000_ffff) return serial_num def _convert_version_to_str(self, version): '''Takes Cyclone V version as a 32-bit int Returns a version string: v..-dirty ''' dirty_ver = (version & (0x80000000)) >> 31 major_ver = (version & (0x7C000000)) >> 26 minor_ver = (version & (0x03FC0000)) >> 18 patch_ver = (version & (0x0003FFFF)) ver_str = "v" + str(major_ver) + "." + str(minor_ver) + "." + str(patch_ver) if dirty_ver == 1: ver_str += ver_str + "-dirty" return ver_str def get_fw_version_str(self, name): '''Takes name, returns the Cyclone V version string on success, None on failure Checks internal data structure first and if valid just use that value. ''' if self._connected_devices[name]['cv_fw_version'] != '0.0.0': return self._connected_devices[name]['cv_fw_version'] else: version_lower = self.read_register(name, 0x80) # self._logger.info(f"version: {version}") if version_lower is None: return None if version_lower == BrokenPipeError: return version_lower version_upper = self.read_register(name, 0x81) # self._logger.info(f"version: {version}") if version_upper is None: return None if version_upper == BrokenPipeError: return version_upper version = ((version_upper << 16) & 0xFFFF_0000) | version_lower self._connected_devices[name]['cv_fw_version'] = self._convert_version_to_str(version) return self._convert_version_to_str(version) def get_serial_device_str(self, name): '''Takes name returns thread_name or serial_name''' try: if name in self._connected_devices.keys(): return self._connected_devices[name]['thread_name'] except: return None def get_sensor_sn_from_serial_device(self, serial_name): '''Takes serial_name returns sensor_sn or None''' try: for name in self._connected_devices.keys(): if self._connected_devices[name]['thread_name'] in serial_name: return self._connected_devices[name]['sensor_serial_num'] else: return None except: return None def _wavegen_shape_from_str(self, shape): '''takes shape, returns integer for that shape''' if shape == "DC": return 0 elif shape == "Square": return 1 elif shape == "Sawtooth": return 2 elif shape == "Triangle": return 3 elif shape == "Sin": return 4 elif shape == "Cos": return 5 elif shape == "TRNG": return 6 else: return WAVEGEN_SHAPE_MAX def _wavegen_shape_to_str(self, shape): '''takes int, returns the string of that shape''' if shape == 0: return "DC" elif shape == 1: return "Square" elif shape == 2: return "Sawtooth" elif shape == 3: return "Triangle" elif shape == 4: return "Sin" elif shape == 5: return "Cos" elif shape == 6: return "TRNG" else: return "unknown shape" def _convert_to_coil_mask(self, list_of_ints): mask = 0 for i in list_of_ints: if i < 28: mask = mask | (1 << i) return mask def wavegen_set_dc_offset(self, name, i, dc_offset): '''takes name, i, dc-offset, sets wavegen i's dc-offset''' print(f"wavegen_set_dc_offset i: {i} dc_offset: {dc_offset}") addr = (0x20 + (i*0x8) + 6) raw_dc_offset = int(round(dc_offset * 32767)) return self.write_register_32(name, addr, raw_dc_offset) def wavegen_set_amplitude(self, name, i, amplitude): '''takes name, i, amplitude, sets wavegen i's amplitude''' print(f"wavegen_set_amplitude i: {i} amplitude: {amplitude}") addr = (0x20 + (i*0x8) + 4) raw_amplitude = int(round(amplitude * 32767)) return self.write_register_32(name, addr, raw_amplitude) def wavegen_set_frequency(self, name, i, frequency): '''takes name, i, frequency, sets wavegen i's frequency''' print(f"wavegen_set_frequency i: {i} frequency: {frequency}") addr = (0x20 + (i*0x8) + 2) freq_val = int(((frequency / ZFS_WAVEGEN_CLK_HZ) * 2**32)) print(f"wavegen_set_frequency i: {i} freq_val: {freq_val}") return self.write_register_32(name, addr, freq_val) def wavegen_set_phase(self, name, i, phase): '''takes name, i, phase, sets wavegen i's phase''' print(f"wavegen_set_phase i: {i} phase: {phase}") phase = ((phase << 16) & 0xFFFF_0000) addr = (0x20 + (i*0x8) + 0) current_val = self.read_register_32(name, addr) current_val = (current_val & 0x0000_FFFF) | phase return self.write_register_32(name, addr, current_val) def wavegen_set_shape(self, name, i, shape): '''takes name, i, shape, sets wavegen i to shape''' print(f"wavegen_set_shape i: {i} shape: {shape}") addr = (0x20 + (i*0x8) + 0) current_val = self.read_register_32(name, addr) current_val = (current_val & 0xFFFF_FFF0) | shape return self.write_register_32(name, addr, current_val) def wavegen_set_coil_zero(self, name, mask): '''takes name and mask, applies the mask to coil zero''' print(f"wavegen_set_coil_zero mask: {mask}") coil_addr = self.pfn_reg_map['pfn_coil_zero'] val = 0 if mask != 0: val = ((mask << 4) & 0xFFFF_FFF0) | 1 return self.write_register_32(name, coil_addr, val) def wavegen_set_coil_one(self, name, mask): '''takes name and mask, applies the mask to coil one''' print(f"wavegen_set_coil_one mask: {mask}") coil_addr = self.pfn_reg_map['pfn_coil_one'] val = 0 if mask != 0: val = ((mask << 4) & 0xFFFF_FFF0) | 1 return self.write_register_32(name, coil_addr, val) def wavegen_set_coil_two(self, name, mask): '''takes name and mask, applies the mask to coil two''' print(f"wavegen_set_coil_two mask: {mask}") coil_addr = self.pfn_reg_map['pfn_coil_two'] val = 0 if mask != 0: val = ((mask << 4) & 0xFFFF_FFF0) | 1 return self.write_register_32(name, coil_addr, val) def wavegen_enable_current_driver(self, name): '''takes name, sets the CD_EN bit in the PFN_CORE_CSR register''' print(f"wavegen_enable_current_driver") pfn_core_csr_addr = self.pfn_reg_map['pfn_core'] pfn_core_val = self.read_register_32(name, pfn_core_csr_addr) pfn_core_val = pfn_core_val | 0x8000 return self.write_register_32(name, pfn_core_csr_addr, pfn_core_val) def wavegen_disable_current_driver(self, name): '''takes name, clears the CD_EN bit in the PFN_CORE_CSR register''' print(f"wavegen_disable_current_driver") pfn_core_csr_addr = self.pfn_reg_map['pfn_core'] pfn_core_val = self.read_register_32(name, pfn_core_csr_addr) pfn_core_val = pfn_core_val & ~0x00008000 return self.write_register_32(name, pfn_core_csr_addr, pfn_core_val) def wavegen_reset(self, name): '''takes name, sets the NCO Reset bit in the PFN_CORE_CSR register''' print(f"wavegen_reset") pfn_core_csr_addr = self.pfn_reg_map['pfn_core'] pfn_core_val = self.read_register_32(name, pfn_core_csr_addr) pfn_core_val = pfn_core_val | 0x0002 return self.write_register_32(name, pfn_core_csr_addr, pfn_core_val) def wavegen_dump(self, name, i, writer=None): '''takes name, i, & optional writer, dumps the wavegen config for i''' coil_zero_addr = self.pfn_reg_map['pfn_coil_zero'] coil_one_addr = self.pfn_reg_map['pfn_coil_one'] coil_two_addr = self.pfn_reg_map['pfn_coil_two'] if i is not None: print(f"wavegen_dump: i {i}") ctrl_addr = (0x20 + (i*0x8) + 0) freq_addr = (0x20 + (i*0x8) + 2) amp_addr = (0x20 + (i*0x8) + 4) offset_addr = (0x20 + (i*0x8) + 6) wavegen_freq = self.read_register_32(name, freq_addr) wavegen_control = self.read_register_32(name, ctrl_addr) wavegen_amp = self.read_register_32(name, amp_addr) wavegen_offset = self.read_register_32(name, offset_addr) coil_zero = self.read_register_32(name, coil_zero_addr) coil_one = self.read_register_32(name, coil_one_addr) coil_two = self.read_register_32(name, coil_two_addr) if writer is not None: if i is not None: writer.write(f"wavegen_dump: i {i}\r\n") writer.write(f"wavegen_control: {hex(wavegen_control)}\r\n") shape_bits = wavegen_control & 0xF writer.write(f"shape: {self._wavegen_shape_to_str(shape_bits)}\r\n") phase_bits = wavegen_control & 0xFFFF_0000 writer.write(f"phase_bits: {hex(phase_bits)}\r\n") writer.write(f"wavegen_freq: {hex(wavegen_freq)}\r\n") writer.write(f"wavegen_amp: {hex(wavegen_amp)}\r\n") writer.write(f"wavegen_offset: {hex(wavegen_offset)}\r\n") writer.write(f"coil_zero: {hex(coil_zero)}\r\n") writer.write(f"coil_one: {hex(coil_one)}\r\n") writer.write(f"coil_two: {hex(coil_two)}\r\n") else: if i is not None: print(f"wavegen_control: {hex(wavegen_control)}") shape_bits = wavegen_control & 0xF print(f"shape: {self._wavegen_shape_to_str(shape_bits)}") phase_bits = wavegen_control & 0xFFFF_0000 print(f"phase_bits: {hex(phase_bits)}") print(f"wavegen_freq: {hex(wavegen_freq)}") print(f"wavegen_amp: {hex(wavegen_amp)}") print(f"wavegen_offset: {hex(wavegen_offset)}") print(f"coil_zero: {hex(coil_zero)}") print(f"coil_one: {hex(coil_one)}") print(f"coil_two: {hex(coil_two)}") def configure_wavegen_telnet(self, dev_name, args, writer=None): '''takes dev_name & args, configures wavegen when called from telnet Differs from configure_wavegen() by not calling api.stop() at the end''' do_dump = False do_en_cd = False do_di_cd = False do_reset = False do_shape = False do_freq = False do_phase = False do_amp = False do_dcoff = False do_coil_zero = False do_coil_one = False do_coil_two = False shape_id = WAVEGEN_SHAPE_MAX coil_zero_mask = 0 coil_one_mask = 0 coil_two_mask = 0 # Wavegen control if args.index is not None: if args.index >= 0 and args.index <= WAVEGEN_INDEX_MAX: print("wavegen index provided, setting up device for wavegen") if args.index < 0 or args.index >= WAVEGEN_INDEX_MAX: print(f"invalid index {args.index} provided, quitting") return -1 if args.shape is not None: shape_id = self._wavegen_shape_from_str(args.shape) if shape_id < 0 or shape_id >= WAVEGEN_SHAPE_MAX: print(f"invalid shape {args.shape} provided, quitting") return -1 else: do_shape = True if args.frequency is not None: if args.frequency < 0 or args.frequency > ZFS_WAVEGEN_CLK_HZ: print(f"invalid frequency {args.frequency} provided, quitting") return -1 else: do_freq = True if args.phase is not None: if args.phase < 0 or args.phase > 360: print(f"invalid phase {args.phase} provided, quitting") return -1 else: do_phase = True if args.amplitude is not None: if args.amplitude < -1 or args.amplitude > 1: print(f"amplitude out of range [-1,1]") return -1 do_amp = True if args.dc_offset is not None: if args.dc_offset < -1 or args.dc_offset > 1: print(f"DC offset out of range [-1,1]") return -1 do_dcoff = True if args.dump_config: do_dump = True if args.en_cd: do_en_cd = True if args.di_cd: do_di_cd = True if args.reset: do_reset = True if args.coil_zero is not None: coil_zero_mask = self._convert_to_coil_mask(args.coil_zero) do_coil_zero = True if args.coil_one is not None: coil_one_mask = self._convert_to_coil_mask(args.coil_one) do_coil_one = True if args.coil_two is not None: coil_two_mask = self._convert_to_coil_mask(args.coil_two) do_coil_two = True pfn_core_csr_addr = self.pfn_reg_map['pfn_core'] pfn_core_csr = self.read_register_32(dev_name, pfn_core_csr_addr) print(f"PFN Core CSR: {hex(pfn_core_csr)}") if (pfn_core_csr & 0x1000_0000 != 0x1000_0000): print("detected invalid device, quitting") return -1 pfn_version_addr = self.pfn_reg_map['version'] version = self.read_register_32(dev_name, pfn_version_addr) version_str = self._convert_version_to_str(version) print(f"PFN version: {version_str}") if do_amp: self.wavegen_set_amplitude(dev_name, args.index, args.amplitude) if do_dcoff: self.wavegen_set_dc_offset(dev_name, args.index, args.dc_offset) if do_freq: self.wavegen_set_frequency(dev_name, args.index, args.frequency) if do_phase: self.wavegen_set_phase(dev_name, args.index, args.phase) if do_shape: self.wavegen_set_shape(dev_name, args.index, shape_id) if do_coil_zero: self.wavegen_set_coil_zero(dev_name, coil_zero_mask) if do_coil_one: self.wavegen_set_coil_one(dev_name, coil_one_mask) if do_coil_two: self.wavegen_set_coil_two(dev_name, coil_two_mask) if do_en_cd: self.wavegen_enable_current_driver(dev_name) if do_di_cd: self.wavegen_disable_current_driver(dev_name) if do_reset: self.wavegen_reset(dev_name) if do_dump: self.wavegen_dump(dev_name, args.index, writer=writer) print("end of configure_wavegen_telnet()") return 0 def configure_wavegen(self, name, args): '''takes args and transport, configures wavegen''' do_dump = False do_shape = False do_freq = False do_phase = False do_amp = False do_dcoff = False do_coil_zero = False do_coil_one = False do_coil_two = False shape_id = WAVEGEN_SHAPE_MAX # Wavegen control if args.index is not None: if args.index >= 0 and args.index <= WAVEGEN_INDEX_MAX: print("wavegen index provided, setting up device for wavegen") if args.index < 0 or args.index >= WAVEGEN_INDEX_MAX: print(f"invalid index {args.index} provided, quitting") self.stop() sys.exit() if args.shape is not None: shape_id = self._wavegen_shape_from_str(args.shape) if shape_id < 0 or shape_id >= WAVEGEN_SHAPE_MAX: print(f"invalid shape {args.shape} provided, quitting") self.stop() sys.exit() else: do_shape = True if args.dump_config: do_dump = True if args.frequency is not None: if args.frequency < 0 or args.frequency > ZFS_WAVEGEN_CLK_HZ: print(f"invalid frequency {args.frequency} provided, quitting") self.stop() sys.exit() else: do_freq = True if args.phase is not None: if args.phase < 0 or args.phase > 360: print(f"invalid phase {args.phase} provided, quitting") self.stop() sys.exit() else: do_phase = True if args.amplitude is not None: do_amp = True if args.dc_offset is not None: if args.dc_offset >= 0 and args.dc_offset <= 0xFFFF: do_dcoff = True if args.coil_zero is not None: if args.coil_zero >= 0 and args.coil_zero <= 0x0FFF_FFFF: do_coil_zero = True if args.coil_one is not None: if args.coil_one >= 0 and args.coil_one <= 0x0FFF_FFFF: do_coil_one = True if args.coil_two is not None: if args.coil_two >= 0 and args.coil_two <= 0x0FFF_FFFF: do_coil_two = True self.wavegen_enable_current_driver(name) pfn_core_csr_addr = self.pfn_reg_map['pfn_core'] pfn_core_csr = self.read_register_32(name, pfn_core_csr_addr) print(f"PFN Core CSR: {hex(pfn_core_csr)}") if (pfn_core_csr & 0x1000_0000 != 0x1000_0000): print("detected invalid device, quitting") self.stop() return pfn_version_addr = self.pfn_reg_map['version'] version = self.read_register_32(name, pfn_version_addr) version_str = self._convert_version_to_str(version) print(f"PFN version: {version_str}") if do_amp: self.wavegen_set_amplitude(name, args.index, args.amplitude) if do_dcoff: self.wavegen_set_dc_offset(name, args.index, args.dc_offset) if do_freq: self.wavegen_set_frequency(name, args.index, args.frequency) if do_phase: self.wavegen_set_phase(name, args.index, args.phase) if do_shape: self.wavegen_set_shape(name, args.index, shape_id) if do_coil_zero: self.wavegen_set_coil_zero(name, args.coil_zero) if do_coil_one: self.wavegen_set_coil_one(name, args.coil_one) if do_coil_two: self.wavegen_set_coil_two(name, args.coil_two) if do_dump: self.wavegen_dump(name, args.index) self.stop() print("end of configure_wavegen()") return pending_sig = None def sig_handler(signum, frame): global pending_sig print(f'Signal received: {signum}') pending_sig = signum def scalar_record(args, scalar, dev_name, record_time): ''' Start a recording for record_time on device dev_name Args: args: arguments dictionary from argparse lib. scalar: fli_scalar_api object dev_name: Scalar serial_number or device name record_time: how long to record for Return: 0 if success -1 on error ''' timeout = 300 sample_rate = 1000 processing_time = 0 streams_to_record = ['dds_output_freq', 'pps_clock_count', 'raw_pps_output'] print(f"starting sensor {dev_name}, this process may take a while, up to {timeout} seconds") ret, sensor_state_dict = scalar.start_sensor(dev_name, True, timeout, streams_to_record) if ret is False: print('Sensor failed to start: FAILED') print(f'failed to start the sensor {dev_name}... quitting') return ret row_header = ['time'] if dev_name is not None: for stream in streams_to_record: row_header.append(f'{dev_name}-{stream}') try: csv_file = open(args.filename, 'a', newline='') csv_writer = csv.writer(csv_file, delimiter=',') except Exception as e: print(f'caught exception {e}, quitting') return -1 # print(f'creating row with header: {row_header}') csv_writer.writerow(row_header) scalar.set_freq(sample_rate) scalar.enable_streams(dev_name, streams_to_record) print(f'start recording for {record_time} seconds at a sample rate of {sample_rate} save the data to disk at {args.filename}') while 1: global pending_sig if pending_sig is not None: print(f'aborting on signal {pending_sig}...') break if record_time - processing_time > 0: sleep(record_time - processing_time) t0 = perf_counter() recorded_data = scalar.get_data(dev_name) if recorded_data is None: print('recorded_data was None, quitting') return -1 N = len(recorded_data['time']) time_samples = recorded_data['time'] maq = recorded_data[streams_to_record[0]] pps = recorded_data[streams_to_record[1]] raw_pps = recorded_data[streams_to_record[2]] for (idx, _) in enumerate(recorded_data['time']): csv_writer.writerow([time_samples[idx], maq[idx], pps[idx], raw_pps[idx]]) t1 = perf_counter() processing_time = t1 - t0 print(f'Processed {N} samples in {processing_time} s') # Break out of this loop as we only want to record record_time of data break return 0 if __name__ == "__main__": signal.signal(signal.SIGINT, sig_handler) signal.signal(signal.SIGTERM, sig_handler) parser = argparse.ArgumentParser(description='FieldLine Industries Scalar Magnetometer API') parser.add_argument('-a', '--address', help='Network address') parser.add_argument("-b", "--baud", type=int, required=False, default=921600, choices=[115200, 230400, 460800, 921600], help="serial port baud") parser.add_argument('-d', '--device', help='Serial port device. E.g. /dev/ttyUSB0 on Linux or COM1 on Windows') parser.add_argument('-l', '--list', action='store_true', help='List serial ports which may be connected to a scalar magnetometer') parser.add_argument('-p', '--port', type=int, default=2000, help='Network Port. default: 2000') parser.add_argument('-q', '--query', action='store_true', help='Query serial number of scalar') parser.add_argument('--scratch', action='store_true', help='Write to scratch register and read it back') parser.add_argument('--startstop', action='store_true', help='Start the sensor') parser.add_argument('--record', action='store_true', help='Start the sensor and record 10 seconds of data') parser.add_argument('--filename', default="scalar_recording.csv", help='The filename to save the recording to') parser.add_argument('--version', action='store_true', help='The version of this API') # TODO: Consider making wavegen a subcommand? so we can run fli_scalar_api.py --wavegen [extra args] to make args parsing simpler parser.add_argument('-c', '--dump-config', action='store_true', help='Dump current wavegen config') parser.add_argument('-i', '--index', type=int, help='Wavegen index') parser.add_argument('--shape', type=str, choices=['DC', 'Square', 'Sawtooth', 'Triangle', 'Sin', 'Cos', 'TRNG'], help='Set waveform shape') parser.add_argument('-f', '--frequency', type=float, help='Set waveform frequency in Hz') parser.add_argument('--phase', type=int, help='Set waveform phase in degrees') parser.add_argument('--amplitude', type=float, help='Set waveform amplitude (A)') parser.add_argument('--dc-offset', type=float, help='Set waveform DC offset (A)') parser.add_argument('--coil-zero', type=int, help='Set coil zero (x axis) to provided mask') parser.add_argument('--coil-one', type=int, help='Set coil one (y axis) to provided mask') parser.add_argument('--coil-two', type=int, help='Set coil two (z axis) to provided mask') args = parser.parse_args() if args.version: print(f"APP_VERSION: {APP_VERSION}") sys.exit() scalar = FliScalarApi() if args.address is not None: print(f"address:port: {args.address}:{args.port}") dev = scalar.connect_to_net_device(args.address, args.port) if dev == False: print(f"failed to connect to device: {args.address}:{args.port}") scalar.stop() sys.exit() dev_name = scalar.get_connected_device_list()[0] print(f"Connected to {dev_name}") if args.index is not None: scalar.configure_wavegen(dev_name, args) elif args.device is not None: if not scalar.connect_to_device(args.device, args.baud): print(f"Failed to connect to device {args.device} exiting...") scalar.stop() sys.exit() dev_name = scalar.get_connected_device_list()[0] print(f"Connected to {dev_name}") if args.index is not None: scalar.configure_wavegen(dev_name, args) else: print("Device or addresss required, quitting") scalar.stop() sys.exit() if args.list: print('Serial Ports:') for dev in scalar.get_human_readable_device_list(): print(f' {dev}') elif args.address: print(f"address:port: {args.address}:{args.port}") dev = scalar.connect_to_net_device(args.address, args.port) if dev == False: print(f"failed to connect to device: {args.address}:{args.port}") sys.exit() dev_name = scalar.get_connected_device_list()[0] print(f"Connected to {dev_name}") if args.query: cv_version = scalar.get_fw_version_str(dev_name) if cv_version is not None: print(f"Cyclone V Version: {cv_version}") else: print("Failed to read CV Version, check device") scalar.stop() sys.exit() sensor_serial_num_base36 = scalar.get_sn_sensor_base36(dev_name) if sensor_serial_num_base36 is not None: print(f"Sensor serial number base36 format {sensor_serial_num_base36}") else: print("Failed to read Sensor serial number base36 format, check device") scalar.stop() sys.exit() serial_num_int = scalar.get_sn_sensor_base10(dev_name) if serial_num_int is not None: print(f"Sensor serial number {hex(serial_num_int)}") else: print("Failed to read Sensor serial number, check device") scalar.stop() sys.exit() serial_num_int = scalar.get_sn_scal_elec_base10(dev_name, "scalar") if serial_num_int is not None: print(f"Scalar Electronics serial number {hex(serial_num_int)}") else: print("Failed to read Scalar Electronics serial number, check device") scalar.stop() sys.exit() elec_serial_num_base36 = scalar.get_sn_scal_elec_base36(dev_name, "scalar") if elec_serial_num_base36 is not None: print(f"Scalar Electronics serial number base36 format {elec_serial_num_base36}") else: print("Failed to read Scalar Electronics serial number base36 format, check device") scalar.stop() sys.exit() if args.scratch: scratch_reg_addr = scalar.reg_map['scratch'] wr_value = 0xdeaf for i in range(0x10): wr_value = i scalar.write_register(dev_name, scratch_reg_addr, i) rd_value = scalar.read_register(dev_name, scratch_reg_addr) if rd_value is not None: print(f"Scratch reg: wrote 0x{wr_value:x}, read 0x{rd_value:x}") if rd_value != wr_value: print("rd_value != wr_value FAILED, exiting loop") break else: print("rd_val returned is None: FAILED, exiting loop") break # scalar.write_register(dev_name, scratch_reg_addr, wr_value) # rd_value = scalar.read_register(dev_name, scratch_reg_addr) # print(f"Scratch reg: wrote 0x{wr_value:x}, read 0x{rd_value:x}") if args.startstop: print("Starting sensor...") ret = scalar.start_sensor(dev_name, True, timeout=300) if ret == True: print("scalar started!") else: print("scalar failed to started!") print("Stopping sensor") scalar.stop_sensor(dev_name) scalar.stop() elif args.query or args.scratch or args.startstop: if args.query: cv_version = scalar.get_fw_version_str(dev_name) if cv_version is not None: print(f"Cyclone V Version: {cv_version}") else: print("Failed to read CV Version, check device") scalar.stop() sys.exit() sensor_serial_num_base36 = scalar.get_sn_sensor_base36(dev_name) if sensor_serial_num_base36 is not None: print(f"Sensor serial number base36 format {sensor_serial_num_base36}") else: print("Failed to read Sensor serial number base36 format, check device") scalar.stop() sys.exit() serial_num_int = scalar.get_sn_sensor_base10(dev_name) if serial_num_int is not None: print(f"Sensor serial number {hex(serial_num_int)}") else: print("Failed to read Sensor serial number, check device") scalar.stop() sys.exit() serial_num_int = scalar.get_sn_scal_elec_base10(dev_name, "scalar") if serial_num_int is not None: print(f"Scalar Electioncs serial number {hex(serial_num_int)}") else: print("Failed to read Scalar Electioncs serial number, check device") scalar.stop() sys.exit() serial_num_base36 = scalar.get_sn_scal_elec_base36(dev_name, "scalar") if serial_num_base36 is not None: print(f"Scalar Electronics serial number base36 format {serial_num_base36}") else: print("Failed to read Scalar Electronics serial number, check device") scalar.stop() sys.exit() if args.scratch: scratch_reg_addr = scalar.reg_map['scratch'] wr_value = 0xdeaf scalar.write_register(dev_name, scratch_reg_addr, wr_value) rd_value = scalar.read_register(dev_name, scratch_reg_addr) if rd_value is not None: print(f"Scratch reg: wrote 0x{wr_value:x}, read 0x{rd_value:x}") else: print(f"Scratch reg: wrote 0x{wr_value:x}, read None... read or write FAILED") if args.startstop: print("Starting sensor...") ret = scalar.start_sensor(dev_name, True, timeout=300) if ret == True: print("scalar started!") else: print("scalar failed to started!") print("Stopping sensor") scalar.stop_sensor(dev_name) scalar.stop() elif args.record: record_ret = scalar_record(args, scalar, dev_name, 10) if record_ret != 0: print(f'recording failed with error: {record_ret}') scalar.stop_sensor(dev_name) scalar.stop() sys.exit() scalar.stop()