import sys
from typing import Tuple, Type
from hgraph._types._time_series_types import TIME_SERIES_TYPE, OUT
from hgraph._types._ts_type import TS
from hgraph._types._tsl_type import TSL
from hgraph._types._scalar_types import SIZE, DEFAULT
from hgraph._types._tsb_type import TimeSeriesSchema, TSB, TS_SCHEMA, TS_SCHEMA_1
from hgraph._wiring._decorators import operator
__all__ = ("str_", "match_", "Match", "replace", "split", "join", "format_", "substr")
[docs]
@operator
def str_(ts: TIME_SERIES_TYPE) -> TS[str]:
"""
Converts the incoming time series to a string representation. Default implementation would be str(ts.value).
"""
...
[docs]
class Match(TimeSeriesSchema):
is_match: TS[bool]
groups: TS[Tuple[str, ...]]
[docs]
@operator
def match_(pattern: TS[str], s: TS[str]) -> TSB[Match]:
"""
Matches the pattern in the string and returns the result and matching groups.
"""
...
[docs]
@operator
def replace(pattern: TS[str], repl: TS[str], s: TS[str]) -> TS[str]:
"""
Replaces the pattern in the string with the replacement.
"""
...
[docs]
@operator # Would need to define substr in imports
def substr(s: TS[str], start: TS[int], end: TS[int] = None) -> TS[str]:
"""
Extracts a substring from the input string time series based on start and end positions.
:param s: Input string time series to perform extraction on.
:param start: Starting position (inclusive) of the substring. Negative values count from the end
:param end: Optional ending position (exclusive) of the substring. If None, extracts to the end of string.
:return: Time series containing the extracted substring
"""
...
[docs]
@operator
def split(s: TS[str], separator: str, maxsplits: int = sys.maxsize, to: Type[OUT] = TS[Tuple[str, ...]]) -> DEFAULT[OUT]:
"""
Splits the string over the separator into one of the given types:
- TS[Tuple[str, ...]],
- TS[Tuple[str, str]],
- TSL[TS[str], SIZE]
"""
...
[docs]
@operator
def join(*strings: TSL[TS[str], SIZE], separator: str) -> TS[str]:
"""
Joins the strings with the separator.
"""
...