release/1.1.5 (#13)

## [1.1.5] - 2025-09-22
### Fixed
- 🩹 More durable timestamp extrapolation in time data insertion
- 🩹 Using shutil.move instead of pathlib.Path.rename to allow for differing file systems 
- 🩹 Broken path in docs 

### Changed
- ⚡ Added additional safeguards in TWST Collection to prevent zombie processes   
- ⚡ How random data uses the load/send when used as a class method for more durability

### Added
- 🔥 Better probe identification in logs
- 🔥 Filter on files being ingested by ADVA probes to only attempt files which match expected naming convention from input directory

### Removed
- 🗡️ static version of openSAMPL in docs instructions

---------

Co-authored-by: MacFarland, Midgie <macfarlandmj@ornl.gov>

Author: midgemacf

CHANGELOG.md

@@ -37,6 +37,23 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 *Unreleased* versions radiate potential—-and dread. Once you merge an infernal PR, move its bullet under a new version heading with the actual release date.*
 
 -->
+## [1.1.5] - 2025-09-22
+### Fixed
+- 🩹 More durable timestamp extrapolation in time data insertion
+- 🩹 Using shutil.move instead of pathlib.Path.rename to allow for differing file systems 
+- 🩹 Broken path in docs 
+
+### Changed
+- ⚡ Added additional safeguards in TWST Collection to prevent zombie processes   
+- ⚡ How random data uses the load/send when used as a class method for more durability
+
+### Added
+- 🔥 Better probe identification in logs
+- 🔥 Filter on files being ingested by ADVA probes to only attempt files which match expected naming convention from input directory
+
+### Removed
+- 🗡️ static version of openSAMPL in docs instructions
+
 ## [1.1.4] - 2025-08-22
 ### Added
 - 🔥 Random data generation for all supported probes

docs/getting-started/installation.md

@@ -5,7 +5,7 @@
 1. Ensure you have Python 3.9 or higher installed
 2. Pip install the latest version of openSAMPL: 
 ```bash
-pip install opensampl==0.2.0
+pip install opensampl
 ```
 
 ## Installation for Developers

docs/getting-started/quickstart.md

@@ -13,4 +13,4 @@ This creates:
 In order to ingest data, you simply have to run `opensampl load` with appropriate arguments, and it will go straight into your new TimescaleDB
 instance and become visible on Grafana. 
 
-See the [Configuration](../guides/configuration#opensampl-server) page on configuring your server instance. 
+See the [Configuration](../guides/configuration.md#opensampl-server) page on configuring your server instance. 

docs/guides/opensampl-server.md

@@ -16,9 +16,9 @@ The openSAMPL Server CLI provides commands for managing a Docker Compose deploym
 - Docker and Docker Compose installed
 - The openSAMPL Python package installed
 
-Install with the server extra: 
+Install openSAMPL as normal: 
 ```
-pip install "opensampl==0.2.0[server]"
+pip install opensampl
 ```
 
 ## Basic Commands
@@ -131,6 +131,17 @@ opensampl-server up --env-file ./my-custom-env.env
   - `BACKEND_URL=http://localhost:8015`
   - `ROUTE_TO_BACKEND=true`
 
+## Power users 
+
+For those who are more familiar with docker, there is a `opensampl-server2` which corresponds to the following, more directly 
+exposing the docker to users.
+`OPENSAMPL_SERVER__COMPOSE_FILE` is set in your .env file or environment.
+
+```bash
+opensampl-server2 --env-file ENV_FILE args
+docker compose --env-file ${ENV_FILE} -f ${OPENSAMPL_SERVER__COMPOSE_FILE} $@
+```
+
 ## Troubleshooting
 
 If you encounter issues:
@@ -172,3 +183,5 @@ nano ./dev.env
 # Start the server with custom environment
 opensampl-server up --env-file ./dev.env
 ```
+
+

opensampl/collect/microchip/twst/readings.py

@@ -8,8 +8,12 @@
 import asyncio
 from typing import Optional
 
+from loguru import logger
+
 from opensampl.collect.modem import ModemReader, require_conn
 
+SENTINEL = b"--openSAMPL stop reading--"  # type: ignore[assignment]
+
 
 class ModemStatusReader(ModemReader):
     """
@@ -34,6 +38,7 @@ def __init__(self, host: str, duration: int = 60, keys: Optional[list[str]] = No
         self.keys = keys
         self.queue = asyncio.Queue()
         self.readings = []
+        self.continue_reading = False
         super().__init__(host=host, port=port)
 
     @require_conn
@@ -43,10 +48,16 @@ async def reader_task(self):
 
         Reads lines from the telnet connection and queues them for processing.
         """
-        while True:
-            line = await self.reader.readline()
-            if line:
+        try:
+            while self.continue_reading:
+                line = await asyncio.wait_for(self.reader.readline(), timeout=5.0)
+                if not line:
+                    break  # EOF
                 await self.queue.put(line)
+        except asyncio.TimeoutError:
+            logger.debug(f"Timeout waiting for data from {self.host}:{self.port}")
+        finally:
+            await self.queue.put(SENTINEL)
 
     def parse_line(self, line: str):
         """
@@ -90,11 +101,16 @@ async def processor_task(self):
         """
         while True:
             line = await self.queue.get()
-            parsed = self.parse_line(line)
-            if parsed:
-                timestamp, definition, value = parsed
-                if self.should_keep(definition):
-                    self.readings.append(parsed)
+            try:
+                if line == SENTINEL:
+                    break
+                parsed = self.parse_line(line)
+                if parsed:
+                    timestamp, definition, value = parsed
+                    if self.should_keep(definition):
+                        self.readings.append(parsed)
+            finally:
+                self.queue.task_done()
 
     async def collect_readings(self):
         """
@@ -104,11 +120,15 @@ async def collect_readings(self):
         duration, then cancels the tasks.
         """
         async with self.connect():
+            self.continue_reading = True
             read_coroutine = asyncio.create_task(self.reader_task())
             process_coroutine = asyncio.create_task(self.processor_task())
 
             try:
                 await asyncio.sleep(self.duration)
+                self.continue_reading = False
+                await self.queue.join()
+
             finally:
                 # Cancel tasks and wait for them to complete
                 read_coroutine.cancel()

opensampl/load_data.py

@@ -113,6 +113,7 @@ def load_time_data(
     if not isinstance(session, Session):
         raise TypeError("Session must be a SQLAlchemy session")
 
+    probe_readable = str(probe_key)
     try:
         from opensampl.load.data import DataFactory
 
@@ -124,6 +125,10 @@ def load_time_data(
             strict=strict,
             session=session,
         )
+        probe_readable = (
+            data_definition.probe.name  # ty: ignore[possibly-unbound-attribute]
+            or f"{data_definition.probe.ip_address} ({data_definition.probe.probe_id})"  # ty: ignore[possibly-unbound-attribute]
+        )
 
         if any(x is None for x in [data_definition.probe, data_definition.metric, data_definition.reference]):
             raise RuntimeError(f"Not all required definition fields filled: {data_definition.dump_factory()}")  # noqa: TRY301
@@ -134,9 +139,8 @@ def load_time_data(
         df["metric_type_uuid"] = data_definition.metric.uuid  # ty: ignore[possibly-unbound-attribute]
         logger.debug(df.head())
         # Ensure correct dtypes
-        df["time"] = pd.to_datetime(df["time"], utc=True, errors="raise")
+        df["time"] = pd.to_datetime(df["time"], format="mixed", utc=True, errors="raise")
         df["value"] = df["value"].apply(json.dumps)
-
         records = df.to_dict(orient="records")
         insert_stmt = text(f"""
         INSERT INTO {ProbeData.__table__.schema}.{ProbeData.__tablename__}
@@ -152,16 +156,20 @@ def load_time_data(
             total_rows = len(records)
             inserted = result.rowcount  # ty: ignore[unresolved-attribute]
             excluded = total_rows - inserted
-            logger.warning(f"Inserted {inserted}/{total_rows} rows; {excluded}/{total_rows} rejected due to conflicts")
+
+            logger.warning(
+                f"Inserted {inserted}/{total_rows} rows for {probe_readable}; "
+                f"{excluded}/{total_rows} rejected due to conflicts"
+            )
 
         except Exception as e:
             # In case of an error, roll back the session
             session.rollback()
-            logger.error(f"Error inserting rows: {e}")
+            logger.error(f"Error inserting rows for {probe_readable}: {e}")
             raise
 
     except Exception as e:
-        logger.exception(f"Error writing time data: {e}")
+        logger.exception(f"Error writing time data for {probe_readable}: {e}")
         session.rollback()
         raise
 

opensampl/vendors/adva.py

@@ -5,7 +5,7 @@
 import re
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import TextIO, Union
+from typing import ClassVar, TextIO, Union
 
 import click
 import pandas as pd
@@ -25,6 +25,13 @@ class AdvaProbe(BaseProbe):
     start_time: datetime
     vendor = VENDORS.ADVA
 
+    file_pattern: ClassVar = re.compile(
+        r"(?P<ip>\d+\.\d+\.\d+\.\d+)(?P<type>CLOCK_PROBE|PTP_CLOCK_PROBE)"
+        r"-(?P<identifier>\d+-\d+)-"
+        r"(?P<year>\d+)-(?P<month>\d+)-(?P<day>\d+)-"
+        r"(?P<hour>\d+)-(?P<minute>\d+)-(?P<second>\d+)\.txt(?:\.gz)?"
+    )
+
     class RandomDataConfig(BaseProbe.RandomDataConfig):
         """Model for storing random data generation configurations as provided by CLI or YAML"""
 
@@ -60,20 +67,19 @@ def __init__(self, input_file: Union[str, Path]):
         super().__init__(input_file=input_file)
         self.probe_key, self.timestamp = self.parse_file_name(self.input_file)
 
+    @classmethod
+    def filter_files(cls, files: list[Path]) -> list[Path]:
+        """Filter the files found in the input directory when loading to those which match the regex"""
+        return [f for f in files if cls.file_pattern.fullmatch(f.name)]
+
     @classmethod
     def parse_file_name(cls, file_name: Path) -> tuple[ProbeKey, datetime]:
         """
         Parse file name into identifying parts
 
         Expected format: <ip_address>CLOCK_PROBE-<probe_id>-YYYY-MM-DD-HH-MM-SS.txt.gz
         """
-        pattern = (
-            r"(?P<ip>\d+\.\d+\.\d+\.\d+)(?P<type>CLOCK_PROBE|PTP_CLOCK_PROBE)"
-            r"-(?P<identifier>\d+-\d+)-"
-            r"(?P<year>\d+)-(?P<month>\d+)-(?P<day>\d+)-"
-            r"(?P<hour>\d+)-(?P<minute>\d+)-(?P<second>\d+)\.txt(?:\.gz)?"
-        )
-        match = re.match(pattern, file_name.name)
+        match = re.match(cls.file_pattern, file_name.name)
         if match:
             ip_address = match.group("ip")
             probe_id = match.group("identifier")