Started working on new BMS system

sensors/bms/bms/freya_bms.py

@@ -0,0 +1,174 @@
+import subprocess
+
+# TODO: Fix so that it automatically detects which USB port to use
+# TODO: Check that data is received on initialization (find out if batteries are 
+# connected)
+
+class BMS:
+    """ Class containing Freya's BMS system
+
+    Attributes:
+    -----------
+        voltage (float): Voltage being delivered by the BMS
+        current (float): Current being delivered by the BMS
+        design_capacity (float): Capacity of the BMS
+        remaining_capacity (float): Remaining capacity of the BMS
+        percent_capacity (float): Remaining capacity as a float from 0-1
+        cycle_count (int): idk
+        probes (int): idk
+        strings (int): idk
+        temps (Tuple[float, float, float]): Temperatures of the cell arrays
+        cells (Tuple[float, ...]): Voltages of individual cells (6 elements long)
+        balance (str): idk
+        cell_total (float): Sum of cell voltages
+        cell_min (float): Smallest cell voltage
+        cell_max (float): Largest cell voltage
+        cell_diff (float): Difference between largest and smallest cell voltage
+        cell_avg (float): Average of cell voltages
+        device_name (str): Device name
+        manufacture_date (str): Date of manufacturing
+        version (str): Version
+        FET (str): idk
+
+    Methods:
+    --------
+        get_bms_data() -> str | None:
+            returns pure BMS data string, or None if exception is thrown 
+        change_usb_port(usb_port: str) -> None:
+            changes the usb port for the BMS
+    """
+
+    def __init__(self, usb_port: str) -> None:
+        """
+            Parameters:
+                usb_port (str): USB port to connect to, either ttyUSB0 or ttyUSB1
+
+            Returns:
+                None
+
+            Note: Private members are denoted by _variable_name            
+        """
+        self.usb_port = usb_port
+
+        self.command = ["jbdtool", "-t", f"serial:/dev/{usb_port}"]
+        self._voltage = 0
+        self._current = 0
+        self._design_capacity = 0
+        self._remaining_capacity = 0
+        self._percent_capacity = 0
+        self._cycle_count = 0
+        self._probes = 0
+        self._strings = 0
+        self._temps = 0
+        self._cells = 0
+        self._balance = ""
+        self._cell_total = 0
+        self._cell_min = 0
+        self._cell_max = 0
+        self._cell_avg = 0
+        self._device_name = ""
+        self._manufacture_date = ""
+        self._version = ""
+        self._FET = ""
+
+    def get_bms_data(self) -> str | None:
+        """
+            Function for getting data from the BMS
+
+            Parameters: 
+
+            Returns: 
+                if the jbdtool call works, it returns the BMS data as a string, 
+                otherwise it prints the error and returns None
+
+            Example output:
+                Voltage                   22.050
+                Current                   -2.000
+                DesignCapacity            62.000
+                RemainingCapacity         8.560
+                PercentCapacity           14
+                CycleCount                1
+                Probes                    3
+                Strings                   6
+                Temps                     24.9,23.3,23.2
+                Cells                     3.648,3.681,3.682,3.678,3.680,3.680
+                Balance                   000000
+                CellTotal                 22.049
+                CellMin                   3.648
+                CellMax                   3.682
+                CellDiff                  0.034
+                CellAvg                   3.675
+                DeviceName                JBD-AP21S001-L21S-200A-B
+                ManufactureDate           20221206
+                Version                   6.8
+                FET                       Charge,Discharge
+        """
+
+        try: 
+            response = subprocess.run(self.command,
+                                  stdout=subprocess.PIPE,
+                                  stderr=subprocess.PIPE,
+                                  check=True)
+
+            return response
+        except subprocess.CalledProcessError as e:
+            print("An error occured when getting BMS data")
+            print(f"Error: {e.stderr.decode()}")
+            print("Please check that USBs are connected to Raspberry PI")
+            return None
+
+    def parse_bms_data(self, bms_data: subprocess.CompletedProcess) -> None:
+        """
+            Parses BMS data and updates class members accordingly
+
+            Parameters:
+                bms_data (subprocess.CompletedProcess): object containing result 
+                of the jbdtool command
+
+            Returns: None
+        """
+
+        data = bms_data.stdout.decode().split("\n")
+
+        for element in data:
+
+            element = element.split()
+            print(element)
+
+        self._voltage = float(data[0])
+        self._current = float(data[1])
+        self._design_capacity = float(data[2])
+        self._remaining_capacity = float(data[3])
+        self._percent_capacity = float(data[4]) / 100
+        self._cycle_count = int(data[5])
+        self._probes = int(data[6])
+        self._strings = int(data[7])
+        self._temps = int(data[8].split(","))
+        self._cells = int(data[9].split(","))
+        self._balance = data[10]
+        self._cell_total = float(data[11])
+        self._cell_min = float(data[12])
+        self._cell_max = float(data[13])
+        self._cell_avg = float(data[14])
+        self._device_name = data[15]
+        self._manufacture_date = data[16]
+        self._version = data[17]
+        self._FET = data[18]
+
+
+    def change_usb_port(self, usb_port: str) -> None:
+        """
+            Changes the usb port.
+
+            Parameters: 
+                usb_port (str): The name of the port to change to
+
+            Returns: 
+                None
+        """
+
+        self.usb_port = usb_port
+        self.command = ["jbdtool", "-t", f"serial:/dev/{usb_port}"]
+
+test = BMS
+test.change_usb_port

