API documentation

This module provides the core functionality of the Juice Core Uplink API Client in a more convenient way than the generated code.

May functionalities are not yet implemented, but some are.

SHTRestInterface

SHTRestInterface

Main entry point for interacting with the Juice Core Uplink API

Source code in src/juice_core/SHTRestInterface.py

@define(auto_attribs=True, eq=False)
class SHTRestInterface:
    """
    Main entry point for interacting with the Juice Core Uplink API
    """

    client: Client | None = None
    timeout: float = 40.0

    def __attrs_post_init__(self):
        if not self.client:
            self.client = Client(DEFAULT_URL)
        # self.client.timeout = self.timeout

    @cache
    @pandas_convertable
    def pcw(self):
        return get_pcw.sync(client=self.client)

    @cache
    @pandas_convertable
    def pcw_by_mnemonic(self, mnemonic: str):
        return get_pcw_by_mnemonic.sync(client=self.client, mnemonic=mnemonic)

    @cache
    @pandas_convertable(time_fields=["created"])
    def plans(self):
        """Retrieve all the plans available on the endpoint"""
        return get_plan.sync(client=self.client)

    def plan_id_by_name(self, name):
        """Retrieve the plan id from the plan name"""
        for plan in self.plans(as_pandas=False):
            if plan.name.lower().strip() == name.lower().strip():
                log.debug(f"Plan {name} has id {plan.id}")
                return plan.id

        log.warning(f"No plan with name {name} found")
        return None

    @cache
    @pandas_convertable(time_fields=["start", "end"])
    def plan_segments(self, plan_id_or_name):
        """Retrieve the segments of a plan"""
        plan = self.plan(plan_id_or_name, as_pandas=False)
        return plan.segments

    @cache
    @pandas_convertable(time_fields=["start", "end"])
    def engineering_segments(self, trajectory=DEFAULT_TRAJECTORY) -> pd.DataFrame:
        """Retrieve the engineering segments for a mnemonic"""
        return get_trajectory_engineering_segments_by_mnemonic.sync(
            mnemonic=trajectory,
            client=self.client,
        )

    @cache
    @pandas_convertable
    def plan(self, plan_id_or_name):
        """Retrieve the plan from the plan id or name"""
        if isinstance(plan_id_or_name, str):
            plan_id_or_name = self.plan_id_by_name(plan_id_or_name)

        return get_plan_by_id.sync(plan_id_or_name, client=self.client)

    @cache
    @pandas_convertable
    def known_series(self, trajectory=DEFAULT_TRAJECTORY):
        """Retrieve all the series available on the endpoint"""
        return get_trajectory_series_by_mnemonic.sync(
            client=self.client,
            mnemonic=trajectory,
        )

    @cache
    @pandas_convertable(is_timeseries=True)
    def series(
        self,
        series_name,
        trajectory=DEFAULT_TRAJECTORY,
        start=DEFAULT_START,
        end=DEFAULT_END,
    ):
        """Retrieve a serie from the endpoint"""

        q = {
            "start": str(start),
            "end": str(end),
            "trajectory": trajectory,
            "series": series_name,
        }

        body = json.dumps(q)
        return get_series.sync(client=self.client, body=body)

    def series_multi_(
        self,
        series_names,
        trajectory=DEFAULT_TRAJECTORY,
        start=DEFAULT_START,
        end=DEFAULT_END,
    ):
        loop = asyncio.get_event_loop()
        coroutine = self.series_multi(
            series_names,
            trajectory=trajectory,
            start=start,
            end=end,
        )
        return loop.run_until_complete(coroutine)

    def series_multi(
        self,
        series_names,
        trajectory=DEFAULT_TRAJECTORY,
        start=DEFAULT_START,
        end=DEFAULT_END,
    ):
        """Retrieve multiple series from the endpoint"""
        out = []
        for series_name in series_names:
            q = {
                "start": str(start),
                "end": str(end),
                "trajectory": trajectory,
                "series": series_name,
            }

            body = json.dumps(q)
            got = get_series.asyncio(client=self.client, body=body)
            out.append(got)

        return asyncio.gather(*out)

    @cache
    @pandas_convertable
    def event_types(self, trajectory=DEFAULT_TRAJECTORY):
        """Retrieve all the events applicable for a trajectory"""
        return get_trajectory_event_by_mnemonic.sync(
            client=self.client,
            mnemonic=trajectory,
        )

    @cache
    def segment_definition(self, mnemonic):
        return get_segment_definition_by_mnemonic.sync(
            client=self.client,
            mnemonic=mnemonic,
        )

    @pandas_convertable
    def segment_definitions(self, mnemonics: list[str]):
        return [self.segment_definition(m) for m in mnemonics]

    @cache
    @pandas_convertable(time_fields=["start", "end"])
    def events(
        self,
        mnemonics: list[str] | str = [],
        trajectory: str = DEFAULT_TRAJECTORY,
        start=DEFAULT_START,
        end=DEFAULT_END,
    ):
        """Retrieve events of a given type from the endpoint"""
        if isinstance(mnemonics, str):
            mnemonics = [mnemonics]

        if len(mnemonics) == 0:
            types = self.event_types(trajectory=trajectory, as_pandas=False)
            mnemonics = [m.mnemonic for m in types]
            log.info(f"Retrieving all known events {mnemonics}")

        q = {
            "start": str(start),
            "end": str(end),
            "trajectory": trajectory,
            "mnemonics": mnemonics,
        }

        body = json.dumps(q)
        return get_events.sync(client=self.client, body=body)

