Merge pull request #238 from abalabin-bamfunds/merge-adaptors-2026-01-16

Rewrite Python engine subscription management, minor fixes

Author: hhenson

cpp/CMakeLists.txt

@@ -16,7 +16,7 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 if (UNIX)
     set(CMAKE_CXX_FLAGS_RELEASE "-g -O3")
     set(CMAKE_CXX_FLAGS_RELWITHDEBINFO "-g -O3 -DNDEBUG")
-    set(CMAKE_CXX_FLAGS_DEBUG "-g -v")
+    set(CMAKE_CXX_FLAGS_DEBUG "-g")
     set(CMAKE_CXX_FLAGS "-Wall")
     # Sanitizers (Clang/GCC only)
     if (CMAKE_CXX_COMPILER_ID MATCHES "Clang|GNU")

cpp/src/cpp/types/tsb.cpp

@@ -83,7 +83,7 @@ namespace hgraph {
             size_t index{static_cast<size_t>(std::distance(_schema->keys().begin(), it))};
             return this->operator[](index);
         }
-        throw std::out_of_range("Key not found in TimeSeriesSchema");
+        throw std::out_of_range(std::format("Key {} not found in TimeSeriesSchema", key));
     }
 
     template<typename T_TS>

cpp/src/cpp/types/tsd.cpp

@@ -936,6 +936,7 @@ namespace hgraph
                     _add_key_value(pv_key.view(), value);
                 } else {
                     to_keep.emplace(pv_key.view().clone(), std::make_pair(value, was_valid));
+                    value->make_passive();
                 }
             }
             std::swap(_removed_items, to_keep);

hgraph/_impl/_builder/_ts_builder.py

@@ -52,18 +52,7 @@ class PythonOutputBuilder:
 
     def release_instance(self, item: "PythonTimeSeriesOutput"):
         if sys.exc_info()[0] is None:
-            if len(item._subscribers) != 0: 
-                try:
-                    logger.error(
-                        f"Output instance still has subscribers when released, this is a bug. \n"
-                        f"output belongs to node {item.owning_node}\n"
-                        f"subscriber nodes are {[i.owning_node if isinstance(i, TimeSeries) else i for i in item._subscribers]}\n\n"
-                        f"subscriber inputs are {[i for i in item._subscribers if isinstance(i, TimeSeries)]}\n\n"
-                        f"{item}"
-                    )
-                except Exception:
-                    ...
-            
+            item._subscribers.assert_empty(item)
             item._parent_or_node = None
 
 

hgraph/_impl/_operators/_conversion_operators/_date_time_conversion_operators.py

@@ -1,4 +1,4 @@
-from datetime import date, timedelta, datetime
+from datetime import date, time, timedelta, datetime
 
 from hgraph import compute_node, TS, OUT, WiringPort, combine, convert
 
@@ -24,6 +24,11 @@ def convert_datetime_to_date(ts: TS[datetime]) -> TS[date]:
     return v.date()
 
 
+@compute_node(overloads=combine)
+def combine_date_and_time(date: TS[date], time: TS[time]) -> TS[datetime]:
+    return datetime.combine(date.value, time.value)
+
+
 @compute_node(overloads=combine, requires=lambda m: m[OUT].py_type == TS[date])
 def combine_date(year: TS[int], month: TS[int], day: TS[int]) -> TS[date]:
     return date(year.value, month.value, day.value)

hgraph/_impl/_operators/_conversion_operators/_to_data_frame_converters.py

@@ -4,7 +4,7 @@
 
 from dataclasses import asdict
 from datetime import date, datetime
-from typing import Dict, Type, Callable
+from typing import Dict, Tuple, Type, Callable
 
 import polars as pl
 from frozendict import frozendict
@@ -324,6 +324,15 @@ def _check_schema(scalar, bundle):
     return True
 
 
+@compute_node(overloads=convert, requires=lambda m: m[OUT].py_type == TS[Frame] or m[OUT].matches_type(TS[Frame[m[COMPOUND_SCALAR].py_type]]))
+def convert_tuple_to_frame(
+    ts: TS[Tuple[COMPOUND_SCALAR]],
+    _tp: Type[TS[Frame[COMPOUND_SCALAR]]] = DEFAULT[OUT],
+    _cs: Type[COMPOUND_SCALAR] = AUTO_RESOLVE,
+) -> TS[Frame[COMPOUND_SCALAR]]:
+    return pl.DataFrame(ts.value)
+
+
 @compute_node(
     overloads=combine,
     requires=lambda m: _check_schema(m[COMPOUND_SCALAR], m[TS_SCHEMA]),

hgraph/_impl/_operators/_number_operators.py

@@ -19,6 +19,7 @@
     DivideByZero,
     ln,
 )
