Nortech

Main class for interacting with the Nortech SDK.

Attributes:

• metadata Metadata - Client for interacting with the Nortech Metadata API.
• datatools Datatools - Client for interacting with the Nortech Datatools API.
• derivers Derivers - Client for interacting with the Nortech Derivers API.

constructor

def __init__(url: str = "https://api.apps.nor.tech",
             api_key: str | None = None,
             ignore_pagination: bool | None = None,
             user_agent: str | None = None,
             experimental_features: bool | None = None,
             timeout: float | Timeout | None = None,
             retry: int | Retry | None = None)

Initialize the Nortech class.

Arguments:

• url str - The URL of the Nortech API. Defaults to "https://api.apps.nor.tech".
• api_key str | None - The API key for the Nortech API.
• ignore_pagination bool | None - Whether to ignore pagination.
• user_agent str | None - The user agent for the Nortech API.
• experimental_features bool | None - Whether to enable experimental features.
• timeout float | Timeout | None - The timeout setting for the API request. From urllib3 package.
• retry int | Retry | None - The retry setting for the API request. From urllib3 package.

Example:

from urllib3 import Retry, Timeout

from nortech import Nortech

nortech = Nortech()  # Uses environment variables for configs

nortech = Nortech(api_key="my_api_key")  # Sets the API key

nortech = Nortech(ignore_pagination=False)  # Use pagination

nortech = Nortech(user_agent="my_user_agent")  # Sets the user agent

nortech = Nortech(timeout=Timeout(connect=10, read=60))  # Sets the timeout

nortech = Nortech(
    retry=Retry(
        total=5,
        backoff_factor=1,
        status_forcelist=[502, 503, 504],
        allowed_methods=["GET", "POST"],
        raise_on_status=False,
    )
)  # Sets the retry configuration

metadata

Metadata

Client for interacting with the Nortech Metadata API.

Attributes:

• workspace Workspace - Client for interacting with the Nortech Metadata Workspace API.
• asset Asset - Client for interacting with the Nortech Metadata Asset API.
• division Division - Client for interacting with the Nortech Metadata Division API.
• unit Unit - Client for interacting with the Nortech Metadata Unit API.
• signal Signal - Client for interacting with the Nortech Metadata Signal API.

Workspace

Workspace.

get

def get(
    workspace: int | str | WorkspaceInputDict | WorkspaceInput
    | WorkspaceOutput | WorkspaceListOutput
) -> WorkspaceOutput

Get a workspace by ID or name.

Arguments:

• workspace int | str | WorkspaceInputDict | WorkspaceInput | WorkspaceOutput | WorkspaceListOutput - The workspace identifier, which can be:
  • int: The workspace "ID".
  • str: The workspace "name".
  • WorkspaceInputDict: A dictionary representation of a workspace input.
  • WorkspaceInput: A pydantic model representing a workspace input.
  • WorkspaceOutput: A pydantic model representing a workspace output. Obtained from requesting a workspace metadata.
  • WorkspaceListOutput: A pydantic model representing a listed workspace output. Obtained from requesting workspaces metadata.

Returns:

• WorkspaceOutput - The workspace details.

Example:

from nortech import Nortech
from nortech.metadata.values.workspace import WorkspaceInput

nortech = Nortech()

# Get by ID
workspace = nortech.metadata.workspace.get(123)

# Get by name
workspace = nortech.metadata.workspace.get("my-workspace")

# Get by input dict
workspace = nortech.metadata.workspace.get({"workspace": "my-workspace"})

# Get by WorkspaceInput pydantic object
workspace = nortech.metadata.workspace.get(WorkspaceInput(workspace="my-workspace"))

print(workspace)
# WorkspaceOutput(
#     id=123,
#     name="my-workspace",
#     description="my-description",
#     created_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     updated_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0)
# )

list

def list(
    pagination_options: PaginationOptions[Literal["id", "name", "description"]]
    | None = None
) -> PaginatedResponse[WorkspaceListOutput, Literal["id", "name",
                                                    "description"]]

List all workspaces.

Arguments:

• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[WorkspaceListOutput] - A paginated list of workspaces.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all workspaces
workspaces = nortech.metadata.workspace.list()

# List with pagination
workspaces = nortech.metadata.workspace.list(PaginationOptions(size=10, sortBy="name"))