engineering_segments(trajectory=DEFAULT_TRAJECTORY) -> pd.DataFrame cached

Retrieve the engineering segments for a mnemonic

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable(time_fields=["start", "end"])
def engineering_segments(self, trajectory=DEFAULT_TRAJECTORY) -> pd.DataFrame:
    """Retrieve the engineering segments for a mnemonic"""
    return get_trajectory_engineering_segments_by_mnemonic.sync(
        mnemonic=trajectory,
        client=self.client,
    )

event_types(trajectory=DEFAULT_TRAJECTORY) cached

Retrieve all the events applicable for a trajectory

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable
def event_types(self, trajectory=DEFAULT_TRAJECTORY):
    """Retrieve all the events applicable for a trajectory"""
    return get_trajectory_event_by_mnemonic.sync(
        client=self.client,
        mnemonic=trajectory,
    )

events(mnemonics: list[str] | str = [], trajectory: str = DEFAULT_TRAJECTORY, start=DEFAULT_START, end=DEFAULT_END) cached

Retrieve events of a given type from the endpoint

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable(time_fields=["start", "end"])
def events(
    self,
    mnemonics: list[str] | str = [],
    trajectory: str = DEFAULT_TRAJECTORY,
    start=DEFAULT_START,
    end=DEFAULT_END,
):
    """Retrieve events of a given type from the endpoint"""
    if isinstance(mnemonics, str):
        mnemonics = [mnemonics]

    if len(mnemonics) == 0:
        types = self.event_types(trajectory=trajectory, as_pandas=False)
        mnemonics = [m.mnemonic for m in types]
        log.info(f"Retrieving all known events {mnemonics}")

    q = {
        "start": str(start),
        "end": str(end),
        "trajectory": trajectory,
        "mnemonics": mnemonics,
    }

    body = json.dumps(q)
    return get_events.sync(client=self.client, body=body)

known_series(trajectory=DEFAULT_TRAJECTORY) cached

Retrieve all the series available on the endpoint

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable
def known_series(self, trajectory=DEFAULT_TRAJECTORY):
    """Retrieve all the series available on the endpoint"""
    return get_trajectory_series_by_mnemonic.sync(
        client=self.client,
        mnemonic=trajectory,
    )

