Merge branch 'staging' into main

Author: sjpark608

.bumpversion.cfg

@@ -1,3 +1,3 @@
 [bumpversion]
 tag_name = rc/v{new_version}
-current_version = 1.2.6
+current_version = 1.2.8

setup.py

@@ -7,7 +7,7 @@
 
 setuptools.setup(
     name="edgepi-python-sdk",
-    version="1.2.6",
+    version="1.2.8",
     author="S.Park",
     author_email="spark@osensa.com",
     description="EdgePi Python SDK package",

src/edgepi/adc/edgepi_adc.py

@@ -212,10 +212,11 @@ def __read_register(self, start_addx: ADCReg, num_regs: int = 1):
         if num_regs < 1:
             raise ValueError("Number of registers to read must be at least 1")
         code = self.adc_ops.read_register_command(start_addx.value, num_regs)
-        out = self.transfer(code)
-        _logger.debug(f"__read_register: received {out}")
-        # first 2 entries are null bytes
-        return out[2:]
+        with self.spi_open():
+            out = self.transfer(code)
+            _logger.debug(f"__read_register: received {out}")
+            # first 2 entries are null bytes
+            return out[2:]
 
     def __write_register(self, start_addx: ADCReg, data: list[int]):
         """
@@ -231,7 +232,8 @@ def __write_register(self, start_addx: ADCReg, data: list[int]):
             raise ValueError("Number of registers to write to must be at least 1")
         code = self.adc_ops.write_register_command(start_addx.value, data)
         _logger.debug(f"__write_register: sending {code}")
-        out = self.transfer(code)
+        with self.spi_open():
+            out = self.transfer(code)
         return out
 
     def __set_rtd_pin(self, enable: bool = False):
@@ -273,12 +275,14 @@ def stop_conversions(self, adc_num: ADCNum):
         Halt voltage read conversions when ADC is set to perform continuous conversions
         """
         stop_cmd = self.adc_ops.stop_adc(adc_num=adc_num.value)
-        self.transfer(stop_cmd)
+        with self.spi_open():
+            self.transfer(stop_cmd)
 
     def __send_start_command(self, adc_num: ADCNum):
         """Triggers ADC conversion(s)"""
         start_cmd = self.adc_ops.start_adc(adc_num=adc_num.value)
-        self.transfer(start_cmd)
+        with self.spi_open():
+            self.transfer(start_cmd)
 
     def start_conversions(self, adc_num: ADCNum):
         """
@@ -321,7 +325,8 @@ def clear_reset_bit(self):
 
     def __read_data(self, adc: ADCNum, data_size: int):
         """Sends command to ADC to get new voltage conversion data"""
-        return self.transfer([adc.value.read_cmd] + [255] * data_size)
+        with self.spi_open():
+            return self.transfer([adc.value.read_cmd] + [255] * data_size)
 
     def __voltage_read(self, adc_num: ADCNum):
         """
@@ -554,19 +559,21 @@ def reset(self):
         Note this state differs from ADS1263 default power-on, due to
         application of custom power-on configurations required by EdgePi.
         """
-        self.transfer(self.adc_ops.reset_adc())
+        with self.spi_open():
+            self.transfer(self.adc_ops.reset_adc())
         self.__reapply_config()
 
     def __is_data_ready(self, adc_num: ADCNum):
         # pylint: disable=unused-private-member
         # required for integration testing in test_conversion_times.py
         """Utility for testing conversion times, returns True if ADC indicates new voltage data"""
-        read_data = self.transfer([adc_num.value.read_cmd] + [255] * 6)
-        if adc_num is ADCNum.ADC_1:
-            ready = (read_data[1] & 0b01000000) == 0b01000000
-        if adc_num is ADCNum.ADC_2:
-            ready = (read_data[1] & 0b10000000) == 0b10000000
-        return ready
+        with self.spi_open():
+            read_data = self.transfer([adc_num.value.read_cmd] + [255] * 6)
+            if adc_num is ADCNum.ADC_1:
+                ready = (read_data[1] & 0b01000000) == 0b01000000
+            if adc_num is ADCNum.ADC_2:
+                ready = (read_data[1] & 0b10000000) == 0b10000000
+            return ready
 
     def __read_registers_to_map(self):
         """

src/edgepi/dac/edgepi_dac.py

@@ -124,7 +124,8 @@ def write_voltage(self, analog_out: DACChannel, voltage: float):
         self.log.debug(f'Code: {code}')
 
         # update DAC register
-        self.transfer(self.dac_ops.generate_write_and_update_command(analog_out.value, code))
+        with self.spi_open():
+            self.transfer(self.dac_ops.generate_write_and_update_command(analog_out.value, code))
         # send voltage to analog out pin
         self.__send_to_gpio_pins(analog_out.value, voltage)
         return code
@@ -145,15 +146,17 @@ def set_power_mode(self, analog_out: DACChannel, power_mode: PowerMode):
         self.__dac_power_state[analog_out.value] = power_mode.value
         power_code = self.dac_ops.generate_power_code(self.__dac_power_state.values())
         cmd = self.dac_ops.combine_command(COM.COM_POWER_DOWN_OP.value, NULL_BITS, power_code)
-        self.transfer(cmd)
+        with self.spi_open():
+            self.transfer(cmd)
 
     def reset(self):
         """
         Performs a software reset of the EdgePi DAC to power-on default values,
         and stops all voltage transmissions through pins.
         """
         cmd = self.dac_ops.combine_command(COM.COM_SW_RESET.value, NULL_BITS, SW_RESET)
-        self.transfer(cmd)
+        with self.spi_open():
+            self.transfer(cmd)
 
     def channel_readback(self, analog_out: DACChannel) -> int:
         """
@@ -170,10 +173,11 @@ def channel_readback(self, analog_out: DACChannel) -> int:
         cmd = self.dac_ops.combine_command(
                 COM.COM_READBACK.value, DACChannel(analog_out.value).value, NULL_BITS
             )
-        self.transfer(cmd)
-        # all zero dummy command to trigger second transfer which
-        # contains the DAC register contents.
-        read_data = self.transfer([NULL_BITS, NULL_BITS, NULL_BITS])
+        with self.spi_open():
+            self.transfer(cmd)
+            # all zero dummy command to trigger second transfer which
+            # contains the DAC register contents.
+            read_data = self.transfer([NULL_BITS, NULL_BITS, NULL_BITS])
         self.log.debug(f"reading code {read_data}")
         return self.dac_ops.extract_read_data(read_data)
 
@@ -247,18 +251,19 @@ def __auto_code_handler(self, set_gain: bool):
         # gain have different steps of sending code value and toggling the GPIO pin.
 
         # gain being enabled, change code first than enable gain
-        if set_gain:
-            for ch, code in enumerate(codes):
-                # update DAC register
-                self.transfer(self.dac_ops.generate_write_and_update_command(ch, code))
-            self.gpio.set_pin_state(GainPin.DAC_GAIN.value)
-
-        # Disabling gain, change gain first than change codes
-        else:
-            self.gpio.clear_pin_state(GainPin.DAC_GAIN.value)
-            for ch, code in enumerate(codes):
-                # update DAC register
-                self.transfer(self.dac_ops.generate_write_and_update_command(ch, code))
+        with self.spi_open():
+            if set_gain:
+                for ch, code in enumerate(codes):
+                    # update DAC register
+                    self.transfer(self.dac_ops.generate_write_and_update_command(ch, code))
+                self.gpio.set_pin_state(GainPin.DAC_GAIN.value)
+
+            # Disabling gain, change gain first than change codes
+            else:
+                self.gpio.clear_pin_state(GainPin.DAC_GAIN.value)
+                for ch, code in enumerate(codes):
+                    # update DAC register
+                    self.transfer(self.dac_ops.generate_write_and_update_command(ch, code))
 
     def set_dac_gain(self, set_gain: bool, auto_code_change: bool = False):
         """

src/edgepi/eeprom/edgepi_eeprom.py

@@ -82,7 +82,8 @@ def __allocated_memory(self, offset):
         Return:
             length (int): size of memory to read
         '''
-        read_data = self.__sequential_read(offset, EEPROMInfo.PAGE_SIZE.value)
+        with self.i2c_open():
+            read_data = self.__sequential_read(offset, EEPROMInfo.PAGE_SIZE.value)
         check_crc(read_data[:-1], read_data[-1])
         return (read_data[0]<<8)| read_data[1]
 
@@ -124,10 +125,11 @@ def __write_edgepi_reserved_memory(self, pb_serial_list: bytes):
         pages = self.__generate_list_of_pages_crc(data)
 
         mem_offset = start_mem
-        for page in pages:
-            self.__page_write_register(mem_offset, page)
-            mem_offset = mem_offset+len(page)
-            time.sleep(PAGE_WRITE_CYCLE_TIME)
+        with self.i2c_open():
+            for page in pages:
+                self.__page_write_register(mem_offset, page)
+                mem_offset = mem_offset+len(page)
+                time.sleep(PAGE_WRITE_CYCLE_TIME)
 
     def __read_edgepi_reserved_memory(self):
         '''
@@ -149,11 +151,12 @@ def __read_edgepi_reserved_memory(self):
                     int((buff_and_len)/(page_size-CRC_BYTE_SIZE))+1
 
         mem_offset = EdgePiMemoryInfo.PRIVATE_SPACE_START_BYTE.value
-        for page in range(num_pages):
-            buff_list = self.__sequential_read(mem_offset+(page*page_size), page_size)
-            check_crc(buff_list[:-1], buff_list[-1])
-            buff+=buff_list[:-1]
-            time.sleep(0.01)
+        with self.i2c_open():
+            for page in range(num_pages):
+                buff_list = self.__sequential_read(mem_offset+(page*page_size), page_size)
+                check_crc(buff_list[:-1], buff_list[-1])
+                buff+=buff_list[:-1]
+                time.sleep(0.01)
         return bytes(buff[2:buff_and_len])
 
     def read_edgepi_data(self):
@@ -290,10 +293,11 @@ def read_user_space(self, mem_size: int = None):
             num_pages = int((buff_and_len)/(page_size-CRC_BYTE_SIZE))+1
 
         mem_offset = EdgePiMemoryInfo.USER_SPACE_START_BYTE.value
-        for page in range(num_pages):
-            buff_list = self.__sequential_read(mem_offset+(page*page_size), page_size)
-            check_crc(buff_list[:-1], buff_list[-1])
-            buff+=buff_list[:-1]
+        with self.i2c_open():
+            for page in range(num_pages):
+                buff_list = self.__sequential_read(mem_offset+(page*page_size), page_size)
+                check_crc(buff_list[:-1], buff_list[-1])
+                buff+=buff_list[:-1]
         return buff[2:buff_and_len]
 
     def write_user_space(self, data: bytes):
@@ -314,10 +318,11 @@ def write_user_space(self, data: bytes):
         pages = self.__generate_list_of_pages_crc(data)
 
         mem_offset = start_mem
-        for page in pages:
-            self.__page_write_register(mem_offset, page)
-            mem_offset = mem_offset+len(page)
-            time.sleep(PAGE_WRITE_CYCLE_TIME)
+        with self.i2c_open():
+            for page in pages:
+                self.__page_write_register(mem_offset, page)
+                mem_offset = mem_offset+len(page)
+                time.sleep(PAGE_WRITE_CYCLE_TIME)
 
 # TODO why not separate it into a class
     def init_memory(self):
@@ -381,10 +386,11 @@ def reset_user_space(self):
         reset_vals = [255] * page_size
 
         mem_offset = start_address_page
-        for _ in range(tatal_page):
-            self.__page_write_register(mem_offset, reset_vals)
-            mem_offset = mem_offset+page_size
-            time.sleep(PAGE_WRITE_CYCLE_TIME)
+        with self.i2c_open():
+            for _ in range(tatal_page):
+                self.__page_write_register(mem_offset, reset_vals)
+                mem_offset = mem_offset+page_size
+                time.sleep(PAGE_WRITE_CYCLE_TIME)
 
     def reset_edgepi_memory(self, bin_hash: str = None, bin_bytes: bytes = None):
         """

src/edgepi/gpio/edgepi_gpio_chip.py

@@ -39,11 +39,10 @@ def read_gpio_pin_state(self, pin_name: str = None):
         Returns:
             `bool`: True if state is high, False if state is low
         """
-        self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
+        with self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
                        pin_dir=self.gpiochip_pins_dict[pin_name].dir,
-                       pin_bias=self.gpiochip_pins_dict[pin_name].bias)
-        state = self.read_state()
-        self.close_gpio()
+                       pin_bias=self.gpiochip_pins_dict[pin_name].bias):
+            state = self.read_state()
         return state
 
     def write_gpio_pin_state(self, pin_name: str = None, state: bool = None):
