import time
from collections.abc import Iterable
from collections.abc import Iterable as Iterable_T
from contextlib import contextmanager
from typing import Optional, Union
import pandas as pd
import teensytoany
from multiuserfilelock import MultiUserFileLock, Timeout
from packaging.version import Version
from teensytoany import TeensyToAny
from ..util.util import MULTIUSERFILELOCK_GROUP, MULTIUSERFILELOCK_TMPDIR
known_devices = pd.json_normalize([
{
'teensy_serial_number': '7039070',
'connected_device_adddresses': [0x66 << 1, 0x68 << 1],
}
])
known_devices = known_devices.set_index('teensy_serial_number')
known_serial_numbers = list(known_devices.index)
locks_dir = MULTIUSERFILELOCK_TMPDIR / 'owl'
[docs]
class MultiLidar:
"""Open a Ramona Optics Multi-Lidar ranging system.
Example
-------
>>> from owl.instruments import MultiLidar
>>> lidar = MultiLidar()
>>> print(lidar.distance)
Parameters
----------
serial_number: str
Serial_number of microcontroller. If not provided, auto-detection
of the microcontroller will be attempted.
device_addresses: int or List[int]
For devices shipped with Ramona Optics products, this should be
left as ``None``. Device addresses of Lidar units. 8-bit addresses
are assumed. One address can be provided as a single integer.
Multiple addresses can be provided as a list.
open_device: bool
If ``True``, the last operation of the constructor will be to call
the ``open`` method to ensure the device is ready for reading.
.. versionadded:: 0.18.11
The ``open_device`` parameter.
"""
# register addresses
_ACQ_COMMANDS = 0x00
_FULL_DELAY_LOW = 0x10
_FULL_DELAY_HIGH = 0x11
_UNIT_ID = 0x16
_I2C_SEC_ADDR = 0x1A
_I2C_CONFIG = 0x1B
_CP_VER_LO = 0x72
_CP_VER_HI = 0x73
_BOARD_TEMPERATURE = 0xE0
_HARDWARE_VERSION = 0xE1
_FACTORY_RESET = 0xE4
_ENABLE_FLASH_STORAGE = 0xEA
_SOC_TEMPERATURE = 0xEC
_ENABLE_ANT_RADIO = 0xF0
def __init__(self, *,
serial_number: str=None,
device_addresses: Optional[Union[int, Iterable_T[int]]]=None,
open_device=True,
**kwargs) -> None:
# Notes
# -----
# To obtain the 8-bit address from a 7-bit address use:
# >>> address_8bit = address_7bit << 1
if (device_addresses is not None) and not isinstance(device_addresses, Iterable):
# make it a tuple
device_addresses = (device_addresses, )
self._broadcast_address = 0xC4
# hidden argument to disable the use of a filelock
self._use_instrument_lock = kwargs.get('use_instrument_lock', True)
self._bias_correction = False
self._use_lock = True
self._lock = None
self._device_addresses = device_addresses
self._serial_number = serial_number
self._teensy = None
if open_device:
self.open(_stacklevel_increment=1)
@property
def distance(self) -> list[float]:
"""The distance as read from the sensor in meters.
This returns a list of floating point values corresponding to the
reading from each lidar sensor.
"""
# Disable bias correction
self._take_measurement(self._bias_correction)
distance = []
for device_address in self._device_addresses:
low_byte = self._read(device_address, self._FULL_DELAY_LOW)
high_byte = self._read(device_address, self._FULL_DELAY_HIGH)
distance_cm = (high_byte << 8) | low_byte
distance.append(distance_cm * 0.01)
return distance
@property
def lidar_count(self):
"""The number of lidar sensors connected to the module.
.. versionadded:: 0.18.11
"""
if self._device_addresses is None:
return 0
else:
return len(self._device_addresses)
[docs]
def open(self, *, _stacklevel_increment=0) -> None:
"""Open the device for communication
See also
--------
MultiLidar, close
"""
serial_number = self._serial_number
if serial_number is None:
serial_number = TeensyToAny.list_all_serial_numbers(
known_serial_numbers, device_name="Lidar",
)[0]
if serial_number not in known_serial_numbers:
from warnings import warn
warn(f"The serial number {serial_number} is not known to Ramona Optics.",
stacklevel=2 + _stacklevel_increment)
if self._device_addresses is None and serial_number not in known_serial_numbers:
raise ValueError("Device addresses must be provided if the "
"serial number is not known to Ramona Optics.")
if self._device_addresses is None:
self._device_addresses = known_devices.loc[serial_number]['connected_device_adddresses']
self._lock_acquire(serial_number)
try:
# Backward compatibility if statement added in July 18, 2025
# Remove on Sept 1st, 2025
if Version(teensytoany.__version__) >= Version("0.11.1"):
self._teensy = TeensyToAny(serial_number, device_name="Lidar",)
else:
self._teensy = TeensyToAny(serial_number)
# device contains 8bit addressable registers.
self._teensy.i2c_init(register_space=1)
except Exception as e:
self._lock_release()
raise e
[docs]
def close(self) -> None:
"""Close the device for communication
See also
--------
MultiLidar, open
."""
if self._teensy is not None:
self._teensy.close()
self._teensy = None
self._lock_release()
@property
def _unit_id(self) -> list[int]:
unit_id = []
for device_address in self._device_addresses:
bytes_read = [self._read(device_address, self._UNIT_ID + i)
for i in range(4)]
# >>> hex(int.from_bytes([4, 3, 2, 1], byteorder='big'), signed=False)
# '0x4030201'
value = int.from_bytes(bytes_read, byte_order='big', signed=False)
unit_id.append(value)
return unit_id
@staticmethod
def _make_lock(
serial_number, group=MULTIUSERFILELOCK_GROUP, chmod=0o666) -> MultiUserFileLock:
# 0o666 is chosen because that is the default permission set
# by most people installing the teensy by default
# https://www.pjrc.com/teensy/loader_linux.html
# Check the udev rules file
unique_lidar_locktxt = locks_dir / f"lidar_{serial_number}.lock"
# lock will only be called if the device is closed
# (when isOpen is called).
return MultiUserFileLock(unique_lidar_locktxt,
group=group, chmod=chmod,
timeout=0.001)
def _lock_acquire(self, serial_number) -> None:
if not self._use_lock:
# If the use isn't requesting to use a lock, simply return
# immediately
return
lock = self._make_lock(serial_number)
try:
lock.acquire()
except Timeout:
raise RuntimeError(
"This lidar system has been opened already. "
"Establish a new connection by closing this system in the other program."
)
# Only assign the new lock object after it has been acquired.
self._lock = lock
def _lock_release(self) -> None:
# During garbage collection, the serial
# device might have been closed first
# Make sure we cleanup the lock in either case
if self._lock is not None:
self._lock.release()
self._lock = None
@property
def serial_number(self):
return self._serial_number
_i2c_configuration_values = {
'default': 0x00,
'secondary': 0x01,
'both': 0x02,
}
def _set_i2c_configuration(self, device_address: int, value: str) -> None:
"""Internal method to set how the devices respond to different I2C.
This enables one to setup the device to respond to either:
* Only the default address ``0xC4`` (8-bit).
* Only the secondary address (As defined by the I2C address register)
* Both addresses
Note
----
Since this method is rather specialized, we don't expose it to end
users.
"""
value = value.lower()
try:
value_int = self._i2c_configuration_values[value]
except KeyError as e:
valid_i2c_configurations = tuple(self._i2c_configuration_values.keys())
raise ValueError(
f"i2c_configuration must be one of {valid_i2c_configurations},"
f"got {value}.") from e
self._write(self._I2C_CONFIG, value_int)
def _change_i2c_address(self, new_address: int,
*,
persist_after_reboot: bool=True,
i2c_configuration: str='both') -> None:
"""Change the LIDAR device's I2C address.
Parameters
----------
new_address: int
New address the Lidar will use on the I2C bus. This action will
permanently change the address even after it is restarted.
The address should be an even number in the rage of
``0x12`` and ``0xEE`` (not including ``0xEE``).
persist_after_reboot:
If ``True``, the change will be made persistent even after the
device is restarted.
i2c_configuration: 'default', 'secondary', 'both'
Specifies whether or not the device should respond to the default
(``0xC4``), secondary (provided as the ``new_address`` parameter
in this function), or both I2C addresses.
Note
----
This function requires the ability to write 5 bytes data which
is not implemented in the current version of TeensyToAny.
"""
valid_i2c_configurations = tuple(self._i2c_configuration_values.keys())
i2c_configuration = i2c_configuration.lower()
# We check that the I2C configuration is valid early so as to avoid
# a late stage error
if i2c_configuration not in valid_i2c_configurations:
raise ValueError(
f"i2c_configuration must be one of {valid_i2c_configurations},"
f"got {i2c_configuration}.")
if new_address not in range(0x12, 0xEE, 2):
raise ValueError(
"New address must be an even number between "
f"0x12 and 0xEE (not including 0xEE), got 0x{new_address:02x}."
)
new_address_7bit = new_address >> 1
# Get the unit ID. We need it to "unlock" changing the I2C address.
unit_id = [self._read(self._broadcast_address, i)
for i in range(self._UNIT_ID, self._I2C_SEC_ADDR)]
unit_id.append(new_address_7bit)
with self._make_persistent(persist_after_reboot):
# change the I2C address by writing the required 5 byte
# sequence starting at the UNIT_ID address
self._teensy.i2c_write_payload(self._UNIT_ID, unit_id)
time.sleep(0.1)
self.i2c_configuration = i2c_configuration
self._device_addresses = new_address
@property
def bias_correction(self) -> bool:
return self._bias_correction
@bias_correction.setter
def bias_correction(self, value: bool):
value = bool(value)
self._bias_correction = value
def _factory_reset(self) -> None:
"""Reset all LIDAR configuration to their factory defaults."""
self._write(self._broadcast_address, self._FACTORY_RESET, 0x01)
@property
def _flash_enabled(self) -> bool:
# value is 0x11 when enabled, 0x00 when disabled
value = self._read(self._broadcast_address, self._ENABLE_FLASH_STORAGE)
return value == 0x11
@_flash_enabled.setter
def _flash_enabled(self, value: bool) -> None:
if value:
value_int = 0x11
else:
value_int = 0x00
self._write(self._broadcast_address,
self._ENABLE_FLASH_STORAGE, value_int)
@contextmanager
def _make_persistent(self, persist: bool=True, wait_until_ready: bool=True) -> None:
previous_value = self._flash_enabled
self._flash_enabled = persist
if wait_until_ready:
time.sleep(0.1)
yield self
self._flash_enabled = previous_value
# sleep???
def _take_measurement(self, bias_correction: bool) -> None:
"""
Parameters
----------
bias_correction:
If ``True`` enable correction bias on the receiver.
Notes
-----
The datasheet states:
Take distance measurement with/without receiver bias correction.
"""
if bias_correction:
value = 0x04
else:
value = 0x03
self._write(self._broadcast_address, self._ACQ_COMMANDS, value)
def _read(self, device_address: int, register_address: int) -> int:
if not self._teensy:
raise RuntimeError("Device is not connected for communication. "
"Please call the open method first.")
return self._teensy.i2c_read_uint8(device_address, register_address)
def _write(self, device_address: int, register_address: int, data: int) -> None:
if not self._teensy:
raise RuntimeError("Device is not connected for communication. "
"Please call the open method first.")
self._teensy.i2c_write_uint8(device_address, register_address, data)
@property
def temperature_C(self) -> list[float]:
"""The temperature of the device in degrees Celsius.
Multiple values are returned as a list, one value for each Lidar unit.
"""
temperature = []
for device_address in self._device_addresses:
value = self._read(device_address, self._BOARD_TEMPERATURE)
# It is stored as an 8 bit 2's compliment signed number
if value >= 128:
value = value - 256
value = float(value)
temperature.append(value)
return temperature
@property
def soc_temperature_C(self) -> list[float]:
"""The temperature of the system on a chip (SOC) in Celsius."""
temperature = []
for device_address in self._device_addresses:
value = self._read(device_address, self._SOC_TEMPERATURE)
# It is stored as an 8 bit 2's compliment signed number
if value >= 128:
value = value - 256
value = float(value)
temperature.append(value)
return temperature
@property
def hardware_version(self) -> list[int]:
"""Hardware Version.
We expect to see Version 16.
"""
version = []
for device_address in self._device_addresses:
value = self._read(device_address, self._HARDWARE_VERSION)
version.append(value)
return version
@property
def coprocessor_firmware_version(self) -> list[int]:
"""Copressor firmware version.
We expect to see Version 210.
"""
version = []
for device_address in self._device_addresses:
low_byte = self._read(device_address, self._CP_VER_LO)
high_byte = self._read(device_address, self._CP_VER_HI)
value = (high_byte << 8) | low_byte
version.append(value)
return version