sensors/bms/bms/old_freya_bms.py

@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+import subprocess
+
+
+def execute_jbdtool(usb_port):
+    """
+    This function executes the jbdtool command to query the BMS (Battery Management System)
+    connected to the specified serial port. It then parses the output to extract 
+    information about Voltage, Current, Temps, and Cells.
+
+    :param usb_port: The USB port number to connect to (e.g., "USB0", "USB1", etc.)
+    :type usb_port: str
+
+    Output:
+        - Prints the Voltage, Current, Temps, and Cells to the console.
+        - If there's an error in executing the command, prints an error message.
+
+    Example Output:
+        Voltage: 24.010
+        Current: 0.000
+        Temps: 23.1,23.1,22.9
+        Cells: 4.006,4.005,4.000,4.005,4.000,4.000
+    """
+
+    command = ["./jbdtool", "-t", f"serial:/dev/{usb_port}"]
+    working_directory = "/home/vortex/bms/jbdtool"
+
+    try:
+        response = subprocess.run(command,
+                                  stdout=subprocess.PIPE,
+                                  stderr=subprocess.PIPE,
+                                  check=True,
+                                  cwd=working_directory)
+        return response
+
+    except subprocess.CalledProcessError as e:
+        print("An error occurred while executing the command.")
+        print("Error:", e.stderr.decode())
+
+
+def parse_bms_data(response):
+    output = response.stdout.decode()
+
+    voltage = output.split("Voltage")[1].split("\n")[0].strip()
+    current = output.split("Current")[1].split("\n")[0].strip()
+    temps = output.split("Temps")[1].split("\n")[0].strip()
+    cells = output.split("Cells")[1].split("\n")[0].strip()
+
+    return voltage, current, temps, cells

sensors/bms/bms/old_freya_bms_node.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python3
+
+import rospy
+from sensor_msgs.msg import BatteryState
+from freya_bms import execute_jbdtool, parse_bms_data
+
+
+def main():
+    # Initialize the node with the name 'freya_bms'
+    rospy.init_node('freya_bms')
+
+    # Create a Publisher object to publish messages to the 'battery_info' topic
+    battery_pub = rospy.Publisher('/internal/status/bms',
+                                  BatteryState,
+                                  queue_size=10)
+
+    # List of USB devices to loop through
+    usb_ports = ["ttyUSB1", "ttyUSB2"]  # Add as many as you need
+
+    # Rate object to control the loop rate (in Hz)
+    rate = rospy.Rate(1)  # 1 Hz
+
+    while not rospy.is_shutdown():
+        for usb_port in usb_ports:
+            response = execute_jbdtool(usb_port)
+            if response.stdout != b'':  # Check if data is received
+
+                voltage, current, temps, cells = parse_bms_data(response)
+
+                battery_msg = BatteryState()
+                battery_msg.voltage = float(
+                    voltage)  # Assuming voltage is in Volts
+                battery_msg.current = float(
+                    current)  # Assuming current is in Amperes
+                battery_msg.temperature = float(temps.split(
+                    ',')[0])  # Assuming temperature is in degrees Celsius
+                battery_msg.location = usb_port
+
+                # Publish the Battery message to the 'battery_info' topic
+                battery_pub.publish(battery_msg)
+
+        # Sleep until the next iteration
+        rate.sleep()
+
+
+if __name__ == '__main__':
+    try:
+        main()
+        rospy.spin()
+    except rospy.ROSInterruptException:
+        pass