@@ -55,12 +54,11 @@ def write_gpio_pin_state(self, pin_name: str = None, state: bool = None):
         Return:
             N/A
         """
-        self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
+        with self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
                        pin_dir=self.gpiochip_pins_dict[pin_name].dir,
-                       pin_bias=self.gpiochip_pins_dict[pin_name].bias)
-        self.write_state(state)
-        read_back = self.read_state()
-        self.close_gpio()
+                       pin_bias=self.gpiochip_pins_dict[pin_name].bias):
+            self.write_state(state)
+            read_back = self.read_state()
         return read_back
 
     def set_gpio_pin_dir(self, pin_name: str = None, direction: bool = None):
@@ -70,10 +68,10 @@ def set_gpio_pin_dir(self, pin_name: str = None, direction: bool = None):
             pin_name (str): name of the pin
             direction (bool): direction to write, True = Input, False = Output
         """
-        self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
+        with self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
                        pin_dir="in" if direction else "out",
-                       pin_bias=self.gpiochip_pins_dict[pin_name].bias)
-        self.close_gpio()
+                       pin_bias=self.gpiochip_pins_dict[pin_name].bias):
+            return self.gpio.direction
 
     def toggle_gpio_pin_state(self, pin_name: str = None):
         """
@@ -83,12 +81,9 @@ def toggle_gpio_pin_state(self, pin_name: str = None):
         Return:
             N/A
         """
