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.
• device Device - Client for interacting with the Nortech Metadata Device 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.core.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]

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.gateways.nortech_api 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.core.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]

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.gateways.nortech_api import PaginationOptions
from nortech.core.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.core.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]

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.gateways.nortech_api import PaginationOptions
from nortech.core.values.asset import AssetInput

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]

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.gateways.nortech_api 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.core.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]

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.gateways.nortech_api import PaginationOptions
from nortech.core.values.division import DivisionInput

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]

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.gateways.nortech_api 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]

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.gateways.nortech_api 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"
#         )
#     ]
# )

Device

get

def get(
    device: int | DeviceInputDict | DeviceInput | DeviceOutput
    | DeviceListOutput
) -> DeviceOutput

Get a device by ID or input.

Arguments:

• device int | DeviceInputDict | DeviceInput | DeviceOutput | DeviceListOutput - The device identifier, which can be:
  • int: The device "ID".
  • DeviceInputDict: A dictionary representation of a device input.
  • DeviceInput: A pydantic model representing a device input.
  • DeviceOutput: A pydantic model representing a device output. Obtained from requesting a device metadata.
  • DeviceListOutput: A pydantic model representing a listed device output. Obtained from requesting devices metadata.

Returns:

• DeviceOutput - The device details.

Example:

from nortech import Nortech
from nortech.core.values.device import DeviceInput

nortech = Nortech()

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

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

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

print(device)
# DeviceOutput(
#     id=123,
#     name="my-device",
#     type="my-type",
#     onboarded=True,
#     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", "type",
                                                  "onboarded"]] | None = None
) -> PaginatedResponse[DeviceListOutput]

List all devices 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[DeviceListOutput] - A paginated list of devices.

Example:

from nortech import Nortech
from nortech.gateways.nortech_api import PaginationOptions
from nortech.core.values.division import DivisionInput

nortech = Nortech()

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

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

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

print(devices)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         DeviceListOutput(
#             id=1,
#             name="my-device",
#             type="my-type",
#             onboarded=True
#         ),
#         DeviceListOutput(
#             id=2,
#             name="another-device",
#             type="another-type",
#             onboarded=False
#         )
#     ]
# )

list_by_workspace_id

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

List all devices in a workspace.

Arguments:

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

Returns:

• PaginatedResponse[DeviceListOutput] - A paginated list of devices.

Example:

from nortech import Nortech
from nortech.gateways.nortech_api import PaginationOptions

nortech = Nortech()

# List all devices in a workspace
devices = nortech.metadata.device.list_by_workspace_id(123)

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

print(devices)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         DeviceListOutput(
#             id=1,
#             name="my-device",
#             type="my-type",
#             onboarded=True
#         ),
#         DeviceListOutput(
#             id=2,
#             name="another-device",
#             type="another-type",
#             onboarded=False
#         )
#     ]
# )

list_by_asset_id

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

List all devices in an asset.

Arguments:

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

Returns:

• PaginatedResponse[DeviceListOutput] - A paginated list of devices.

Example:

from nortech import Nortech
from nortech.gateways.nortech_api import PaginationOptions

nortech = Nortech()

# List all devices in an asset
devices = nortech.metadata.device.list_by_asset_id(123)

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

print(devices)
# PaginatedResponse(
#     size=2,
#     next=None,
#     data=[
#         DeviceListOutput(
#             id=1,
#             name="my-device",
#             type="my-type",
#             onboarded=True
#         ),
#         DeviceListOutput(
#             id=2,
#             name="another-device",
#             type="another-type",
#             onboarded=False
#         )
#     ]
# )

Signal

get

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

Get a signal by ID or input.

Arguments:

• signal int | SignalInputDict | SignalInput | SignalOutput | SignalListOutput | SignalDeviceInputDict | SignalDeviceInput - 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.
  • SignalDeviceInputDict: A dictionary representation of a device signal input.
  • SignalDeviceInput: A pydantic model representing a device signal input.

Returns:

• SignalOutput - The signal details.

Example:

from nortech import Nortech
from nortech.core.values.signal import SignalDeviceInput, 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 device signal by input dict
signal = nortech.metadata.signal.get(
    {
        "workspace": "my-workspace",
        "asset": "my-asset",
        "division": "my-division",
        "device": "my-device",
        "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"
    )
)

# Get by SignalDeviceInput pydantic object
signal = nortech.metadata.signal.get(
SignalDeviceInput(
        workspace="my-workspace", asset="my-asset", division="my-division", device="my-device", 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"
#     )
#     device=MetadataOutput(
#         id=101,
#         name="my-device"
#     )
# )

list

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

List all signals in a unit or device.

Arguments:

• unit_or_device int | UnitInputDict | UnitInput | UnitOutput | DeviceInputDict | DeviceInput | DeviceOutput - The unit or device identifier, which can be:
  • int: The unit/device "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.
  • DeviceInputDict: A dictionary representation of a device input.
  • DeviceInput: A pydantic model representing a device input.
  • DeviceOutput: A pydantic model representing a device output.
• pagination_options PaginationOptions, optional - Pagination settings.

Returns:

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

Example:

from nortech import Nortech
from nortech.gateways.nortech_api import PaginationOptions
from nortech.core.values.device import DeviceInput
from nortech.core.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"),
)

# List device signals with pagination
signals = nortech.metadata.signal.list(
    {"workspace": "my-workspace", "asset": "my-asset", "division": "my-division", "device": "my-device"},
    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")
)

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

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]

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.gateways.nortech_api 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]

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.gateways.nortech_api 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]

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.gateways.nortech_api 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.core.values.signal import SignalInput, SignalInputDict
from nortech.datatools.values.windowing import TimeWindow