sensors/bms/bms/testfile.py

@@ -0,0 +1,36 @@
+import freya_bms
+# import usb
+
+import re
+import subprocess
+
+device_re = re.compile(b"Bus\s+(?P<bus>\d+)\s+Device\s+(?P<device>\d+).+ID\s(?P<id>\w+:\w+)\s(?P<tag>.+)$", re.I)
+df = subprocess.check_output("lsusb")
+devices = []
+for i in df.split(b'\n'):
+    if i:
+        info = device_re.match(i)
+        if info:
+            dinfo = info.groupdict()
+            dinfo['device'] = '/dev/bus/usb/%s/%s' % (dinfo.pop('bus'), dinfo.pop('device'))
+            devices.append(dinfo)
+
+for device in devices:
+    print(device)
+
+# print(devices)
+
+test = freya_bms.BMS("ttyUSB0")
+test.parse_bms_data(test.get_bms_data())
+
+print(test._cells)
+
+string = "hei                på"
+print(string.split())
+
+
+
+
+# [17908.264303] usb 1-1.1: USB disconnect, device number 6
+# [17908.265921] ftdi_sio ttyUSB0: FTDI USB Serial Device converter now disconnected from ttyUSB0
+# [17908.266006] ftdi_sio 1-1.1:1.0: device disconnected

sensors/bms/package.xml

@@ -0,0 +1,18 @@
+<?xml version="1.0"?>
+<?xml-model href="http://download.ros.org/schema/package_format3.xsd" schematypens="http://www.w3.org/2001/XMLSchema"?>
+<package format="3">
+  <name>bms</name>
+  <version>0.0.0</version>
+  <description>TODO: Package description</description>
+  <maintainer email="[email protected]">vortex</maintainer>
+  <license>TODO: License declaration</license>
+
+  <test_depend>ament_copyright</test_depend>
+  <test_depend>ament_flake8</test_depend>
+  <test_depend>ament_pep257</test_depend>
+  <test_depend>python3-pytest</test_depend>
+
+  <export>
+    <build_type>ament_python</build_type>
+  </export>
+</package>

sensors/bms/setup.cfg

@@ -0,0 +1,4 @@
+[develop]
+script_dir=$base/lib/bms
+[install]
+install_scripts=$base/lib/bms

sensors/bms/setup.py

@@ -0,0 +1,25 @@
+from setuptools import find_packages, setup
+
+package_name = 'bms'
+
+setup(
+    name=package_name,
+    version='0.0.0',
+    packages=find_packages(exclude=['test']),
+    data_files=[
+        ('share/ament_index/resource_index/packages',
+            ['resource/' + package_name]),
+        ('share/' + package_name, ['package.xml']),
+    ],
+    install_requires=['setuptools'],
+    zip_safe=True,
+    maintainer='vortex',
+    maintainer_email='[email protected]',
+    description='TODO: Package description',
+    license='TODO: License declaration',
+    tests_require=['pytest'],
+    entry_points={
+        'console_scripts': [
+        ],
+    },
+)

sensors/bms/test/test_copyright.py

@@ -0,0 +1,25 @@
+# Copyright 2015 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from ament_copyright.main import main
+import pytest
+
+
+# Remove the `skip` decorator once the source file(s) have a copyright header
+@pytest.mark.skip(reason='No copyright header has been placed in the generated source file.')
+@pytest.mark.copyright
+@pytest.mark.linter
+def test_copyright():
+    rc = main(argv=['.', 'test'])
+    assert rc == 0, 'Found errors'