plan(plan_id_or_name) cached

Retrieve the plan from the plan id or name

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable
def plan(self, plan_id_or_name):
    """Retrieve the plan from the plan id or name"""
    if isinstance(plan_id_or_name, str):
        plan_id_or_name = self.plan_id_by_name(plan_id_or_name)

    return get_plan_by_id.sync(plan_id_or_name, client=self.client)

plan_id_by_name(name)

Retrieve the plan id from the plan name

Source code in src/juice_core/SHTRestInterface.py

def plan_id_by_name(self, name):
    """Retrieve the plan id from the plan name"""
    for plan in self.plans(as_pandas=False):
        if plan.name.lower().strip() == name.lower().strip():
            log.debug(f"Plan {name} has id {plan.id}")
            return plan.id

    log.warning(f"No plan with name {name} found")
    return None

plan_segments(plan_id_or_name) cached

Retrieve the segments of a plan

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable(time_fields=["start", "end"])
def plan_segments(self, plan_id_or_name):
    """Retrieve the segments of a plan"""
    plan = self.plan(plan_id_or_name, as_pandas=False)
    return plan.segments

plans() cached

Retrieve all the plans available on the endpoint

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable(time_fields=["created"])
def plans(self):
    """Retrieve all the plans available on the endpoint"""
    return get_plan.sync(client=self.client)

series(series_name, trajectory=DEFAULT_TRAJECTORY, start=DEFAULT_START, end=DEFAULT_END) cached

Retrieve a serie from the endpoint

Source code in src/juice_core/SHTRestInterface.py

@cache
@pandas_convertable(is_timeseries=True)
def series(
    self,
    series_name,
    trajectory=DEFAULT_TRAJECTORY,
    start=DEFAULT_START,
    end=DEFAULT_END,
):
    """Retrieve a serie from the endpoint"""

    q = {
        "start": str(start),
        "end": str(end),
        "trajectory": trajectory,
        "series": series_name,
    }

    body = json.dumps(q)
    return get_series.sync(client=self.client, body=body)

series_multi(series_names, trajectory=DEFAULT_TRAJECTORY, start=DEFAULT_START, end=DEFAULT_END)

Retrieve multiple series from the endpoint

Source code in src/juice_core/SHTRestInterface.py

def series_multi(
    self,
    series_names,
    trajectory=DEFAULT_TRAJECTORY,
    start=DEFAULT_START,
    end=DEFAULT_END,
):
    """Retrieve multiple series from the endpoint"""
    out = []
    for series_name in series_names:
        q = {
            "start": str(start),
            "end": str(end),
            "trajectory": trajectory,
            "series": series_name,
        }

        body = json.dumps(q)
        got = get_series.asyncio(client=self.client, body=body)
        out.append(got)

    return asyncio.gather(*out)

expand_column(tab: pd.DataFrame, column_name='description') -> pd.DataFrame

Some tables have a description column that contains additional information.

This function expands the description column into multiple columns.

Parameters

tab (pd.DataFrame): table to expand, must have a description column

Returns

pd.DataFrame: a new dataframe with the description column expanded

Source code in src/juice_core/SHTRestInterface.py

def expand_column(tab: pd.DataFrame, column_name="description") -> pd.DataFrame:
    """Some tables have a description column that contains additional information.

    This function expands the description column into multiple columns.

    Parameters
    ----------
    tab (pd.DataFrame):
        table to expand, must have a description column

    Returns
    -------
    pd.DataFrame:
        a new dataframe with the description column expanded

    """
    additional_columns = []
    for d in tab[column_name]:
        values = {}
        for item in d.split(";"):
            key, value = item.split("=")
            values[key.strip()] = value.strip()

        additional_columns.append(values)

    tab_ = tab.drop(columns=[column_name], inplace=False)
    newd = pd.DataFrame(additional_columns)

    return tab_.join(newd)