+from hgraph._operators._operators import sign
 from hgraph._types._scalar_types import NUMBER_2
 
 __all__ = tuple()
@@ -278,3 +279,19 @@ def eq_float_float(lhs: TS[float], rhs: TS[float], epsilon: TS[float] = EPSILON)
 @compute_node(overloads=ln)
 def ln_impl(ts: TS[float]) -> TS[float]:
     return math.log(ts.value)
+
+
+@compute_node(overloads=sign)
+def sign_impl(ts: TS[int]) -> TS[int]:
+    if ts.value < 0:
+        return -1
+    else:
+        return 1
+    
+    
+@compute_node(overloads=sign)
+def sign_float_impl(ts: TS[float]) -> TS[float]:
+    if ts.value < 0:
+        return -1.0
+    else:
+        return 1.0

hgraph/_impl/_operators/_stream_operators.py

@@ -118,6 +118,29 @@ def lag_timedelta_start(_state: STATE[LagState]):
     _state.buffer = deque[SCALAR]()
 
 
+@compute_node(overloads=lag)
+def lag_timedelta_ts(
+    ts: TIME_SERIES_TYPE,
+    period: TS[timedelta],
+    on_wall_clock: bool = False,
+    _scheduler: SCHEDULER = None,
+    _state: STATE[LagState] = None,
+) -> TIME_SERIES_TYPE:
+    # Uses the scheduler to keep track of when to deliver the values recorded in the buffer.
+    buffer: deque[SCALAR] = _state.buffer
+    if ts.modified:
+        buffer.append(ts.delta_value)
+        _scheduler.schedule(period.value, f"{ts.last_modified_time}", on_wall_clock=on_wall_clock)
+
+    if _scheduler.is_scheduled_now:
+        return buffer.popleft()
+
+
+@lag_timedelta_ts.start
+def lag_timedelta_ts_start(_state: STATE[LagState]):
+    _state.buffer = deque[SCALAR]()
+
+
 @generator(overloads=schedule)
 def schedule_scalar(delay: timedelta, initial_delay: bool = True, max_ticks: int = sys.maxsize) -> TS[bool]:
     initial_timedelta = delay if initial_delay else timedelta()

hgraph/_impl/_operators/_time_series_properties.py

@@ -9,6 +9,7 @@
     valid,
     last_modified_date,
     last_modified_time,
+    last_modified_wall_clock_time,
     modified,
     SCHEDULER,
     TS_OUT,
@@ -62,6 +63,11 @@ def last_modified_time_impl(ts: SIGNAL) -> TS[datetime]:
     return ts.last_modified_time
 
 
+@compute_node(overloads=last_modified_wall_clock_time)
+def last_modified_wall_clock_time_time_impl(ts: SIGNAL, _clock: EvaluationClock = None) -> TS[datetime]:
+    return _clock.now
+
+
 @compute_node(overloads=last_modified_date)
 def last_modified_date_impl(ts: SIGNAL) -> TS[date]:
     return ts.last_modified_time.date()

hgraph/_impl/_operators/_tuple_operators.py

@@ -140,6 +140,19 @@ def index_of_tuple(ts: TS[tuple[SCALAR, ...]], item: TS[SCALAR]) -> TS[int]:
         return -1
 
 
+@compute_node(overloads=add_, valid=lambda m, __strict__: ("lhs", "rhs") if __strict__ else tuple())
+def add_tuple_tuple(lhs: TS[Tuple[SCALAR, ...]], rhs: TS[Tuple[SCALAR, ...]], __strict__: bool = True) -> TS[Tuple[SCALAR, ...]]:
+    """Adds the tuples together."""
+    if lhs.valid:
+        if rhs.valid:
+            return lhs.value + rhs.value
+        else:
+            return lhs.value
+    elif rhs.valid:
+        # If lhs is not valid, rhs must be valid as we were activated by something ticking.
+        return rhs.value
+    
+
 @compute_node(overloads=add_, valid=tuple())
 def add_tuple_scalar(lhs: TS[Tuple[SCALAR, ...]], rhs: TS[SCALAR]) -> TS[Tuple[SCALAR, ...]]:
     """Adds the element to the tuple."""