Refactor for Python 3.8+ compatibility

.github/workflows/example-shm-ci.yml

@@ -9,6 +9,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, windows-latest, macos-latest]
+        python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
 
     steps:
       # ───── 1. Checkout ─────
@@ -17,7 +18,7 @@ jobs:
       # ───── 2. Python + Poetry ─────
       - uses: actions/setup-python@v5
         with:
-          python-version: "3.12"
+          python-version: ${{ matrix.python-version }}
 
       - name: Install Poetry
         env:
@@ -46,11 +47,11 @@ jobs:
 
       # ───── 5. Tests ─────
       - name: Run pytest (with coverage)
-        if: matrix.os == 'ubuntu-latest'
+        if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12'
         run: poetry run pytest --cov=src --cov-report=xml --cov-report=term-missing
 
       - name: Upload coverage to Codecov
-        if: matrix.os == 'ubuntu-latest'
+        if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12'
         uses: codecov/codecov-action@v5
         with:
           files: coverage.xml
@@ -59,7 +60,7 @@ jobs:
 
       # Windows/macOS → test without coverage
       - name: Run pytest (no coverage)
-        if: matrix.os != 'ubuntu-latest'
+        if: matrix.os != 'ubuntu-latest' || matrix.python-version != '3.12'
         run: poetry run pytest --no-cov -q
 
       # ───── 6. Build + dist artefacts ─────
@@ -72,7 +73,7 @@ jobs:
           poetry build --format sdist
 
       - name: Upload dist artefacts
-        if: matrix.os == 'ubuntu-latest'
+        if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12'
         uses: actions/upload-artifact@v4
         with:
           name: example-shm-dist

README.md