print(workspaces)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         WorkspaceListOutput(
#             id=1,
#             name="my-workspace",
#             description="my-description"
#         ),
#         WorkspaceListOutput(
#             id=2,
#             name="my-workspace",
#             description="my-description"
#         )
#     ]
# )

Asset

get

def get(
    asset: int | AssetInputDict | AssetInput | AssetOutput | AssetListOutput
) -> AssetOutput

Get an asset by ID or input.

Arguments:

• asset int | AssetInputDict | AssetInput | AssetOutput | AssetListOutput - The asset identifier, which can be:
  • int: The asset "ID".
  • AssetInputDict: A dictionary representation of an asset input.
  • AssetInput: A pydantic model representing an asset input.
  • AssetOutput: A pydantic model representing an asset output. Obtained from requesting an asset metadata.
  • AssetListOutput: A pydantic model representing a listed asset output. Obtained from requesting assets metadata.

Returns:

• AssetOutput - The asset details.

Example:

from nortech import Nortech
from nortech.metadata.values.asset import AssetInput

nortech = Nortech()

# Get by ID
asset = nortech.metadata.asset.get(123)

# Get by input dict
asset = nortech.metadata.asset.get({"workspace": "my-workspace", "asset": "my-asset"})

# Get by AssetInput pydantic object
asset = nortech.metadata.asset.get(AssetInput(workspace="my-workspace", asset="my-asset"))

print(asset)
# AssetOutput(
#     id=123,
#     name="my-asset",
#     description="my-description",
#     created_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     updated_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     workspace=MetadataOutput(
#         id=123,
#         name="my-workspace",
#     )
# )

list

def list(
    workspace: int | str | WorkspaceInputDict | WorkspaceInput
    | WorkspaceOutput | WorkspaceListOutput,
    pagination_options: PaginationOptions[Literal["id", "name", "description"]]
    | None = None
) -> PaginatedResponse[AssetListOutput, Literal["id", "name", "description"]]

List all assets in a workspace.

Arguments:

• workspace int | str | WorkspaceInputDict | WorkspaceInput | WorkspaceOutput | WorkspaceListOutput - The workspace identifier, which can be:
  • int: The workspace "ID".
  • str: The workspace "name".
  • WorkspaceInputDict: A dictionary representation of a workspace input.
  • WorkspaceInput: A pydantic model representing a workspace input.
  • WorkspaceOutput: A pydantic model representing a workspace output.
  • WorkspaceListOutput: A pydantic model representing a listed workspace output.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[AssetListOutput] - A paginated list of assets.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions
from nortech.metadata.values.workspace import WorkspaceInput

nortech = Nortech()

# List all assets in a workspace
assets = nortech.metadata.asset.list(123)  # using workspace ID

# List with pagination
assets = nortech.metadata.asset.list(
    "my-workspace",  # using workspace name
    PaginationOptions(size=10, sortBy="name"),
)

# Using WorkspaceInputDict dictionary
assets = nortech.metadata.asset.list({"workspace": "my-workspace"})

# Using WorkspaceInput pydantic object
assets = nortech.metadata.asset.list(WorkspaceInput(workspace="my-workspace"))

print(assets)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         AssetListOutput(
#             id=1,
#             name="my-asset",
#             description="my-description"
#         ),
#         AssetListOutput(
#             id=2,
#             name="another-asset",
#             description="another-description"
#         )
#     ]
# )

Division

get

def get(
    division: int | DivisionInputDict | DivisionInput | DivisionOutput
    | DivisionListOutput
) -> DivisionOutput

Get a division by ID or input.

Arguments:

• division int | DivisionInputDict | DivisionInput | DivisionOutput | DivisionListOutput - The division identifier, which can be:
  • int: The division "ID".
  • DivisionInputDict: A dictionary representation of a division input.
  • DivisionInput: A pydantic model representing a division input.
  • DivisionOutput: A pydantic model representing a division output. Obtained from requesting a division metadata.
  • DivisionListOutput: A pydantic model representing a listed division output. Obtained from requesting divisions metadata.

Returns:

• DivisionOutput - The division details.

Example:

from nortech import Nortech
from nortech.metadata.values.division import DivisionInput

nortech = Nortech()

# Get by ID
division = nortech.metadata.division.get(123)

# Get by input dict
division = nortech.metadata.division.get({"workspace": "my-workspace", "asset": "my-asset", "division": "my-division"})

# Get by DivisionInput pydantic object
division = nortech.metadata.division.get(
    DivisionInput(workspace="my-workspace", asset="my-asset", division="my-division")
)

print(division)
# DivisionOutput(
#     id=123,
#     name="my-division",
#     description="my-description",
#     created_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     updated_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     workspace=MetadataOutput(
#         id=123,
#         name="my-workspace"
#     ),
#     asset=MetadataOutput(
#         id=456,
#         name="my-asset"
#     )
# )

list

def list(
    asset: int | AssetInputDict | AssetInput | AssetOutput | AssetListOutput,
    pagination_options: PaginationOptions[Literal["id", "name", "description"]]
    | None = None
) -> PaginatedResponse[DivisionListOutput, Literal["id", "name",
                                                   "description"]]

List all divisions in an asset.

Arguments:

• asset int | AssetInputDict | AssetInput | AssetOutput | AssetListOutput - The asset identifier, which can be:
  • int: The asset "ID".
  • AssetInputDict: A dictionary representation of an asset input.
  • AssetInput: A pydantic model representing an asset input.
  • AssetOutput: A pydantic model representing an asset output.
  • AssetListOutput: A pydantic model representing a listed asset output.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[DivisionListOutput] - A paginated list of divisions.

Example:

from nortech import Nortech
from nortech.metadata.values.asset import AssetInput
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all divisions in an asset
divisions = nortech.metadata.division.list(123)  # using asset ID

# List with pagination
divisions = nortech.metadata.division.list(
    {"workspace": "my-workspace", "asset": "my-asset"},  # using AssetInputDict
    PaginationOptions(size=10, sortBy="name"),
)

# Using AssetInput pydantic object
divisions = nortech.metadata.division.list(AssetInput(workspace="my-workspace", asset="my-asset"))

print(divisions)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         DivisionListOutput(
#             id=1,
#             name="my-division",
#             description="my-description"
#         ),
#         DivisionListOutput(
#             id=2,
#             name="another-division",
#             description="another-description"
#         )
#     ]
# )

list_by_workspace_id

def list_by_workspace_id(
    workspace_id: int,
    pagination_options: PaginationOptions[Literal["id", "name", "description"]]
    | None = None
) -> PaginatedResponse[DivisionListOutput, Literal["id", "name",
                                                   "description"]]

List all divisions in a workspace.

Arguments:

• workspace_id int - The workspace ID.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[DivisionListOutput] - A paginated list of divisions.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all divisions in a workspace
divisions = nortech.metadata.division.list_by_workspace_id(123)

# List with pagination
divisions = nortech.metadata.division.list_by_workspace_id(123, PaginationOptions(size=10, sortBy="name"))

print(divisions)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         DivisionListOutput(
#             id=1,
#             name="my-division",
#             description="my-description"
#         ),
#         DivisionListOutput(
#             id=2,
#             name="another-division",
#             description="another-description"
#         )
#     ]
# )

Unit

get

def get(
    unit: int | UnitInputDict | UnitInput | UnitOutput | UnitListOutput
) -> UnitOutput

Get a unit by ID or input.

Arguments:

• unit int | UnitInputDict | UnitInput | UnitOutput | UnitListOutput - The unit identifier, which can be:
  • int: The unit "ID".
  • UnitInputDict: A dictionary representation of a unit input.
  • UnitInput: A pydantic model representing a unit input.
  • UnitOutput: A pydantic model representing a unit output. Obtained from requesting a unit metadata.
  • UnitListOutput: A pydantic model representing a listed unit output. Obtained from requesting units metadata.

Returns:

• UnitOutput - The unit details.

Example:

from nortech import Nortech
from nortech.metadata.values.unit import UnitInput

nortech = Nortech()

# Get by ID
unit = nortech.metadata.unit.get(123)

# Get by input dict
unit = nortech.metadata.unit.get(
    {"workspace": "my-workspace", "asset": "my-asset", "division": "my-division", "unit": "my-unit"}
)

# Get by UnitInput pydantic object
unit = nortech.metadata.unit.get(
    UnitInput(workspace="my-workspace", asset="my-asset", division="my-division", unit="my-unit")
)

print(unit)
# UnitOutput(
#     id=123,
#     name="my-unit",
#     created_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     updated_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     workspace=MetadataOutput(
#         id=123,
#         name="my-workspace"
#     ),
#     asset=MetadataOutput(
#         id=456,
#         name="my-asset"
#     ),
#     division=MetadataOutput(
#         id=789,
#         name="my-division"
#     )
# )

list

def list(
    division: int | DivisionInputDict | DivisionInput | DivisionOutput
    | DivisionListOutput,
    pagination_options: PaginationOptions[Literal["id", "name"]] | None = None
) -> PaginatedResponse[UnitListOutput, Literal["id", "name"]]

List all units in a division.

Arguments:

• division int | DivisionInputDict | DivisionInput | DivisionOutput | DivisionListOutput - The division identifier, which can be:
  • int: The division "ID".
  • DivisionInputDict: A dictionary representation of a division input.
  • DivisionInput: A pydantic model representing a division input.
  • DivisionOutput: A pydantic model representing a division output.
  • DivisionListOutput: A pydantic model representing a listed division output.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[UnitListOutput] - A paginated list of units.

Example:

from nortech import Nortech
from nortech.metadata.values.division import DivisionInput
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all units in a division
units = nortech.metadata.unit.list(123)  # using division ID

# List with pagination
units = nortech.metadata.unit.list(
    {"workspace": "my-workspace", "asset": "my-asset", "division": "my-division"},  # using DivisionInputDict
    PaginationOptions(size=10, sortBy="name"),
)

# Using DivisionInput pydantic object
units = nortech.metadata.unit.list(DivisionInput(workspace="my-workspace", asset="my-asset", division="my-division"))

print(units)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         UnitListOutput(
#             id=1,
#             name="my-unit"
#         ),
#         UnitListOutput(
#             id=2,
#             name="another-unit"
#         )
#     ]
# )

list_by_workspace_id

def list_by_workspace_id(
    workspace_id: int,
    pagination_options: PaginationOptions[Literal["id", "name"]] | None = None
) -> PaginatedResponse[UnitListOutput, Literal["id", "name"]]

List all units in a workspace.

Arguments:

• workspace_id int - The workspace ID.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[UnitListOutput] - A paginated list of units.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all units in a workspace
units = nortech.metadata.unit.list_by_workspace_id(123)

# List with pagination
units = nortech.metadata.unit.list_by_workspace_id(123, PaginationOptions(size=10, sortBy="name"))

print(units)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         UnitListOutput(
#             id=1,
#             name="my-unit"
#         ),
#         UnitListOutput(
#             id=2,
#             name="another-unit"
#         )
#     ]
# )

list_by_asset_id

def list_by_asset_id(
    asset_id: int,
    pagination_options: PaginationOptions[Literal["id", "name"]] | None = None
) -> PaginatedResponse[UnitListOutput, Literal["id", "name"]]

List all units in an asset.

Arguments:

• asset_id int - The asset ID.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[UnitListOutput] - A paginated list of units.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all units in an asset
units = nortech.metadata.unit.list_by_asset_id(123)

# List with pagination
units = nortech.metadata.unit.list_by_asset_id(123, PaginationOptions(size=10, sortBy="name"))

print(units)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         UnitListOutput(
#             id=1,
#             name="my-unit"
#         ),
#         UnitListOutput(
#             id=2,
#             name="another-unit"
#         )
#     ]
# )

Signal

get

def get(
    signal: int | SignalInputDict | SignalInput | SignalOutput
    | SignalListOutput
) -> SignalOutput

Get a signal by ID or input.

Arguments:

• signal int | SignalInputDict | SignalInput | SignalOutput | SignalListOutput - The signal identifier, which can be:
  • int: The signal "ID".
  • SignalInputDict: A dictionary representation of a signal input.
  • SignalInput: A pydantic model representing a signal input.
  • SignalOutput: A pydantic model representing a signal output. Obtained from requesting a signal metadata.
  • SignalListOutput: A pydantic model representing a listed signal output. Obtained from requesting signals metadata.

Returns:

• SignalOutput - The signal details.

Example:

from nortech import Nortech
from nortech.metadata.values.signal import SignalInput

nortech = Nortech()

# Get by ID
signal = nortech.metadata.signal.get(123)

# Get unit signal by input dict
signal = nortech.metadata.signal.get(
    {
        "workspace": "my-workspace",
        "asset": "my-asset",
        "division": "my-division",
        "unit": "my-unit",
        "signal": "my-signal",
    }
)


# Get by SignalInput pydantic object
signal = nortech.metadata.signal.get(
    SignalInput(workspace="my-workspace", asset="my-asset", division="my-division", unit="my-unit", signal="my-signal")
)

print(signal)
# SignalOutput(
#     id=123,
#     name="my-signal",
#     physical_unit="°C",
#     data_type="float64",
#     description="Temperature sensor",
#     long_description="Main temperature sensor for the unit",
#     created_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     updated_at=datetime.datetime(2024, 1, 1, 0, 0, 0, 0),
#     workspace=MetadataOutput(
#         id=123,
#         name="my-workspace"
#     ),
#     asset=MetadataOutput(
#         id=456,
#         name="my-asset"
#     ),
#     division=MetadataOutput(
#         id=789,
#         name="my-division"
#     ),
#     unit=MetadataOutput(
#         id=101,
#         name="my-unit"
#     )
# )

list

def list(
    unit: int | UnitInputDict | UnitInput | UnitOutput,
    pagination_options: PaginationOptions[Literal[
        "id",
        "name",
        "physical_unit",
        "data_type",
        "description",
        "long_description",
    ]]
    | None = None
) -> PaginatedResponse[SignalListOutput,
                       Literal["id", "name", "physical_unit", "data_type",
                               "description", "long_description"]]

List all signals in a unit.

Arguments:

• unit int | UnitInputDict | UnitInput | UnitOutput - The unit identifier, which can be:
  • int: The unit "ID".
  • UnitInputDict: A dictionary representation of a unit input.
  • UnitInput: A pydantic model representing a unit input.
  • UnitOutput: A pydantic model representing a unit output.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[SignalListOutput] - A paginated list of signals.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions
from nortech.metadata.values.unit import UnitInput

nortech = Nortech()

# List all signals in a unit
signals = nortech.metadata.signal.list(123)  # using unit ID

# List unit signals with pagination
signals = nortech.metadata.signal.list(
    {"workspace": "my-workspace", "asset": "my-asset", "division": "my-division", "unit": "my-unit"},
    PaginationOptions(size=10, sortBy="name"),
)

# Using UnitInput pydantic object
signals = nortech.metadata.signal.list(
    UnitInput(workspace="my-workspace", asset="my-asset", division="my-division", unit="my-unit")
)

print(signals)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         SignalListOutput(
#             id=1,
#             name="my-signal",
#             physical_unit="°C",
#             data_type="float64",
#             description="Temperature sensor",
#             long_description="Main temperature sensor for the unit"
#         ),
#         SignalListOutput(
#             id=2,
#             name="another-signal",
#             physical_unit="bar",
#             data_type="float64",
#             description="Pressure sensor",
#             long_description="Main pressure sensor for the unit"
#         )
#     ]
# )

list_by_workspace_id

def list_by_workspace_id(
    workspace_id: int,
    pagination_options: PaginationOptions[Literal[
        "id",
        "name",
        "physical_unit",
        "data_type",
        "description",
        "long_description",
    ]]
    | None = None
) -> PaginatedResponse[SignalListOutput,
                       Literal["id", "name", "physical_unit", "data_type",
                               "description", "long_description"]]

List all signals in a workspace.

Arguments:

• workspace_id int - The workspace ID.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[SignalListOutput] - A paginated list of signals.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all signals in a workspace
signals = nortech.metadata.signal.list_by_workspace_id(123)

# List with pagination
signals = nortech.metadata.signal.list_by_workspace_id(123, PaginationOptions(size=10, sortBy="name"))

print(signals)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         SignalListOutput(
#             id=1,
#             name="my-signal",
#             physical_unit="°C",
#             data_type="float64",
#             description="Temperature sensor",
#             long_description="Main temperature sensor for the unit"
#         ),
#         SignalListOutput(
#             id=2,
#             name="another-signal",
#             physical_unit="bar",
#             data_type="float64",
#             description="Pressure sensor",
#             long_description="Main pressure sensor for the unit"
#         )
#     ]
# )

list_by_asset_id

def list_by_asset_id(
    asset_id: int,
    pagination_options: PaginationOptions[Literal[
        "id",
        "name",
        "physical_unit",
        "data_type",
        "description",
        "long_description",
    ]]
    | None = None
) -> PaginatedResponse[SignalListOutput,
                       Literal["id", "name", "physical_unit", "data_type",
                               "description", "long_description"]]

List all signals in an asset.

Arguments:

• asset_id int - The asset ID.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[SignalListOutput] - A paginated list of signals.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all signals in an asset
signals = nortech.metadata.signal.list_by_asset_id(123)

# List with pagination
signals = nortech.metadata.signal.list_by_asset_id(123, PaginationOptions(size=10, sortBy="name"))

print(signals)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         SignalListOutput(
#             id=1,
#             name="my-signal",
#             physical_unit="°C",
#             data_type="float64",
#             description="Temperature sensor",
#             long_description="Main temperature sensor for the unit"
#         ),
#         SignalListOutput(
#             id=2,
#             name="another-signal",
#             physical_unit="bar",
#             data_type="float64",
#             description="Pressure sensor",
#             long_description="Main pressure sensor for the unit"
#         )
#     ]
# )

list_by_division_id

def list_by_division_id(
    division_id: int,
    pagination_options: PaginationOptions[Literal[
        "id",
        "name",
        "physical_unit",
        "data_type",
        "description",
        "long_description",
    ]]
    | None = None
) -> PaginatedResponse[SignalListOutput,
                       Literal["id", "name", "physical_unit", "data_type",
                               "description", "long_description"]]

List all signals in a division.

Arguments:

• division_id int - The division ID.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

• PaginatedResponse[SignalListOutput] - A paginated list of signals.

Example:

from nortech import Nortech
from nortech.metadata.values.pagination import PaginationOptions

nortech = Nortech()

# List all signals in a division
signals = nortech.metadata.signal.list_by_division_id(123)

# List with pagination
signals = nortech.metadata.signal.list_by_division_id(123, PaginationOptions(size=10, sortBy="name"))

print(signals)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         SignalListOutput(
#             id=1,
#             name="my-signal",
#             physical_unit="°C",
#             data_type="float64",
#             description="Temperature sensor",
#             long_description="Main temperature sensor for the unit"
#         ),
#         SignalListOutput(
#             id=2,
#             name="another-signal",
#             physical_unit="bar",
#             data_type="float64",
#             description="Pressure sensor",
#             long_description="Main pressure sensor for the unit"
#         )
#     ]
# )

datatools

Download

download_data

def download_data(signals: Sequence[int | SignalInput | SignalInputDict
                                    | SignalOutput | SignalListOutput],
                  time_window: TimeWindow, output_path: str,
                  file_format: Format)

Download data for the specified signals within the given time window. If experimental features are enabled, live data will also be downloaded.

Arguments:

• signals Sequence[int | SignalInput | SignalInputDict | SignalOutput | SignalListOutput] - A list of signals to download, which can be of the following types:
  • int: The signal "ID".
  • SignalInputDict: A dictionary representation of a signal input.
  • SignalInput: A pydantic model representing a signal input.
  • SignalOutput: A pydantic model representing a signal output. Obtained from requesting a signal metadata.
  • SignalListOutput: A pydantic model representing a listed signal output. Obtained from requesting signals metadata.
• time_window TimeWindow - The time window for which data should be downloaded.
• output_path str - The file path where the downloaded data will be saved.
• file_format Format - The format of the output file. Can be "parquet", "csv", or "json".

Raises:

• NotImplementedError - If the time window corresponds to hot storage, which is not yet supported.

Example:

from datetime import datetime

from nortech import Nortech
from nortech.datatools.values.windowing import TimeWindow
from nortech.metadata.values.signal import SignalInput, SignalInputDict

# Initialize the Nortech client
nortech = Nortech()

# Define signals to download
signal1: SignalInputDict = {
    "workspace": "workspace1",
    "asset": "asset1",
    "division": "division1",
    "unit": "unit1",
    "signal": "signal1",
}
signal2 = 789  # Signal ID
signal3 = SignalInput(workspace="workspace2", asset="asset2", division="division2", unit="unit2", signal="signal2")

fetched_signals = nortech.metadata.signal.list(  # Fetched signals
    {"workspace": "workspace3", "asset": "asset3", "division": "division3", "unit": "unit3"}
).data

# Define the time window for data download
my_time_window = TimeWindow(start=datetime(2023, 1, 1), end=datetime(2023, 1, 31))

# Specify the output path and file format
output_path = "path/to/output"
file_format = "parquet"

# Call the download_data function with manually defined signals or fetched signals
nortech.datatools.download.download_data(
    signals=[signal1, signal2, signal3] + fetched_signals,
    time_window=my_time_window,
    output_path=output_path,
    file_format=file_format,
)

Pandas

get_df

def get_df(signals: Sequence[int | SignalInput | SignalInputDict | SignalOutput
                             | SignalListOutput],
           time_window: TimeWindow) -> DataFrame

Retrieve a pandas DataFrame for the specified signals within the given time window. If experimental features are enabled, live data will also be retrieved.

Arguments:

• signals Sequence[int | SignalInput | SignalInputDict | SignalOutput | SignalListOutput] - A list of signals to download, which can be of the following types:
  • int: The signal "ID".
  • SignalInputDict: A dictionary representation of a signal input.
  • SignalInput: A pydantic model representing a signal input.
  • SignalOutput: A pydantic model representing a signal output. Obtained from requesting a signal metadata.
  • SignalListOutput: A pydantic model representing a listed signal output. Obtained from requesting signals metadata.
• time_window TimeWindow - The time window for which data should be retrieved.

Returns:

• DataFrame - A pandas DataFrame containing the data.

Raises:

• NoSignalsRequestedError - Raised when no signals are requested.
• InvalidTimeWindow - Raised when the start date is after the end date.

Example:

from datetime import datetime

from nortech import Nortech
from nortech.datatools.values.windowing import TimeWindow
from nortech.metadata.values.signal import SignalInput, SignalInputDict

# Initialize the Nortech client
nortech = Nortech()

# Define signals to download
signal1: SignalInputDict = {
    "workspace": "workspace1",
    "asset": "asset1",
    "division": "division1",
    "unit": "unit1",
    "signal": "signal1",
}
signal2 = 789  # Signal ID
signal3 = SignalInput(workspace="workspace2", asset="asset2", division="division2", unit="unit2", signal="signal2")

fetched_signals = nortech.metadata.signal.list(  # Fetched signals
    {"workspace": "workspace3", "asset": "asset3", "division": "division3", "unit": "unit3"}
).data

# Define the time window for data download
my_time_window = TimeWindow(start=datetime(2023, 1, 1), end=datetime(2023, 1, 31))

# Call the get_df function with manually defined signals or fetched signals
df = nortech.datatools.pandas.get_df(
    signals=[signal1, signal2, signal3] + fetched_signals,
    time_window=my_time_window,
)

print(df.columns)
# [
#     "timestamp",
#     "workspace_1/asset_1/division_1/unit_1/signal_1",
#     "workspace_1/asset_1/division_1/unit_1/signal_2",
#     "workspace_2/asset_2/division_2/unit_2/signal_3",
#     "workspace_3/asset_3/division_3/unit_3/signal_4",
#     "workspace_3/asset_3/division_3/unit_3/signal_5",
# ]

Polars

get_lazy_df

def get_lazy_df(signals: Sequence[int | SignalInput | SignalInputDict
                                  | SignalOutput | SignalListOutput],
                time_window: TimeWindow) -> LazyFrame

Retrieve a polars LazyFrame for the specified signals within the given time window. If experimental features are enabled, live data will also be retrieved.

Arguments:

• signals Sequence[int | SignalInput | SignalInputDict | SignalOutput | SignalListOutput] - A list of signals to download, which can be of the following types:
  • int: The signal "ID".
  • SignalInputDict: A dictionary representation of a signal input.
  • SignalInput: A pydantic model representing a signal input.
  • SignalOutput: A pydantic model representing a signal output. Obtained from requesting a signal metadata.
  • SignalListOutput: A pydantic model representing a listed signal output. Obtained from requesting signals metadata.
• time_window TimeWindow - The time window for which data should be retrieved.

Returns:

• LazyFrame - A polars LazyFrame containing the data.

Raises:

• NoSignalsRequestedError - Raised when no signals are requested.
• InvalidTimeWindow - Raised when the start date is after the end date.

Example:

from datetime import datetime

from nortech import Nortech
from nortech.datatools.values.windowing import TimeWindow
from nortech.metadata.values.signal import SignalInput, SignalInputDict

# Initialize the Nortech client
nortech = Nortech()

# Define signals to download
signal1: SignalInputDict = {
    "workspace": "workspace1",
    "asset": "asset1",
    "division": "division1",
    "unit": "unit1",
    "signal": "signal1",
}
signal2 = 789  # Signal ID
signal3 = SignalInput(workspace="workspace2", asset="asset2", division="division2", unit="unit2", signal="signal2")

fetched_signals = nortech.metadata.signal.list(  # Fetched signals
    {"workspace": "workspace3", "asset": "asset3", "division": "division3", "unit": "unit3"}
).data

# Define the time window for data download
my_time_window = TimeWindow(start=datetime(2023, 1, 1), end=datetime(2023, 1, 31))

# Call the get_df function with manually defined signals or fetched signals
df = nortech.datatools.polars.get_lazy_df(
    signals=[signal1, signal2, signal3] + fetched_signals,
    time_window=my_time_window,
)

print(df.columns)
# [
#     "timestamp",
#     "workspace_1/asset_1/division_1/unit_1/signal_1",
#     "workspace_1/asset_1/division_1/unit_1/signal_2",
#     "workspace_2/asset_2/division_2/unit_2/signal_3",
#     "workspace_3/asset_3/division_3/unit_3/signal_4",
#     "workspace_3/asset_3/division_3/unit_3/signal_5",
# ]

get_df

def get_df(signals: Sequence[int | SignalInput | SignalInputDict | SignalOutput
                             | SignalListOutput],
           time_window: TimeWindow) -> PolarsDataFrame

Retrieve a polars DataFrame for the specified signals within the given time window. If experimental features are enabled, live data will also be retrieved.

Arguments:

• signals Sequence[int | SignalInput | SignalInputDict | SignalOutput | SignalListOutput] - A list of signals to download, which can be of the following types:
  • int: The signal "ID".
  • SignalInputDict: A dictionary representation of a signal input.
  • SignalInput: A pydantic model representing a signal input.
  • SignalOutput: A pydantic model representing a signal output. Obtained from requesting a signal metadata.
  • SignalListOutput: A pydantic model representing a listed signal output. Obtained from requesting signals metadata.
• time_window TimeWindow - The time window for which data should be retrieved.

Returns:

• DataFrame - A polars DataFrame containing the data.

Raises:

• NoSignalsRequestedError - Raised when no signals are requested.
• InvalidTimeWindow - Raised when the start date is after the end date.

Example:

from datetime import datetime

from nortech import Nortech
from nortech.datatools.values.windowing import TimeWindow
from nortech.metadata.values.signal import SignalInput, SignalInputDict

# Initialize the Nortech client
nortech = Nortech()

# Define signals to download
signal1: SignalInputDict = {
    "workspace": "workspace1",
    "asset": "asset1",
    "division": "division1",
    "unit": "unit1",
    "signal": "signal1",
}
signal2 = 789  # Signal ID
signal3 = SignalInput(workspace="workspace2", asset="asset2", division="division2", unit="unit2", signal="signal2")

fetched_signals = nortech.metadata.signal.list(  # Fetched signals
    {"workspace": "workspace3", "asset": "asset3", "division": "division3", "unit": "unit3"}
).data

# Define the time window for data download
my_time_window = TimeWindow(start=datetime(2023, 1, 1), end=datetime(2023, 1, 31))

# Call the get_df function with manually defined signals or fetched signals
df = nortech.datatools.polars.get_df(
    signals=[signal1, signal2, signal3] + fetched_signals,
    time_window=my_time_window,
)

print(df.columns)
# [
#     "timestamp",
#     "workspace_1/asset_1/division_1/unit_1/signal_1",
#     "workspace_1/asset_1/division_1/unit_1/signal_2",
#     "workspace_2/asset_2/division_2/unit_2/signal_3",
#     "workspace_3/asset_3/division_3/unit_3/signal_4",
#     "workspace_3/asset_3/division_3/unit_3/signal_5",
# ]

derivers

Derivers

Client for interacting with the Nortech Derivers API.

Attributes:

• nortech_api NortechAPI - The Nortech API client.

Example:

# To define a deriver, you need to create a class that inherits from the Deriver class.
# The class must have two inner classes: Inputs and Outputs.
# The Inputs class must inherit from DeriverInputs and the Outputs class must inherit from DeriverOutputs.
# The Inputs class must define the inputs of the deriver.
# The Outputs class must define the outputs of the deriver.
# The run method must be defined and return a bytewax stream.

from __future__ import annotations

import bytewax.operators as op

from nortech.derivers import Deriver, DeriverInput, DeriverInputs, DeriverOutput, DeriverOutputs


class MyDeriver(Deriver):
    class Inputs(DeriverInputs):
        input_1: float | None = DeriverInput(
            workspace="workspace1", asset="asset1", division="division1", unit="unit1", signal="signal1"
        )
        input_2: float | None = DeriverInput(
            workspace="workspace2", asset="asset2", division="division2", unit="unit2", signal="signal2"
        )

    class Outputs(DeriverOutputs):
        output_1: float = DeriverOutput(
            workspace="workspace1",
            asset="asset1",
            division="division1",
            unit="unit1",
            signal="new_signal1",
            description="output_1",
            long_description="output_1_long_description",
            physical_unit="m/s",
        )
        output_2: str = DeriverOutput(
            workspace="workspace2",
            asset="asset2",
            division="division2",
            unit="unit2",
            signal="new_signal2",
            description="output_2",
            long_description="output_2_long_description",
            physical_unit="m/s",
        )

    def run(self, inputs: op.Stream[Inputs]) -> op.Stream[Outputs]:
        return op.map(
            "",
            inputs,
            lambda _input: self.Outputs(
                timestamp=_input.timestamp,
                output_1=_input.input_1 or 0,
                output_2=str(_input.input_2),
            ),
        )

list

def list(
    pagination_options: PaginationOptions[Literal["id", "name", "description"]]
    | None = None)

List derivers.

Arguments:

• pagination_options PaginationOptions, optional - The pagination options. Defaults to None.

Returns:

• PaginatedResponse[DeployedDeriver] - Paginated response of derivers.

Example:

from nortech import Nortech
from nortech.derivers import Deriver


# Define Deriver
class MyDeriver(Deriver): ...


nortech = Nortech()

derivers = nortech.derivers.list()
print(derivers)
# PaginatedResponse(
#     size=1,
#     next=None,
#     data=[
#         DeployedDeriverList(
#             deriver=MyDeriver,
#             description="my-description",
#             start_at=None,
#         )
#     ],
# )

get

def get(deriver: str | type[Deriver])

Get a deriver.

Arguments:

• deriver type[Deriver] - Deriver class to fetch or deriver class name.

Returns:

• DeployedDeriver - Deployed deriver.

Example:

from nortech import Nortech
from nortech.derivers import Deriver


# Define Deriver
class MyDeriver(Deriver): ...


nortech = Nortech()

# Get deriver by class or class name
derivers = nortech.derivers.get(MyDeriver)
derivers = nortech.derivers.get("MyDeriver")

print(derivers)
# DeployedDeriver(
#     deriver=MyDeriver,
#     description="my-description",
#     start_at="2025-01-01T12:00:00Z",
#     inputs=[
#         SignalOutput(
#             id=1,
#             name="input_1",
#             description="input_1",
#             long_description="input_1_long_description",
#             data_type="float",
#             physical_unit="m/s",
#             created_at="2025-01-01T12:00:00Z",
#             updated_at="2025-01-01T12:00:00Z",
#             workspace=MetadataOutput(
#                 id=1,
#                 name="workspace1",
#             ),
#             asset=MetadataOutput(
#                 id=1,
#                 name="asset1",
#             ),
#             division=MetadataOutput(
#                 id=1,
#                 name="division1",
#             ),
#             unit=MetadataOutput(
#                 id=1,
#                 name="unit1",
#             ),
#         ),
#     ],
#     outputs=[
#         SignalOutput(
#             id=2,
#             name="output_1",
#             description="output_1",
#             long_description="output_1_long_description",
#             data_type="float",
#             physical_unit="m/s",
#             created_at="2025-01-01T12:00:00Z",
#             updated_at="2025-01-01T12:00:00Z",
#             workspace=MetadataOutput(
#                 id=1,
#                 name="workspace1",
#             ),
#             asset=MetadataOutput(
#                 id=1,
#                 name="asset1",
#             ),
#             division=MetadataOutput(
#                 id=1,
#                 name="division1",
#             ),
#             unit=MetadataOutput(
#                 id=1,
#                 name="unit1",
#             ),
#         ),
#     ]
# )

create

def create(deriver: type[Deriver],
           start_at: datetime | None = None,
           description: str | None = None,
           create_parents: bool = False)

Create a deriver.

Arguments:

• deriver type[Deriver] - Deriver class to create.
• start_at datetime | None, optional - The start time for the deriver. Defaults to current time.
• description str | None, optional - The description for the deriver. Defaults to None.
• create_parents bool, optional - Whether to create parent entities. Defaults to False.

Returns:

• DeployedDeriver - Deployed deriver.

Example:

from datetime import datetime, timezone

from nortech import Nortech
from nortech.derivers import Deriver


# Define Deriver
class MyDeriver(Deriver): ...


nortech = Nortech()

derivers = nortech.derivers.create(MyDeriver, start_at=datetime.now(timezone.utc), description="my-description")
print(derivers)
# DeployedDeriver(
#     deriver=MyDeriver,
#     description="my-description",
#     start_at=None,
#     inputs=[
#         SignalOutput(
#             id=1,
#             name="input_1",
#             description="input_1",
#             long_description="input_1_long_description",
#             data_type="float",
#             physical_unit="m/s",
#             created_at="2025-01-01T12:00:00Z",
#             updated_at="2025-01-01T12:00:00Z",
#             workspace=MetadataOutput(
#                 id=1,
#                 name="workspace1",
#             ),
#             asset=MetadataOutput(
#                 id=1,
#                 name="asset1",
#             ),
#             division=MetadataOutput(
#                 id=1,
#                 name="division1",
#             ),
#             unit=MetadataOutput(
#                 id=1,
#                 name="unit1",
#             ),
#         ),
#     ],
#     outputs=[
#         SignalOutput(
#             id=2,
#             name="output_1",
#             description="output_1",
#             long_description="output_1_long_description",
#             data_type="float",
#             physical_unit="m/s",
#             created_at="2025-01-01T12:00:00Z",
#             updated_at="2025-01-01T12:00:00Z",
#             workspace=MetadataOutput(
#                 id=1,
#                 name="workspace1",
#             ),
#             asset=MetadataOutput(
#                 id=1,
#                 name="asset1",
#             ),
#             division=MetadataOutput(
#                 id=1,
#                 name="division1",
#             ),
#             unit=MetadataOutput(
#                 id=1,
#                 name="unit1",
#             ),
#         ),
#     ]
# )

update

def update(deriver: type[Deriver],
           start_at: datetime | None = None,
           description: str | None = None,
           create_parents: bool = False,
           keep_data: bool = False)

Update a deriver.

Arguments:

• deriver type[Deriver] - Deriver class to update.
• start_at datetime | None, optional - The start time for the deriver. Defaults to current time.
• description str | None, optional - The description for the deriver. Defaults to None.
• create_parents bool, optional - Whether to create parent workspaces. Defaults to False.
• keep_data bool, optional - Whether to keep the data. Defaults to False.

Returns:

• DeployedDeriver - Deployed deriver.

Example:

from datetime import datetime, timezone

from nortech import Nortech
from nortech.derivers import Deriver


# Define Deriver
class MyDeriver(Deriver): ...


nortech = Nortech()

derivers = nortech.derivers.update(MyDeriver, start_at=datetime.now(timezone.utc), description="my-description")
print(derivers)
# DeployedDeriver(
#     deriver=MyDeriver,
#     description="my-description",
#     start_at=None,
#     inputs=[
#         SignalOutput(
#             id=1,
#             name="input_1",
#             description="input_1",
#             long_description="input_1_long_description",
#             data_type="float",
#             physical_unit="m/s",
#             created_at="2025-01-01T12:00:00Z",
#             updated_at="2025-01-01T12:00:00Z",
#             workspace=MetadataOutput(
#                 id=1,
#                 name="workspace1",
#             ),
#             asset=MetadataOutput(
#                 id=1,
#                 name="asset1",
#             ),
#             division=MetadataOutput(
#                 id=1,
#                 name="division1",
#             ),
#             unit=MetadataOutput(
#                 id=1,
#                 name="unit1",
#             ),
#         ),
#     ],
#     outputs=[
#         SignalOutput(
#             id=2,
#             name="output_1",
#             description="output_1",
#             long_description="output_1_long_description",
#             data_type="float",
#             physical_unit="m/s",
#             created_at="2025-01-01T12:00:00Z",
#             updated_at="2025-01-01T12:00:00Z",
#             workspace=MetadataOutput(
#                 id=1,
#                 name="workspace1",
#             ),
#             asset=MetadataOutput(
#                 id=1,
#                 name="asset1",
#             ),
#             division=MetadataOutput(
#                 id=1,
#                 name="division1",
#             ),
#             unit=MetadataOutput(
#                 id=1,
#                 name="unit1",
#             ),
#         ),
#     ]
# )

run_locally_with_df

def run_locally_with_df(deriver: type[Deriver],
                        df: DataFrame,
                        batch_size: int = 10000) -> DataFrame

Run a deriver locally on a DataFrame. The dataframe must have a timestamp index and columns equal to the input names in the deriver definition.

Arguments:

• deriver Deriver - The deriver to run.
• batch_size int, optional - The batch size for processing. Defaults to 10000.
• df DataFrame - The input DataFrame.

Returns:

• DataFrame - The processed DataFrame with derived signals.

Example:

from datetime import timezone

import pandas as pd

from nortech import Nortech
from nortech.derivers import Deriver


class MyDeriver(Deriver): ...


nortech = Nortech()

# Create input DataFrame
df = pd.DataFrame(
    {
        "timestamp": pd.date_range(start="2023-01-01", periods=100, freq="s", tz=timezone.utc),
        "input_signal": [float(i) for i in range(100)],
    }
).set_index("timestamp")

# Run the deriver locally
result_df = nortech.derivers.run_locally_with_df(MyDeriver, df, batch_size=5000)

print(result_df)
#                            output_signal
# timestamp
# 2023-01-01 00:00:00+00:00            0.0
# 2023-01-01 00:00:01+00:00            2.0
# 2023-01-01 00:00:02+00:00            4.0
# 2023-01-01 00:00:03+00:00            6.0
# 2023-01-01 00:00:04+00:00            8.0
# ...                                  ...
# 2023-01-01 00:01:35+00:00          190.0
# 2023-01-01 00:01:36+00:00          192.0
# 2023-01-01 00:01:37+00:00          194.0
# 2023-01-01 00:01:38+00:00          196.0
# 2023-01-01 00:01:39+00:00          198.0

run_locally_with_source_data

def run_locally_with_source_data(deriver: type[Deriver],
                                 time_window: TimeWindow,
                                 batch_size: int = 10000) -> DataFrame

Run a deriver locally by fetching its inputs signal data for a given time window.

Arguments:

• deriver Deriver - The deriver to run.
• time_window TimeWindow - The time window to process.
• batch_size int, optional - The batch size for processing. Defaults to 10000.

Returns:

• DataFrame - The processed DataFrame with derived signals.

Example:

from datetime import datetime, timezone

import pandas as pd

from nortech import Nortech
from nortech.derivers import Deriver, TimeWindow


class MyDeriver(Deriver): ...


nortech = Nortech()

# Create input DataFrame or use nortech.datatools to get data
df = pd.DataFrame(
    {
        "timestamp": pd.date_range(start="2023-01-01", periods=100, freq="s", tz=timezone.utc),
        "input_signal": [float(i) for i in range(100)],
    }
).set_index("timestamp")

# Run the deriver locally
result_df = nortech.derivers.run_locally_with_source_data(
    MyDeriver, time_window=TimeWindow(start=datetime.now(timezone.utc), end=datetime.now(timezone.utc))
)

print(result_df)
#                            output_signal
# timestamp
# 2023-01-01 00:00:00+00:00            0.0
# 2023-01-01 00:00:01+00:00            2.0
# 2023-01-01 00:00:02+00:00            4.0
# 2023-01-01 00:00:03+00:00            6.0
# 2023-01-01 00:00:04+00:00            8.0
# ...                                  ...
# 2023-01-01 00:01:35+00:00          190.0
# 2023-01-01 00:01:36+00:00          192.0
# 2023-01-01 00:01:37+00:00          194.0
# 2023-01-01 00:01:38+00:00          196.0
# 2023-01-01 00:01:39+00:00          198.0

metadata.values.time_window

TimeWindow

Time window model.

Attributes:

• start datetime - Start time.
• end datetime - End time.

metadata.values.pagination

PaginationOptions

Pagination options for list endpoints.

Attributes:

• size int | None, default=100, le=100 - The number of items to return.
• sort_by str | None - The field to sort by.
• sort_order "asc" | "desc", default="asc" - The order to sort by.
• next_token str | None - The next token to use for pagination.

PaginatedResponse

Paginated response from list endpoints.

Attributes:

• size int - The number of items returned.
• data list[obj] - The list of items.
• next.token str | None - The next token to use for pagination. If None, there are no more pages.

metadata.values.workspace

Module containing all schemas related with Workspaces.

WorkspaceInputDict

Dictionary representation of Workspace input data.

Attributes:

• workspace str - The name of the Workspace.

WorkspaceInput

Pydantic model for Workspace input data.

Attributes:

• workspace str - The name of the Workspace.

WorkspaceListOutput

Output model for workspace list entries.

Attributes:

• id int - Id of the Workspace.
• name str - Name of the Workspace.
• description str - A description of the Workspace.

WorkspaceOutput

Detailed output model for a single workspace.

Attributes:

• id int - Id of the Workspace.
• name str - Name of the Workspace.
• description str - A description of the Workspace.
• created_at datetime - Timestamp of when the Workspace was created.
• updated_at datetime - Timestamp of when the Workspace was last updated.

metadata.values.asset

Module containing all schemas related with Assets.

AssetInputDict

Dictionary representation of Asset input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.

AssetInput

Pydantic model for asset input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.

AssetListOutput

Output model for asset list entries.

Attributes:

• id int - Id of the Asset.
• name str - Name of the Asset.
• description str - A description of the Asset.

AssetOutput

Detailed output model for a single asset.

Attributes:

• id int - Id of the Asset.
• name str - Name of the Asset.
• description str - A description of the Asset.
• created_at datetime - Timestamp of when the Asset was created.
• updated_at datetime - Timestamp of when the Asset was last updated.
• workspace - Metadata about the Workspace containing the Asset.
  • id (int): Id of the Workspace.
  • name (str): Name of the Workspace.

metadata.values.division

Module containing all schemas related with Divisions.

DivisionInputDict

Dictionary representation of Division input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.

DivisionInput

Pydantic model for Division input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.

DivisionListOutput

Output model for division list entries.

Attributes:

• id int - Id of the Division.
• name str - Name of the Division.
• description str - A description of the division.

DivisionOutput

Detailed output model for a single division.

Attributes:

• id int - Id of the Division.
• name str - Name of the Division.
• description str - A description of the division.
• created_at datetime - Timestamp of when the Division was created.
• updated_at datetime - Timestamp of when the Division was last updated.
• workspace - Metadata about the Workspace containing the Division.
  • id (int): Id of the Workspace.
  • name (str): Name of the Workspace.
• asset - Metadata about the Asset containing the Division.
  • id (int): Id of the Asset.
  • name (str): Name of the Asset.

metadata.values.unit

Module containing all schemas related with Units.

UnitInputDict

Dictionary representation of Unit input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.
• unit str - The name of the Unit.

UnitInput

Pydantic model for Unit input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.
• unit str - The name of the Unit.

UnitListOutput

Output model for unit list entries.

Attributes:

• id int - Id of the Unit.
• name str - Name of the Unit.

UnitDivision

Output model for unit division entries.

Attributes:

• id int - Id of the Division.
• name str - Name of the Division.

UnitOutput

Detailed output model for a single unit.

Attributes:

• id int - Id of the Unit.
• name str - Name of the Unit.
• created_at datetime - Timestamp of when the Unit was created.
• updated_at datetime - Timestamp of when the Unit was last updated.
• workspace - Metadata about the Workspace containing the Unit.
  • id (int): Id of the Workspace.
  • name (str): Name of the Workspace.
• asset - Metadata about the Asset containing the Unit.
  • id (int): Id of the Asset.
  • name (str): Name of the Asset.
• division - Metadata about the Division containing the Unit.
  • id (int): Id of the Division.
  • name (str): Name of the Division.

metadata.values.signal

Module containing all schemas related with Signals.

SignalInputDict

Dictionary representation of Signal input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.
• unit str - The name of the Unit.
• signal str - The name of the Signal.

SignalInput

Pydantic model for Signal input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.
• unit str - The name of the Unit.
• signal str - The name of the Signal.

SignalListOutput

Output model for signal list entries.

Attributes:

• id int - Id of the Signal.
• name str - Name of the Signal.
• physical_unit str - The physical unit of the Signal.
• data_type Literal["float", "boolean", "string", "json"] - The data type of the Signal.
• description str - A description of the Signal.
• long_description str - A long description of the Signal.