# 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.core.values.signal import SignalInput, SignalInputDict
from nortech.datatools.values.windowing import TimeWindow

# 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.core.values.signal import SignalInput, SignalInputDict
from nortech.datatools.values.windowing import TimeWindow

# 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.core.values.signal import SignalInput, SignalInputDict
from nortech.datatools.values.windowing import TimeWindow

# 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.

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:

# 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:

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

nortech = Nortech()
derivers = nortech.derivers.get(MyDeriver)
print(derivers)
# DeployedDeriver(
#     deriver=MyDeriver,
#     description="my-description",
#     start_at="2025-01-01T12:00:00Z",
#     inputs=[
#         SignalOutputNoDevice(
#             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=[
#         SignalOutputNoDevice(
#             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:

# 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=[
#         SignalOutputNoDevice(
#             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=[
#         SignalOutputNoDevice(
#             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:

# 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=[
#         SignalOutputNoDevice(
#             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=[
#         SignalOutputNoDevice(
#             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 datetime, timezone

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

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

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.device

Module containing all schemas related with Devices.

DeviceInputDict

Dictionary representation of Device input data.

Attributes:

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

DeviceInput

Pydantic model for Device input data.

Attributes:

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

DeviceListOutput

Output model for device list entries.

Attributes:

• id int - Id of the Device.
• name str - Name of the Device.
• type str - The type of the Device.
• onboarded bool - Whether the Device is onboarded.

DeviceOutput

Detailed output model for a single device.

Attributes:

• id int - Id of the Device.
• name str - Name of the Device.
• type str - The type of the Device.
• onboarded bool - Whether the Device is onboarded.
• created_at datetime - Timestamp of when the Device was created.
• updated_at datetime - Timestamp of when the Device was last updated.
• workspace - Metadata about the Workspace containing the Device.
  • id (int): Id of the Workspace.
  • name (str): Name of the Workspace.
• asset - Metadata about the Asset containing the Device.
  • id (int): Id of the Asset.
  • name (str): Name of the Asset.
• division - Metadata about the Division containing the Device.
  • 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.

SignalDeviceInputDict

Dictionary representation of SignalDevice input data.

Attributes:

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

SignalDeviceInput

Pydantic model for SignalDevice input data.

Attributes:

• workspace str - The name of the Workspace.
• asset str - The name of the Asset.
• division str - The name of the Division.
• device str - The name of the Device.
• 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.

SignalOutput

Detailed output model for a single signal.

Attributes:

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