@@ -4,6 +4,11 @@ This project creates a fully-functional demo of CP-SENS project.
 The demo is to be run inside the
 [DTaaS platform](https://github.com/into-cps-association/DTaaS).
 
+## Requirements
+
+- Python 3.8 - 3.12 (Python 3.8+ is supported)
+- Poetry for dependency management
+
 ## Development Setup
 
 This is a [poetry-based project](https://python-poetry.org/docs/).

pyproject.toml

@@ -12,19 +12,19 @@ license = "INTO-CPS Association"
 packages = [{include = "*", from="src"}]
 
 [tool.poetry.dependencies]
-python = ">=3.12, <3.13"
+python = ">=3.8, <3.13"
 paho-mqtt = "^2.1.0"
-numpy = "^2.2.5"
+numpy = "^1.24.0"
 click ="^8.1.8"
 pyOMA-2 = "1.0.0"
 yafem = "1.0.0"
 
 [tool.poetry.group.dev.dependencies]
-pylint = "^3.3.6"
-pytest = "^8.3.5"
-pytest-cov = "^6.1.1"
-pytest-mock = "^3.14.0"
-pytest-env = "^1.1.5"
+pylint = "^3.0.0"
+pytest = "^7.0.0"
+pytest-cov = "^4.0.0"
+pytest-mock = "^3.10.0"
+pytest-env = "^0.8.0"
 
 [build-system]
 requires = ["poetry-core"]

pytest.ini

@@ -1,5 +1,5 @@
 [pytest]
-minversion = 8.3
+minversion = 7.0
 pythonpath = src src/data src/methods  tests
 testpaths =
     tests

src/data/accel/hbk/accelerometer.py

@@ -154,7 +154,7 @@ def clear_used_data(self, start_key: int, samples_to_remove: int) -> None:
                 remaining_to_remove -= num_to_remove
 
 
-    def read(self, requested_samples: int) -> Tuple[(int, np.ndarray)]:
+    def read(self, requested_samples: int) -> Tuple[int, np.ndarray]:
         """
         Reads the oldest accelerometer data from the FIFO buffer and removes only the read samples.
 

src/methods/mode_clustering.py

@@ -57,7 +57,7 @@ def _on_message(_client: mqtt.Client, _userdata: dict, msg: mqtt.MQTTMessage) ->
         print(f"Error processing sysid message: {e}")
 
 
-def cluster_sysid(sysid_output: Any, params: dict[str,Any]) -> Tuple[dict[str,Any], np.ndarray]:
+def cluster_sysid(sysid_output: Any, params: Dict[str, Any]) -> Tuple[Dict[str, Any], np.ndarray]:
     """
     Runs the mode clustering algorithm.
 

src/methods/mode_clustering_functions/align_clusters.py

@@ -1,8 +1,8 @@
 import numpy as np
-from typing import Any
+from typing import Any, Dict
 from functions.calculate_mac import calculate_mac
 
-def alignment(cluster_dict: dict[str,dict], Params: dict[str,Any]) -> dict[str,dict]:
+def alignment(cluster_dict: Dict[str, dict], Params: Dict[str, Any]) -> Dict[str, dict]:
     """
         Alignment/merging of clusters
 
@@ -71,7 +71,7 @@ def alignment(cluster_dict: dict[str,dict], Params: dict[str,Any]) -> dict[str,d
     cluster_dict_alligned = cluster_dict
     return cluster_dict_alligned
 
-def join_clusters(cluster_1: dict[str,Any], cluster_2: dict[str,Any], Params: dict[str,Any]) -> dict[str,Any]:
+def join_clusters(cluster_1: Dict[str, Any], cluster_2: Dict[str, Any], Params: Dict[str, Any]) -> Dict[str, Any]:
     """
         Add two clusters together
 
@@ -126,7 +126,7 @@ def join_clusters(cluster_1: dict[str,Any], cluster_2: dict[str,Any], Params: di
 
     return cluster, cluster_remaining
 
-def append_cluster_data(cluster: dict[str,Any], cluster2: dict[str,Any], id: int) -> dict[str,Any]:
+def append_cluster_data(cluster: Dict[str, Any], cluster2: Dict[str, Any], id: int) -> Dict[str, Any]:
     """
         Add cluster data to an existing cluster
 

src/methods/mode_clustering_functions/clustering.py

@@ -1,4 +1,4 @@
-from typing import Any
+from typing import Any, Dict, Tuple
 import numpy as np
 from functions.clean_sysid_output import (remove_highly_uncertain_points,transform_sysid_features)
 from methods.mode_clustering_functions.create_cluster import cluster_creation
@@ -10,7 +10,7 @@
 # Following the algorithm proposed here: https://doi.org/10.1007/978-3-031-61421-7_56
 # JVM 22/10/2025
 
-def cluster_func(sysid_output: dict[str,Any], Params : dict[str,Any]) -> tuple[dict[str,Any], dict[str,Any], dict[str,Any]]:
+def cluster_func(sysid_output: Dict[str, Any], Params: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any]]:
     """
         Clustering of OMA results
 
@@ -138,7 +138,7 @@ def cluster_func(sysid_output: dict[str,Any], Params : dict[str,Any]) -> tuple[d
 
     return cluster_dict4
 
-def remove_data_from_S(data: dict[str,Any],cluster: dict[str,Any]) -> dict[str,Any]:
+def remove_data_from_S(data: Dict[str, Any], cluster: Dict[str, Any]) -> Dict[str, Any]:
     """
         Remove cluster from data or S
 
@@ -178,7 +178,7 @@ def remove_data_from_S(data: dict[str,Any],cluster: dict[str,Any]) -> dict[str,A
 
     return data2
 
-def sort_cluster(cluster: dict[str,Any]) -> dict[str,Any]:
+def sort_cluster(cluster: Dict[str, Any]) -> Dict[str, Any]:
     """
         Sort cluster based on row/model order
 

src/methods/mode_clustering_functions/expand_cluster.py

@@ -1,8 +1,8 @@
-from typing import Any
+from typing import Any, Dict
 import numpy as np
 from methods.mode_clustering_functions.create_cluster import cluster_creation
 
-def cluster_expansion(cluster: dict[str,Any], data: dict[str,Any], Params: dict[str,Any]) -> dict[str,Any]:
+def cluster_expansion(cluster: Dict[str, Any], data: Dict[str, Any], Params: Dict[str, Any]) -> Dict[str, Any]:
     """
         Expand cluster based on minima and maxima bound
 

src/methods/mode_clustering_functions/initialize_Ip.py

@@ -1,7 +1,7 @@
-from typing import Any
+from typing import Any, List, Dict
 import numpy as np
 
-def cluster_initial(ip: list[float], data: dict[str,Any], bound: float = 2) -> dict[str,Any]:
+def cluster_initial(ip: List[float], data: Dict[str, Any], bound: float = 2) -> Dict[str, Any]:
     """
         Find the initial cluster points
 

src/methods/mode_tracking.py

@@ -9,8 +9,8 @@
 from functions.plot_clusters import plot_clusters
 # pylint: disable=C0103, W0603
 
-def track_clusters(cluster_dict: dict[str,Any], tracked_clusters: dict[str,Any],
-                      params: dict[str,Any]) -> dict[str,Any]:
+def track_clusters(cluster_dict: Dict[str, Any], tracked_clusters: Dict[str, Any],
+                      params: Dict[str, Any]) -> Dict[str, Any]:
     """
     Runs the mode tracking algorithm.
 

src/methods/mode_tracking_functions/match_to_tracked_cluster.py

@@ -1,8 +1,8 @@
-from typing import Any
+from typing import Any, Dict, List
 import numpy as np
 from functions.calculate_mac import calculate_mac
 
-def match_cluster_to_tracked_cluster(cluster_dict: dict[str,Any], tracked_clusters: dict[str,Any], Params: dict[str,Any], result_prev: dict[str,Any] = {},skip_cluster: list = [], skip_tracked_cluster: list = []) -> dict[str,Any]:
+def match_cluster_to_tracked_cluster(cluster_dict: Dict[str, Any], tracked_clusters: Dict[str, Any], Params: Dict[str, Any], result_prev: Dict[str, Any] = None, skip_cluster: List = None, skip_tracked_cluster: List = None) -> Dict[str, Any]:
     """
     Match clusters to tracked clusters
 
@@ -25,6 +25,13 @@ def match_cluster_to_tracked_cluster(cluster_dict: dict[str,Any], tracked_cluste
         result (dict): Dictionary of matches
 
     """
+    if result_prev is None:
+        result_prev = {}
+    if skip_cluster is None:
+        skip_cluster = []
+    if skip_tracked_cluster is None:
+        skip_tracked_cluster = []
+
     result = {}
     for id, key in enumerate(cluster_dict): #Go through all clusters
         if id in skip_cluster: #If this cluster is already matched skip it

src/methods/mode_tracking_functions/mode_tracking.py

@@ -1,10 +1,10 @@
-from typing import Any
+from typing import Any, Dict
 import numpy as np
 from methods.mode_tracking_functions.match_to_tracked_cluster import match_cluster_to_tracked_cluster
 from methods.mode_tracking_functions.resolve_nonunique_matches import resolve_nonunique_matches
 # JVM 22/10/2025
 
-def cluster_tracking(cluster_dict: dict[str,Any],tracked_clusters: dict[str,Any],Params: dict[str,Any]=None) -> dict[str,Any]:
+def cluster_tracking(cluster_dict: Dict[str, Any], tracked_clusters: Dict[str, Any], Params: Dict[str, Any] = None) -> Dict[str, Any]:
     """
     Tracking of modes across experiments
 

src/methods/packages/pyoma/ssiWrapper.py

@@ -202,7 +202,7 @@ def mpe(
 
     def mpe_from_plot(
         self,
-        freqlim: typing.Optional[tuple[float, float]] = None,
+        freqlim: typing.Optional[typing.Tuple[float, float]] = None,
         rtol: float = 1e-2,
     ) -> typing.Any:
         """
@@ -265,7 +265,7 @@ def mpe_from_plot(
 
     def plot_stab(
         self,
-        freqlim: typing.Optional[tuple[float, float]] = None,
+        freqlim: typing.Optional[typing.Tuple[float, float]] = None,
         hide_poles: typing.Optional[bool] = True,
     ) -> typing.Any:
         """
@@ -305,7 +305,7 @@ def plot_stab(
 
     def plot_cluster(
         self,
-        freqlim: typing.Optional[tuple[float, float]] = None,
+        freqlim: typing.Optional[typing.Tuple[float, float]] = None,
         hide_poles: typing.Optional[bool] = True,
     ) -> typing.Any:
         """

src/pt_mock/find_offset.py

@@ -1,6 +1,6 @@
 import time
 import json
-from typing import Any, Dict
+from typing import Any, Dict, List, Tuple
 import busio  # pylint: disable=import-error
 import board  # pylint: disable=import-error
 import adafruit_adxl37x  # pylint: disable=import-error
@@ -53,10 +53,10 @@ def calibrate_sensor(sensor: Any, sensor_label: str,
           f"for {duration} seconds...")
     sensor.range = SENSOR_RANGE
     start_time: float = time.time()
-    samples: list[float] = []
+    samples: List[float] = []
 
     while time.time() - start_time < duration:
-        reading: tuple[float, float, float] = sensor.acceleration
+        reading: Tuple[float, float, float] = sensor.acceleration
         samples.append(reading[0])  # x-axis
 
     avg_x: float = sum(samples) / len(samples)