-        self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
+        with self.open_gpio(pin_num=self.__pin_name_dict[pin_name],
                        pin_dir=self.gpiochip_pins_dict[pin_name].dir,
-                       pin_bias=self.gpiochip_pins_dict[pin_name].bias)
-        state = self.read_state()
-        if state:
-            self.write_state(False)
-        else:
-            self.write_state(True)
-        self.close_gpio()
+                       pin_bias=self.gpiochip_pins_dict[pin_name].bias):
+            state = self.read_state()
+            self.write_state(not state)
+            

src/edgepi/gpio/edgepi_gpio_expander.py

@@ -32,12 +32,13 @@ def __read_register(self, reg_address: int, dev_address: int) -> int:
         Returns:
             `int`: 8-bit uint value of port/register
         '''
-        msg_read = self.set_read_msg(reg_address, [0xFF])
-        _logger.debug(f"Read Message: Register Address {msg_read[0].data}"
-                      f", Msg Place Holder {msg_read[1].data}")
-        self.transfer(dev_address, msg_read)
-        _logger.debug(f"Message Read: Register Address {msg_read[0].data},"
-                      f"Msg Place Holder {msg_read[1].data}")
+        with self.i2c_open():
+            msg_read = self.set_read_msg(reg_address, [0xFF])
+            _logger.debug(f"Read Message: Register Address {msg_read[0].data}"
+                          f", Msg Place Holder {msg_read[1].data}")
+            self.transfer(dev_address, msg_read)
+            _logger.debug(f"Message Read: Register Address {msg_read[0].data},"
+                          f"Msg Place Holder {msg_read[1].data}")
         return msg_read[1].data[0]
 
     def __write_changed_values(self, reg_dict: dict, dev_address: int):
@@ -50,11 +51,12 @@ def __write_changed_values(self, reg_dict: dict, dev_address: int):
         Returns:
             void
         '''
-        for reg_addx, entry in reg_dict.items():
-            if entry['is_changed']:
-                msg_write = self.set_write_msg(reg_addx, [entry['value']])
-                _logger.debug(f'Write Message Content {msg_write[0]}')
-                self.transfer(dev_address, msg_write)
+        with self.i2c_open():
+            for reg_addx, entry in reg_dict.items():
+                if entry['is_changed']:
+                    msg_write = self.set_write_msg(reg_addx, [entry['value']])
+                    _logger.debug(f'Write Message Content {msg_write[0]}')
+                    self.transfer(dev_address, msg_write)
 
     def read_expander_pin(self, pin_name: str) -> bool:
         '''