Windmill Data Classes

examples.windmill.data_classes

Blade

Bases: DomainModel

This represents the reading version of blade.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the blade.	required
data_record		The data record of the blade node.	required
is_damaged		The is damaged field.	required
name		The name field.	required
sensor_positions		The sensor position field.	required

Source code in examples/windmill/data_classes/_blade.py

class Blade(DomainModel):
    """This represents the reading version of blade.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the blade.
        data_record: The data record of the blade node.
        is_damaged: The is damaged field.
        name: The name field.
        sensor_positions: The sensor position field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Blade", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    is_damaged: Optional[bool] = None
    name: Optional[str] = None
    sensor_positions: Optional[list[Union[SensorPosition, str, dm.NodeId]]] = Field(default=None, repr=False)

    def as_write(self) -> BladeWrite:
        """Convert this read version of blade to the writing version."""
        return BladeWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            is_damaged=self.is_damaged,
            name=self.name,
            sensor_positions=[
                sensor_position.as_write() if isinstance(sensor_position, DomainModel) else sensor_position
                for sensor_position in self.sensor_positions or []
            ],
        )

    def as_apply(self) -> BladeWrite:
        """Convert this read version of blade to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, Blade],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._sensor_position import SensorPosition

        for instance in instances.values():
            if edges := edges_by_source_node.get(instance.as_id()):
                sensor_positions: list[SensorPosition | str | dm.NodeId] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("power-models", "Blade.sensor_positions") and isinstance(
                        value, (SensorPosition, str, dm.NodeId)
                    ):
                        sensor_positions.append(value)

                instance.sensor_positions = sensor_positions or None

as_apply()

Convert this read version of blade to the writing version.

Source code in examples/windmill/data_classes/_blade.py

def as_apply(self) -> BladeWrite:
    """Convert this read version of blade to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of blade to the writing version.

Source code in examples/windmill/data_classes/_blade.py

def as_write(self) -> BladeWrite:
    """Convert this read version of blade to the writing version."""
    return BladeWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        is_damaged=self.is_damaged,
        name=self.name,
        sensor_positions=[
            sensor_position.as_write() if isinstance(sensor_position, DomainModel) else sensor_position
            for sensor_position in self.sensor_positions or []
        ],
    )

BladeGraphQL

Bases: GraphQLCore

This represents the reading version of blade, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the blade.	required
data_record		The data record of the blade node.	required
is_damaged		The is damaged field.	required
name		The name field.	required
sensor_positions		The sensor position field.	required

Source code in examples/windmill/data_classes/_blade.py

class BladeGraphQL(GraphQLCore):
    """This represents the reading version of blade, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the blade.
        data_record: The data record of the blade node.
        is_damaged: The is damaged field.
        name: The name field.
        sensor_positions: The sensor position field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Blade", "1")
    is_damaged: Optional[bool] = None
    name: Optional[str] = None
    sensor_positions: Optional[list[SensorPositionGraphQL]] = Field(default=None, repr=False)

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("sensor_positions", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Blade:
        """Convert this GraphQL format of blade to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Blade(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            is_damaged=self.is_damaged,
            name=self.name,
            sensor_positions=[sensor_position.as_read() for sensor_position in self.sensor_positions or []],
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> BladeWrite:
        """Convert this GraphQL format of blade to the writing format."""
        return BladeWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            is_damaged=self.is_damaged,
            name=self.name,
            sensor_positions=[sensor_position.as_write() for sensor_position in self.sensor_positions or []],
        )

as_read()

Convert this GraphQL format of blade to the reading format.

Source code in examples/windmill/data_classes/_blade.py

@no_type_check
def as_read(self) -> Blade:
    """Convert this GraphQL format of blade to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Blade(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        is_damaged=self.is_damaged,
        name=self.name,
        sensor_positions=[sensor_position.as_read() for sensor_position in self.sensor_positions or []],
    )

as_write()

Convert this GraphQL format of blade to the writing format.

Source code in examples/windmill/data_classes/_blade.py

@no_type_check
def as_write(self) -> BladeWrite:
    """Convert this GraphQL format of blade to the writing format."""
    return BladeWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        is_damaged=self.is_damaged,
        name=self.name,
        sensor_positions=[sensor_position.as_write() for sensor_position in self.sensor_positions or []],
    )

BladeList

Bases: DomainModelList[Blade]

List of blades in the read version.

Source code in examples/windmill/data_classes/_blade.py

class BladeList(DomainModelList[Blade]):
    """List of blades in the read version."""

    _INSTANCE = Blade

    def as_write(self) -> BladeWriteList:
        """Convert these read versions of blade to the writing versions."""
        return BladeWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> BladeWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_blade.py

def as_apply(self) -> BladeWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of blade to the writing versions.

Source code in examples/windmill/data_classes/_blade.py

def as_write(self) -> BladeWriteList:
    """Convert these read versions of blade to the writing versions."""
    return BladeWriteList([node.as_write() for node in self.data])

BladeWrite

Bases: DomainModelWrite

This represents the writing version of blade.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the blade.	required
data_record		The data record of the blade node.	required
is_damaged		The is damaged field.	required
name		The name field.	required
sensor_positions		The sensor position field.	required

Source code in examples/windmill/data_classes/_blade.py

class BladeWrite(DomainModelWrite):
    """This represents the writing version of blade.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the blade.
        data_record: The data record of the blade node.
        is_damaged: The is damaged field.
        name: The name field.
        sensor_positions: The sensor position field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Blade", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    is_damaged: Optional[bool] = None
    name: Optional[str] = None
    sensor_positions: Optional[list[Union[SensorPositionWrite, str, dm.NodeId]]] = Field(default=None, repr=False)

    @field_validator("sensor_positions", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.is_damaged is not None or write_none:
            properties["is_damaged"] = self.is_damaged

        if self.name is not None or write_none:
            properties["name"] = self.name

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        edge_type = dm.DirectRelationReference("power-models", "Blade.sensor_positions")
        for sensor_position in self.sensor_positions or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=sensor_position,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        return resources

BladeWriteList

Bases: DomainModelWriteList[BladeWrite]

List of blades in the writing version.

Source code in examples/windmill/data_classes/_blade.py

class BladeWriteList(DomainModelWriteList[BladeWrite]):
    """List of blades in the writing version."""

    _INSTANCE = BladeWrite

DataRecord

Bases: BaseModel

The data record represents the metadata of a node.

Parameters:

Name	Type	Description	Default
created_time		The created time of the node.	required
last_updated_time		The last updated time of the node.	required
deleted_time		If present, the deleted time of the node.	required
version		The version of the node.	required

Source code in examples/windmill/data_classes/_core/base.py

class DataRecord(BaseModel):
    """The data record represents the metadata of a node.

    Args:
        created_time: The created time of the node.
        last_updated_time: The last updated time of the node.
        deleted_time: If present, the deleted time of the node.
        version: The version of the node.
    """

    version: int
    last_updated_time: datetime.datetime
    created_time: datetime.datetime
    deleted_time: Optional[datetime.datetime] = None

DataRecordWrite

Bases: BaseModel

The data record represents the metadata of a node.

Parameters:

Name	Type	Description	Default
existing_version		Fail the ingestion request if the node version is greater than or equal to this value. If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance). If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists. If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.	required

Source code in examples/windmill/data_classes/_core/base.py

class DataRecordWrite(BaseModel):
    """The data record represents the metadata of a node.

    Args:
        existing_version: Fail the ingestion request if the node version is greater than or equal to this value.
            If no existingVersion is specified, the ingestion will always overwrite any existing data for the edge (for the specified container or instance).
            If existingVersion is set to 0, the upsert will behave as an insert, so it will fail the bulk if the item already exists.
            If skipOnVersionConflict is set on the ingestion request, then the item will be skipped instead of failing the ingestion request.
    """

    existing_version: Optional[int] = None

Gearbox

Bases: DomainModel

This represents the reading version of gearbox.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the gearbox.	required
data_record		The data record of the gearbox node.	required
displacement_x		The displacement x field.	required
displacement_y		The displacement y field.	required
displacement_z		The displacement z field.	required

Source code in examples/windmill/data_classes/_gearbox.py

class Gearbox(DomainModel):
    """This represents the reading version of gearbox.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the gearbox.
        data_record: The data record of the gearbox node.
        displacement_x: The displacement x field.
        displacement_y: The displacement y field.
        displacement_z: The displacement z field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Gearbox", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    displacement_x: Union[TimeSeries, str, None] = None
    displacement_y: Union[TimeSeries, str, None] = None
    displacement_z: Union[TimeSeries, str, None] = None

    def as_write(self) -> GearboxWrite:
        """Convert this read version of gearbox to the writing version."""
        return GearboxWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            displacement_x=(
                self.displacement_x.as_write()
                if isinstance(self.displacement_x, CogniteTimeSeries)
                else self.displacement_x
            ),
            displacement_y=(
                self.displacement_y.as_write()
                if isinstance(self.displacement_y, CogniteTimeSeries)
                else self.displacement_y
            ),
            displacement_z=(
                self.displacement_z.as_write()
                if isinstance(self.displacement_z, CogniteTimeSeries)
                else self.displacement_z
            ),
        )

    def as_apply(self) -> GearboxWrite:
        """Convert this read version of gearbox to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of gearbox to the writing version.

Source code in examples/windmill/data_classes/_gearbox.py

def as_apply(self) -> GearboxWrite:
    """Convert this read version of gearbox to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of gearbox to the writing version.

Source code in examples/windmill/data_classes/_gearbox.py

def as_write(self) -> GearboxWrite:
    """Convert this read version of gearbox to the writing version."""
    return GearboxWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        displacement_x=(
            self.displacement_x.as_write()
            if isinstance(self.displacement_x, CogniteTimeSeries)
            else self.displacement_x
        ),
        displacement_y=(
            self.displacement_y.as_write()
            if isinstance(self.displacement_y, CogniteTimeSeries)
            else self.displacement_y
        ),
        displacement_z=(
            self.displacement_z.as_write()
            if isinstance(self.displacement_z, CogniteTimeSeries)
            else self.displacement_z
        ),
    )

GearboxGraphQL

Bases: GraphQLCore

This represents the reading version of gearbox, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the gearbox.	required
data_record		The data record of the gearbox node.	required
displacement_x		The displacement x field.	required
displacement_y		The displacement y field.	required
displacement_z		The displacement z field.	required

Source code in examples/windmill/data_classes/_gearbox.py

class GearboxGraphQL(GraphQLCore):
    """This represents the reading version of gearbox, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the gearbox.
        data_record: The data record of the gearbox node.
        displacement_x: The displacement x field.
        displacement_y: The displacement y field.
        displacement_z: The displacement z field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Gearbox", "1")
    displacement_x: Optional[TimeSeriesGraphQL] = None
    displacement_y: Optional[TimeSeriesGraphQL] = None
    displacement_z: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Gearbox:
        """Convert this GraphQL format of gearbox to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Gearbox(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            displacement_x=self.displacement_x.as_read() if self.displacement_x else None,
            displacement_y=self.displacement_y.as_read() if self.displacement_y else None,
            displacement_z=self.displacement_z.as_read() if self.displacement_z else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> GearboxWrite:
        """Convert this GraphQL format of gearbox to the writing format."""
        return GearboxWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            displacement_x=self.displacement_x.as_write() if self.displacement_x else None,
            displacement_y=self.displacement_y.as_write() if self.displacement_y else None,
            displacement_z=self.displacement_z.as_write() if self.displacement_z else None,
        )

as_read()

Convert this GraphQL format of gearbox to the reading format.

Source code in examples/windmill/data_classes/_gearbox.py

@no_type_check
def as_read(self) -> Gearbox:
    """Convert this GraphQL format of gearbox to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Gearbox(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        displacement_x=self.displacement_x.as_read() if self.displacement_x else None,
        displacement_y=self.displacement_y.as_read() if self.displacement_y else None,
        displacement_z=self.displacement_z.as_read() if self.displacement_z else None,
    )

as_write()

Convert this GraphQL format of gearbox to the writing format.

Source code in examples/windmill/data_classes/_gearbox.py

@no_type_check
def as_write(self) -> GearboxWrite:
    """Convert this GraphQL format of gearbox to the writing format."""
    return GearboxWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        displacement_x=self.displacement_x.as_write() if self.displacement_x else None,
        displacement_y=self.displacement_y.as_write() if self.displacement_y else None,
        displacement_z=self.displacement_z.as_write() if self.displacement_z else None,
    )

GearboxList

Bases: DomainModelList[Gearbox]

List of gearboxes in the read version.

Source code in examples/windmill/data_classes/_gearbox.py

class GearboxList(DomainModelList[Gearbox]):
    """List of gearboxes in the read version."""

    _INSTANCE = Gearbox

    def as_write(self) -> GearboxWriteList:
        """Convert these read versions of gearbox to the writing versions."""
        return GearboxWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> GearboxWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_gearbox.py

def as_apply(self) -> GearboxWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of gearbox to the writing versions.

Source code in examples/windmill/data_classes/_gearbox.py

def as_write(self) -> GearboxWriteList:
    """Convert these read versions of gearbox to the writing versions."""
    return GearboxWriteList([node.as_write() for node in self.data])

GearboxWrite

Bases: DomainModelWrite

This represents the writing version of gearbox.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the gearbox.	required
data_record		The data record of the gearbox node.	required
displacement_x		The displacement x field.	required
displacement_y		The displacement y field.	required
displacement_z		The displacement z field.	required

Source code in examples/windmill/data_classes/_gearbox.py

class GearboxWrite(DomainModelWrite):
    """This represents the writing version of gearbox.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the gearbox.
        data_record: The data record of the gearbox node.
        displacement_x: The displacement x field.
        displacement_y: The displacement y field.
        displacement_z: The displacement z field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Gearbox", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    displacement_x: Union[TimeSeriesWrite, str, None] = None
    displacement_y: Union[TimeSeriesWrite, str, None] = None
    displacement_z: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.displacement_x is not None or write_none:
            properties["displacement_x"] = (
                self.displacement_x
                if isinstance(self.displacement_x, str) or self.displacement_x is None
                else self.displacement_x.external_id
            )

        if self.displacement_y is not None or write_none:
            properties["displacement_y"] = (
                self.displacement_y
                if isinstance(self.displacement_y, str) or self.displacement_y is None
                else self.displacement_y.external_id
            )

        if self.displacement_z is not None or write_none:
            properties["displacement_z"] = (
                self.displacement_z
                if isinstance(self.displacement_z, str) or self.displacement_z is None
                else self.displacement_z.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.displacement_x, CogniteTimeSeriesWrite):
            resources.time_series.append(self.displacement_x)

        if isinstance(self.displacement_y, CogniteTimeSeriesWrite):
            resources.time_series.append(self.displacement_y)

        if isinstance(self.displacement_z, CogniteTimeSeriesWrite):
            resources.time_series.append(self.displacement_z)

        return resources

GearboxWriteList

Bases: DomainModelWriteList[GearboxWrite]

List of gearboxes in the writing version.

Source code in examples/windmill/data_classes/_gearbox.py

class GearboxWriteList(DomainModelWriteList[GearboxWrite]):
    """List of gearboxes in the writing version."""

    _INSTANCE = GearboxWrite

Generator

Bases: DomainModel

This represents the reading version of generator.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the generator.	required
data_record		The data record of the generator node.	required
generator_speed_controller		The generator speed controller field.	required
generator_speed_controller_reference		The generator speed controller reference field.	required

Source code in examples/windmill/data_classes/_generator.py

class Generator(DomainModel):
    """This represents the reading version of generator.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the generator.
        data_record: The data record of the generator node.
        generator_speed_controller: The generator speed controller field.
        generator_speed_controller_reference: The generator speed controller reference field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Generator", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    generator_speed_controller: Union[TimeSeries, str, None] = None
    generator_speed_controller_reference: Union[TimeSeries, str, None] = None

    def as_write(self) -> GeneratorWrite:
        """Convert this read version of generator to the writing version."""
        return GeneratorWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            generator_speed_controller=(
                self.generator_speed_controller.as_write()
                if isinstance(self.generator_speed_controller, CogniteTimeSeries)
                else self.generator_speed_controller
            ),
            generator_speed_controller_reference=(
                self.generator_speed_controller_reference.as_write()
                if isinstance(self.generator_speed_controller_reference, CogniteTimeSeries)
                else self.generator_speed_controller_reference
            ),
        )

    def as_apply(self) -> GeneratorWrite:
        """Convert this read version of generator to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of generator to the writing version.

Source code in examples/windmill/data_classes/_generator.py

def as_apply(self) -> GeneratorWrite:
    """Convert this read version of generator to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of generator to the writing version.

Source code in examples/windmill/data_classes/_generator.py

def as_write(self) -> GeneratorWrite:
    """Convert this read version of generator to the writing version."""
    return GeneratorWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        generator_speed_controller=(
            self.generator_speed_controller.as_write()
            if isinstance(self.generator_speed_controller, CogniteTimeSeries)
            else self.generator_speed_controller
        ),
        generator_speed_controller_reference=(
            self.generator_speed_controller_reference.as_write()
            if isinstance(self.generator_speed_controller_reference, CogniteTimeSeries)
            else self.generator_speed_controller_reference
        ),
    )

GeneratorGraphQL

Bases: GraphQLCore

This represents the reading version of generator, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the generator.	required
data_record		The data record of the generator node.	required
generator_speed_controller		The generator speed controller field.	required
generator_speed_controller_reference		The generator speed controller reference field.	required

Source code in examples/windmill/data_classes/_generator.py

class GeneratorGraphQL(GraphQLCore):
    """This represents the reading version of generator, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the generator.
        data_record: The data record of the generator node.
        generator_speed_controller: The generator speed controller field.
        generator_speed_controller_reference: The generator speed controller reference field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Generator", "1")
    generator_speed_controller: Optional[TimeSeriesGraphQL] = None
    generator_speed_controller_reference: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Generator:
        """Convert this GraphQL format of generator to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Generator(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            generator_speed_controller=(
                self.generator_speed_controller.as_read() if self.generator_speed_controller else None
            ),
            generator_speed_controller_reference=(
                self.generator_speed_controller_reference.as_read()
                if self.generator_speed_controller_reference
                else None
            ),
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> GeneratorWrite:
        """Convert this GraphQL format of generator to the writing format."""
        return GeneratorWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            generator_speed_controller=(
                self.generator_speed_controller.as_write() if self.generator_speed_controller else None
            ),
            generator_speed_controller_reference=(
                self.generator_speed_controller_reference.as_write()
                if self.generator_speed_controller_reference
                else None
            ),
        )

as_read()

Convert this GraphQL format of generator to the reading format.

Source code in examples/windmill/data_classes/_generator.py

@no_type_check
def as_read(self) -> Generator:
    """Convert this GraphQL format of generator to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Generator(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        generator_speed_controller=(
            self.generator_speed_controller.as_read() if self.generator_speed_controller else None
        ),
        generator_speed_controller_reference=(
            self.generator_speed_controller_reference.as_read()
            if self.generator_speed_controller_reference
            else None
        ),
    )

as_write()

Convert this GraphQL format of generator to the writing format.

Source code in examples/windmill/data_classes/_generator.py

@no_type_check
def as_write(self) -> GeneratorWrite:
    """Convert this GraphQL format of generator to the writing format."""
    return GeneratorWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        generator_speed_controller=(
            self.generator_speed_controller.as_write() if self.generator_speed_controller else None
        ),
        generator_speed_controller_reference=(
            self.generator_speed_controller_reference.as_write()
            if self.generator_speed_controller_reference
            else None
        ),
    )

GeneratorList

Bases: DomainModelList[Generator]

List of generators in the read version.

Source code in examples/windmill/data_classes/_generator.py

class GeneratorList(DomainModelList[Generator]):
    """List of generators in the read version."""

    _INSTANCE = Generator

    def as_write(self) -> GeneratorWriteList:
        """Convert these read versions of generator to the writing versions."""
        return GeneratorWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> GeneratorWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_generator.py

def as_apply(self) -> GeneratorWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of generator to the writing versions.

Source code in examples/windmill/data_classes/_generator.py

def as_write(self) -> GeneratorWriteList:
    """Convert these read versions of generator to the writing versions."""
    return GeneratorWriteList([node.as_write() for node in self.data])

GeneratorWrite

Bases: DomainModelWrite

This represents the writing version of generator.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the generator.	required
data_record		The data record of the generator node.	required
generator_speed_controller		The generator speed controller field.	required
generator_speed_controller_reference		The generator speed controller reference field.	required

Source code in examples/windmill/data_classes/_generator.py

class GeneratorWrite(DomainModelWrite):
    """This represents the writing version of generator.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the generator.
        data_record: The data record of the generator node.
        generator_speed_controller: The generator speed controller field.
        generator_speed_controller_reference: The generator speed controller reference field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Generator", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    generator_speed_controller: Union[TimeSeriesWrite, str, None] = None
    generator_speed_controller_reference: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.generator_speed_controller is not None or write_none:
            properties["generator_speed_controller"] = (
                self.generator_speed_controller
                if isinstance(self.generator_speed_controller, str) or self.generator_speed_controller is None
                else self.generator_speed_controller.external_id
            )

        if self.generator_speed_controller_reference is not None or write_none:
            properties["generator_speed_controller_reference"] = (
                self.generator_speed_controller_reference
                if isinstance(self.generator_speed_controller_reference, str)
                or self.generator_speed_controller_reference is None
                else self.generator_speed_controller_reference.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.generator_speed_controller, CogniteTimeSeriesWrite):
            resources.time_series.append(self.generator_speed_controller)

        if isinstance(self.generator_speed_controller_reference, CogniteTimeSeriesWrite):
            resources.time_series.append(self.generator_speed_controller_reference)

        return resources

GeneratorWriteList

Bases: DomainModelWriteList[GeneratorWrite]

List of generators in the writing version.

Source code in examples/windmill/data_classes/_generator.py

class GeneratorWriteList(DomainModelWriteList[GeneratorWrite]):
    """List of generators in the writing version."""

    _INSTANCE = GeneratorWrite

GraphQLList

Bases: UserList

Source code in examples/windmill/data_classes/_core/base.py

class GraphQLList(UserList):
    def __init__(self, nodes: Collection[GraphQLCore] | None = None):
        super().__init__(nodes or [])
        self.page_info: PageInfo | None = None

    # The dunder implementations are to get proper type hints
    def __iter__(self) -> Iterator[GraphQLCore]:
        return super().__iter__()

    @overload
    def __getitem__(self, item: SupportsIndex) -> GraphQLCore: ...

    @overload
    def __getitem__(self, item: slice) -> GraphQLList: ...

    def __getitem__(self, item: SupportsIndex | slice) -> GraphQLCore | GraphQLList:
        value = self.data[item]
        if isinstance(item, slice):
            return type(self)(value)
        return cast(GraphQLCore, value)

    def dump(self) -> list[dict[str, Any]]:
        return [node.model_dump() for node in self.data]

    def to_pandas(self) -> pd.DataFrame:
        """
        Convert the list of nodes to a pandas.DataFrame.

        Returns:
            A pandas.DataFrame with the nodes as rows.
        """
        df = pd.DataFrame(self.dump())
        if df.empty:
            df = pd.DataFrame(columns=GraphQLCore.model_fields)
        # Reorder columns to have the most relevant first
        id_columns = ["space", "external_id"]
        end_columns = ["data_record"]
        fixed_columns = set(id_columns + end_columns)
        columns = (
            id_columns + [col for col in df if col not in fixed_columns] + [col for col in end_columns if col in df]
        )
        return df[columns]

    def _repr_html_(self) -> str:
        return self.to_pandas()._repr_html_()  # type: ignore[operator]

to_pandas()

Convert the list of nodes to a pandas.DataFrame.

Returns:

Type	Description
DataFrame	A pandas.DataFrame with the nodes as rows.

Source code in examples/windmill/data_classes/_core/base.py

def to_pandas(self) -> pd.DataFrame:
    """
    Convert the list of nodes to a pandas.DataFrame.

    Returns:
        A pandas.DataFrame with the nodes as rows.
    """
    df = pd.DataFrame(self.dump())
    if df.empty:
        df = pd.DataFrame(columns=GraphQLCore.model_fields)
    # Reorder columns to have the most relevant first
    id_columns = ["space", "external_id"]
    end_columns = ["data_record"]
    fixed_columns = set(id_columns + end_columns)
    columns = (
        id_columns + [col for col in df if col not in fixed_columns] + [col for col in end_columns if col in df]
    )
    return df[columns]

HighSpeedShaft

Bases: DomainModel

This represents the reading version of high speed shaft.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the high speed shaft.	required
data_record		The data record of the high speed shaft node.	required
bending_moment_y		The bending moment y field.	required
bending_monent_x		The bending monent x field.	required
torque		The torque field.	required

Source code in examples/windmill/data_classes/_high_speed_shaft.py

class HighSpeedShaft(DomainModel):
    """This represents the reading version of high speed shaft.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the high speed shaft.
        data_record: The data record of the high speed shaft node.
        bending_moment_y: The bending moment y field.
        bending_monent_x: The bending monent x field.
        torque: The torque field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "HighSpeedShaft", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    bending_moment_y: Union[TimeSeries, str, None] = None
    bending_monent_x: Union[TimeSeries, str, None] = None
    torque: Union[TimeSeries, str, None] = None

    def as_write(self) -> HighSpeedShaftWrite:
        """Convert this read version of high speed shaft to the writing version."""
        return HighSpeedShaftWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            bending_moment_y=(
                self.bending_moment_y.as_write()
                if isinstance(self.bending_moment_y, CogniteTimeSeries)
                else self.bending_moment_y
            ),
            bending_monent_x=(
                self.bending_monent_x.as_write()
                if isinstance(self.bending_monent_x, CogniteTimeSeries)
                else self.bending_monent_x
            ),
            torque=self.torque.as_write() if isinstance(self.torque, CogniteTimeSeries) else self.torque,
        )

    def as_apply(self) -> HighSpeedShaftWrite:
        """Convert this read version of high speed shaft to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of high speed shaft to the writing version.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

def as_apply(self) -> HighSpeedShaftWrite:
    """Convert this read version of high speed shaft to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of high speed shaft to the writing version.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

def as_write(self) -> HighSpeedShaftWrite:
    """Convert this read version of high speed shaft to the writing version."""
    return HighSpeedShaftWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        bending_moment_y=(
            self.bending_moment_y.as_write()
            if isinstance(self.bending_moment_y, CogniteTimeSeries)
            else self.bending_moment_y
        ),
        bending_monent_x=(
            self.bending_monent_x.as_write()
            if isinstance(self.bending_monent_x, CogniteTimeSeries)
            else self.bending_monent_x
        ),
        torque=self.torque.as_write() if isinstance(self.torque, CogniteTimeSeries) else self.torque,
    )

HighSpeedShaftGraphQL

Bases: GraphQLCore

This represents the reading version of high speed shaft, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the high speed shaft.	required
data_record		The data record of the high speed shaft node.	required
bending_moment_y		The bending moment y field.	required
bending_monent_x		The bending monent x field.	required
torque		The torque field.	required

Source code in examples/windmill/data_classes/_high_speed_shaft.py

class HighSpeedShaftGraphQL(GraphQLCore):
    """This represents the reading version of high speed shaft, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the high speed shaft.
        data_record: The data record of the high speed shaft node.
        bending_moment_y: The bending moment y field.
        bending_monent_x: The bending monent x field.
        torque: The torque field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "HighSpeedShaft", "1")
    bending_moment_y: Optional[TimeSeriesGraphQL] = None
    bending_monent_x: Optional[TimeSeriesGraphQL] = None
    torque: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> HighSpeedShaft:
        """Convert this GraphQL format of high speed shaft to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return HighSpeedShaft(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            bending_moment_y=self.bending_moment_y.as_read() if self.bending_moment_y else None,
            bending_monent_x=self.bending_monent_x.as_read() if self.bending_monent_x else None,
            torque=self.torque.as_read() if self.torque else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> HighSpeedShaftWrite:
        """Convert this GraphQL format of high speed shaft to the writing format."""
        return HighSpeedShaftWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            bending_moment_y=self.bending_moment_y.as_write() if self.bending_moment_y else None,
            bending_monent_x=self.bending_monent_x.as_write() if self.bending_monent_x else None,
            torque=self.torque.as_write() if self.torque else None,
        )

as_read()

Convert this GraphQL format of high speed shaft to the reading format.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

@no_type_check
def as_read(self) -> HighSpeedShaft:
    """Convert this GraphQL format of high speed shaft to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return HighSpeedShaft(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        bending_moment_y=self.bending_moment_y.as_read() if self.bending_moment_y else None,
        bending_monent_x=self.bending_monent_x.as_read() if self.bending_monent_x else None,
        torque=self.torque.as_read() if self.torque else None,
    )

as_write()

Convert this GraphQL format of high speed shaft to the writing format.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

@no_type_check
def as_write(self) -> HighSpeedShaftWrite:
    """Convert this GraphQL format of high speed shaft to the writing format."""
    return HighSpeedShaftWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        bending_moment_y=self.bending_moment_y.as_write() if self.bending_moment_y else None,
        bending_monent_x=self.bending_monent_x.as_write() if self.bending_monent_x else None,
        torque=self.torque.as_write() if self.torque else None,
    )

HighSpeedShaftList

Bases: DomainModelList[HighSpeedShaft]

List of high speed shafts in the read version.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

class HighSpeedShaftList(DomainModelList[HighSpeedShaft]):
    """List of high speed shafts in the read version."""

    _INSTANCE = HighSpeedShaft

    def as_write(self) -> HighSpeedShaftWriteList:
        """Convert these read versions of high speed shaft to the writing versions."""
        return HighSpeedShaftWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> HighSpeedShaftWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

def as_apply(self) -> HighSpeedShaftWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of high speed shaft to the writing versions.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

def as_write(self) -> HighSpeedShaftWriteList:
    """Convert these read versions of high speed shaft to the writing versions."""
    return HighSpeedShaftWriteList([node.as_write() for node in self.data])

HighSpeedShaftWrite

Bases: DomainModelWrite

This represents the writing version of high speed shaft.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the high speed shaft.	required
data_record		The data record of the high speed shaft node.	required
bending_moment_y		The bending moment y field.	required
bending_monent_x		The bending monent x field.	required
torque		The torque field.	required

Source code in examples/windmill/data_classes/_high_speed_shaft.py

class HighSpeedShaftWrite(DomainModelWrite):
    """This represents the writing version of high speed shaft.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the high speed shaft.
        data_record: The data record of the high speed shaft node.
        bending_moment_y: The bending moment y field.
        bending_monent_x: The bending monent x field.
        torque: The torque field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "HighSpeedShaft", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    bending_moment_y: Union[TimeSeriesWrite, str, None] = None
    bending_monent_x: Union[TimeSeriesWrite, str, None] = None
    torque: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.bending_moment_y is not None or write_none:
            properties["bending_moment_y"] = (
                self.bending_moment_y
                if isinstance(self.bending_moment_y, str) or self.bending_moment_y is None
                else self.bending_moment_y.external_id
            )

        if self.bending_monent_x is not None or write_none:
            properties["bending_monent_x"] = (
                self.bending_monent_x
                if isinstance(self.bending_monent_x, str) or self.bending_monent_x is None
                else self.bending_monent_x.external_id
            )

        if self.torque is not None or write_none:
            properties["torque"] = (
                self.torque if isinstance(self.torque, str) or self.torque is None else self.torque.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.bending_moment_y, CogniteTimeSeriesWrite):
            resources.time_series.append(self.bending_moment_y)

        if isinstance(self.bending_monent_x, CogniteTimeSeriesWrite):
            resources.time_series.append(self.bending_monent_x)

        if isinstance(self.torque, CogniteTimeSeriesWrite):
            resources.time_series.append(self.torque)

        return resources

HighSpeedShaftWriteList

Bases: DomainModelWriteList[HighSpeedShaftWrite]

List of high speed shafts in the writing version.

Source code in examples/windmill/data_classes/_high_speed_shaft.py

class HighSpeedShaftWriteList(DomainModelWriteList[HighSpeedShaftWrite]):
    """List of high speed shafts in the writing version."""

    _INSTANCE = HighSpeedShaftWrite

MainShaft

Bases: DomainModel

This represents the reading version of main shaft.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the main shaft.	required
data_record		The data record of the main shaft node.	required
bending_x		The bending x field.	required
bending_y		The bending y field.	required
calculated_tilt_moment		The calculated tilt moment field.	required
calculated_yaw_moment		The calculated yaw moment field.	required
torque		The torque field.	required

Source code in examples/windmill/data_classes/_main_shaft.py

class MainShaft(DomainModel):
    """This represents the reading version of main shaft.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the main shaft.
        data_record: The data record of the main shaft node.
        bending_x: The bending x field.
        bending_y: The bending y field.
        calculated_tilt_moment: The calculated tilt moment field.
        calculated_yaw_moment: The calculated yaw moment field.
        torque: The torque field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "MainShaft", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    bending_x: Union[TimeSeries, str, None] = None
    bending_y: Union[TimeSeries, str, None] = None
    calculated_tilt_moment: Union[TimeSeries, str, None] = None
    calculated_yaw_moment: Union[TimeSeries, str, None] = None
    torque: Union[TimeSeries, str, None] = None

    def as_write(self) -> MainShaftWrite:
        """Convert this read version of main shaft to the writing version."""
        return MainShaftWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            bending_x=self.bending_x.as_write() if isinstance(self.bending_x, CogniteTimeSeries) else self.bending_x,
            bending_y=self.bending_y.as_write() if isinstance(self.bending_y, CogniteTimeSeries) else self.bending_y,
            calculated_tilt_moment=(
                self.calculated_tilt_moment.as_write()
                if isinstance(self.calculated_tilt_moment, CogniteTimeSeries)
                else self.calculated_tilt_moment
            ),
            calculated_yaw_moment=(
                self.calculated_yaw_moment.as_write()
                if isinstance(self.calculated_yaw_moment, CogniteTimeSeries)
                else self.calculated_yaw_moment
            ),
            torque=self.torque.as_write() if isinstance(self.torque, CogniteTimeSeries) else self.torque,
        )

    def as_apply(self) -> MainShaftWrite:
        """Convert this read version of main shaft to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of main shaft to the writing version.

Source code in examples/windmill/data_classes/_main_shaft.py

def as_apply(self) -> MainShaftWrite:
    """Convert this read version of main shaft to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of main shaft to the writing version.

Source code in examples/windmill/data_classes/_main_shaft.py

def as_write(self) -> MainShaftWrite:
    """Convert this read version of main shaft to the writing version."""
    return MainShaftWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        bending_x=self.bending_x.as_write() if isinstance(self.bending_x, CogniteTimeSeries) else self.bending_x,
        bending_y=self.bending_y.as_write() if isinstance(self.bending_y, CogniteTimeSeries) else self.bending_y,
        calculated_tilt_moment=(
            self.calculated_tilt_moment.as_write()
            if isinstance(self.calculated_tilt_moment, CogniteTimeSeries)
            else self.calculated_tilt_moment
        ),
        calculated_yaw_moment=(
            self.calculated_yaw_moment.as_write()
            if isinstance(self.calculated_yaw_moment, CogniteTimeSeries)
            else self.calculated_yaw_moment
        ),
        torque=self.torque.as_write() if isinstance(self.torque, CogniteTimeSeries) else self.torque,
    )

MainShaftGraphQL

Bases: GraphQLCore

This represents the reading version of main shaft, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the main shaft.	required
data_record		The data record of the main shaft node.	required
bending_x		The bending x field.	required
bending_y		The bending y field.	required
calculated_tilt_moment		The calculated tilt moment field.	required
calculated_yaw_moment		The calculated yaw moment field.	required
torque		The torque field.	required

Source code in examples/windmill/data_classes/_main_shaft.py

class MainShaftGraphQL(GraphQLCore):
    """This represents the reading version of main shaft, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the main shaft.
        data_record: The data record of the main shaft node.
        bending_x: The bending x field.
        bending_y: The bending y field.
        calculated_tilt_moment: The calculated tilt moment field.
        calculated_yaw_moment: The calculated yaw moment field.
        torque: The torque field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "MainShaft", "1")
    bending_x: Optional[TimeSeriesGraphQL] = None
    bending_y: Optional[TimeSeriesGraphQL] = None
    calculated_tilt_moment: Optional[TimeSeriesGraphQL] = None
    calculated_yaw_moment: Optional[TimeSeriesGraphQL] = None
    torque: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> MainShaft:
        """Convert this GraphQL format of main shaft to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return MainShaft(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            bending_x=self.bending_x.as_read() if self.bending_x else None,
            bending_y=self.bending_y.as_read() if self.bending_y else None,
            calculated_tilt_moment=self.calculated_tilt_moment.as_read() if self.calculated_tilt_moment else None,
            calculated_yaw_moment=self.calculated_yaw_moment.as_read() if self.calculated_yaw_moment else None,
            torque=self.torque.as_read() if self.torque else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> MainShaftWrite:
        """Convert this GraphQL format of main shaft to the writing format."""
        return MainShaftWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            bending_x=self.bending_x.as_write() if self.bending_x else None,
            bending_y=self.bending_y.as_write() if self.bending_y else None,
            calculated_tilt_moment=self.calculated_tilt_moment.as_write() if self.calculated_tilt_moment else None,
            calculated_yaw_moment=self.calculated_yaw_moment.as_write() if self.calculated_yaw_moment else None,
            torque=self.torque.as_write() if self.torque else None,
        )

as_read()

Convert this GraphQL format of main shaft to the reading format.

Source code in examples/windmill/data_classes/_main_shaft.py

@no_type_check
def as_read(self) -> MainShaft:
    """Convert this GraphQL format of main shaft to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return MainShaft(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        bending_x=self.bending_x.as_read() if self.bending_x else None,
        bending_y=self.bending_y.as_read() if self.bending_y else None,
        calculated_tilt_moment=self.calculated_tilt_moment.as_read() if self.calculated_tilt_moment else None,
        calculated_yaw_moment=self.calculated_yaw_moment.as_read() if self.calculated_yaw_moment else None,
        torque=self.torque.as_read() if self.torque else None,
    )

as_write()

Convert this GraphQL format of main shaft to the writing format.

Source code in examples/windmill/data_classes/_main_shaft.py

@no_type_check
def as_write(self) -> MainShaftWrite:
    """Convert this GraphQL format of main shaft to the writing format."""
    return MainShaftWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        bending_x=self.bending_x.as_write() if self.bending_x else None,
        bending_y=self.bending_y.as_write() if self.bending_y else None,
        calculated_tilt_moment=self.calculated_tilt_moment.as_write() if self.calculated_tilt_moment else None,
        calculated_yaw_moment=self.calculated_yaw_moment.as_write() if self.calculated_yaw_moment else None,
        torque=self.torque.as_write() if self.torque else None,
    )

MainShaftList

Bases: DomainModelList[MainShaft]

List of main shafts in the read version.

Source code in examples/windmill/data_classes/_main_shaft.py

class MainShaftList(DomainModelList[MainShaft]):
    """List of main shafts in the read version."""

    _INSTANCE = MainShaft

    def as_write(self) -> MainShaftWriteList:
        """Convert these read versions of main shaft to the writing versions."""
        return MainShaftWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> MainShaftWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_main_shaft.py

def as_apply(self) -> MainShaftWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of main shaft to the writing versions.

Source code in examples/windmill/data_classes/_main_shaft.py

def as_write(self) -> MainShaftWriteList:
    """Convert these read versions of main shaft to the writing versions."""
    return MainShaftWriteList([node.as_write() for node in self.data])

MainShaftWrite

Bases: DomainModelWrite

This represents the writing version of main shaft.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the main shaft.	required
data_record		The data record of the main shaft node.	required
bending_x		The bending x field.	required
bending_y		The bending y field.	required
calculated_tilt_moment		The calculated tilt moment field.	required
calculated_yaw_moment		The calculated yaw moment field.	required
torque		The torque field.	required

Source code in examples/windmill/data_classes/_main_shaft.py

class MainShaftWrite(DomainModelWrite):
    """This represents the writing version of main shaft.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the main shaft.
        data_record: The data record of the main shaft node.
        bending_x: The bending x field.
        bending_y: The bending y field.
        calculated_tilt_moment: The calculated tilt moment field.
        calculated_yaw_moment: The calculated yaw moment field.
        torque: The torque field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "MainShaft", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    bending_x: Union[TimeSeriesWrite, str, None] = None
    bending_y: Union[TimeSeriesWrite, str, None] = None
    calculated_tilt_moment: Union[TimeSeriesWrite, str, None] = None
    calculated_yaw_moment: Union[TimeSeriesWrite, str, None] = None
    torque: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.bending_x is not None or write_none:
            properties["bending_x"] = (
                self.bending_x
                if isinstance(self.bending_x, str) or self.bending_x is None
                else self.bending_x.external_id
            )

        if self.bending_y is not None or write_none:
            properties["bending_y"] = (
                self.bending_y
                if isinstance(self.bending_y, str) or self.bending_y is None
                else self.bending_y.external_id
            )

        if self.calculated_tilt_moment is not None or write_none:
            properties["calculated_tilt_moment"] = (
                self.calculated_tilt_moment
                if isinstance(self.calculated_tilt_moment, str) or self.calculated_tilt_moment is None
                else self.calculated_tilt_moment.external_id
            )

        if self.calculated_yaw_moment is not None or write_none:
            properties["calculated_yaw_moment"] = (
                self.calculated_yaw_moment
                if isinstance(self.calculated_yaw_moment, str) or self.calculated_yaw_moment is None
                else self.calculated_yaw_moment.external_id
            )

        if self.torque is not None or write_none:
            properties["torque"] = (
                self.torque if isinstance(self.torque, str) or self.torque is None else self.torque.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.bending_x, CogniteTimeSeriesWrite):
            resources.time_series.append(self.bending_x)

        if isinstance(self.bending_y, CogniteTimeSeriesWrite):
            resources.time_series.append(self.bending_y)

        if isinstance(self.calculated_tilt_moment, CogniteTimeSeriesWrite):
            resources.time_series.append(self.calculated_tilt_moment)

        if isinstance(self.calculated_yaw_moment, CogniteTimeSeriesWrite):
            resources.time_series.append(self.calculated_yaw_moment)

        if isinstance(self.torque, CogniteTimeSeriesWrite):
            resources.time_series.append(self.torque)

        return resources

MainShaftWriteList

Bases: DomainModelWriteList[MainShaftWrite]

List of main shafts in the writing version.

Source code in examples/windmill/data_classes/_main_shaft.py

class MainShaftWriteList(DomainModelWriteList[MainShaftWrite]):
    """List of main shafts in the writing version."""

    _INSTANCE = MainShaftWrite

Metmast

Bases: DomainModel

This represents the reading version of metmast.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the metmast.	required
data_record		The data record of the metmast node.	required
position		The position field.	required
temperature		The temperature field.	required
tilt_angle		The tilt angle field.	required
wind_speed		The wind speed field.	required

Source code in examples/windmill/data_classes/_metmast.py

class Metmast(DomainModel):
    """This represents the reading version of metmast.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the metmast.
        data_record: The data record of the metmast node.
        position: The position field.
        temperature: The temperature field.
        tilt_angle: The tilt angle field.
        wind_speed: The wind speed field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Metmast", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    position: Optional[float] = None
    temperature: Union[TimeSeries, str, None] = None
    tilt_angle: Union[TimeSeries, str, None] = None
    wind_speed: Union[TimeSeries, str, None] = None

    def as_write(self) -> MetmastWrite:
        """Convert this read version of metmast to the writing version."""
        return MetmastWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            position=self.position,
            temperature=(
                self.temperature.as_write() if isinstance(self.temperature, CogniteTimeSeries) else self.temperature
            ),
            tilt_angle=(
                self.tilt_angle.as_write() if isinstance(self.tilt_angle, CogniteTimeSeries) else self.tilt_angle
            ),
            wind_speed=(
                self.wind_speed.as_write() if isinstance(self.wind_speed, CogniteTimeSeries) else self.wind_speed
            ),
        )

    def as_apply(self) -> MetmastWrite:
        """Convert this read version of metmast to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of metmast to the writing version.

Source code in examples/windmill/data_classes/_metmast.py

def as_apply(self) -> MetmastWrite:
    """Convert this read version of metmast to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of metmast to the writing version.

Source code in examples/windmill/data_classes/_metmast.py

def as_write(self) -> MetmastWrite:
    """Convert this read version of metmast to the writing version."""
    return MetmastWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        position=self.position,
        temperature=(
            self.temperature.as_write() if isinstance(self.temperature, CogniteTimeSeries) else self.temperature
        ),
        tilt_angle=(
            self.tilt_angle.as_write() if isinstance(self.tilt_angle, CogniteTimeSeries) else self.tilt_angle
        ),
        wind_speed=(
            self.wind_speed.as_write() if isinstance(self.wind_speed, CogniteTimeSeries) else self.wind_speed
        ),
    )

MetmastGraphQL

Bases: GraphQLCore

This represents the reading version of metmast, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the metmast.	required
data_record		The data record of the metmast node.	required
position		The position field.	required
temperature		The temperature field.	required
tilt_angle		The tilt angle field.	required
wind_speed		The wind speed field.	required

Source code in examples/windmill/data_classes/_metmast.py

class MetmastGraphQL(GraphQLCore):
    """This represents the reading version of metmast, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the metmast.
        data_record: The data record of the metmast node.
        position: The position field.
        temperature: The temperature field.
        tilt_angle: The tilt angle field.
        wind_speed: The wind speed field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Metmast", "1")
    position: Optional[float] = None
    temperature: Optional[TimeSeriesGraphQL] = None
    tilt_angle: Optional[TimeSeriesGraphQL] = None
    wind_speed: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Metmast:
        """Convert this GraphQL format of metmast to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Metmast(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            position=self.position,
            temperature=self.temperature.as_read() if self.temperature else None,
            tilt_angle=self.tilt_angle.as_read() if self.tilt_angle else None,
            wind_speed=self.wind_speed.as_read() if self.wind_speed else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> MetmastWrite:
        """Convert this GraphQL format of metmast to the writing format."""
        return MetmastWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            position=self.position,
            temperature=self.temperature.as_write() if self.temperature else None,
            tilt_angle=self.tilt_angle.as_write() if self.tilt_angle else None,
            wind_speed=self.wind_speed.as_write() if self.wind_speed else None,
        )

as_read()

Convert this GraphQL format of metmast to the reading format.

Source code in examples/windmill/data_classes/_metmast.py

@no_type_check
def as_read(self) -> Metmast:
    """Convert this GraphQL format of metmast to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Metmast(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        position=self.position,
        temperature=self.temperature.as_read() if self.temperature else None,
        tilt_angle=self.tilt_angle.as_read() if self.tilt_angle else None,
        wind_speed=self.wind_speed.as_read() if self.wind_speed else None,
    )

as_write()

Convert this GraphQL format of metmast to the writing format.

Source code in examples/windmill/data_classes/_metmast.py

@no_type_check
def as_write(self) -> MetmastWrite:
    """Convert this GraphQL format of metmast to the writing format."""
    return MetmastWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        position=self.position,
        temperature=self.temperature.as_write() if self.temperature else None,
        tilt_angle=self.tilt_angle.as_write() if self.tilt_angle else None,
        wind_speed=self.wind_speed.as_write() if self.wind_speed else None,
    )

MetmastList

Bases: DomainModelList[Metmast]

List of metmasts in the read version.

Source code in examples/windmill/data_classes/_metmast.py

class MetmastList(DomainModelList[Metmast]):
    """List of metmasts in the read version."""

    _INSTANCE = Metmast

    def as_write(self) -> MetmastWriteList:
        """Convert these read versions of metmast to the writing versions."""
        return MetmastWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> MetmastWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_metmast.py

def as_apply(self) -> MetmastWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of metmast to the writing versions.

Source code in examples/windmill/data_classes/_metmast.py

def as_write(self) -> MetmastWriteList:
    """Convert these read versions of metmast to the writing versions."""
    return MetmastWriteList([node.as_write() for node in self.data])

MetmastWrite

Bases: DomainModelWrite

This represents the writing version of metmast.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the metmast.	required
data_record		The data record of the metmast node.	required
position		The position field.	required
temperature		The temperature field.	required
tilt_angle		The tilt angle field.	required
wind_speed		The wind speed field.	required

Source code in examples/windmill/data_classes/_metmast.py

class MetmastWrite(DomainModelWrite):
    """This represents the writing version of metmast.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the metmast.
        data_record: The data record of the metmast node.
        position: The position field.
        temperature: The temperature field.
        tilt_angle: The tilt angle field.
        wind_speed: The wind speed field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Metmast", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    position: Optional[float] = None
    temperature: Union[TimeSeriesWrite, str, None] = None
    tilt_angle: Union[TimeSeriesWrite, str, None] = None
    wind_speed: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.position is not None or write_none:
            properties["position"] = self.position

        if self.temperature is not None or write_none:
            properties["temperature"] = (
                self.temperature
                if isinstance(self.temperature, str) or self.temperature is None
                else self.temperature.external_id
            )

        if self.tilt_angle is not None or write_none:
            properties["tilt_angle"] = (
                self.tilt_angle
                if isinstance(self.tilt_angle, str) or self.tilt_angle is None
                else self.tilt_angle.external_id
            )

        if self.wind_speed is not None or write_none:
            properties["wind_speed"] = (
                self.wind_speed
                if isinstance(self.wind_speed, str) or self.wind_speed is None
                else self.wind_speed.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.temperature, CogniteTimeSeriesWrite):
            resources.time_series.append(self.temperature)

        if isinstance(self.tilt_angle, CogniteTimeSeriesWrite):
            resources.time_series.append(self.tilt_angle)

        if isinstance(self.wind_speed, CogniteTimeSeriesWrite):
            resources.time_series.append(self.wind_speed)

        return resources

MetmastWriteList

Bases: DomainModelWriteList[MetmastWrite]

List of metmasts in the writing version.

Source code in examples/windmill/data_classes/_metmast.py

class MetmastWriteList(DomainModelWriteList[MetmastWrite]):
    """List of metmasts in the writing version."""

    _INSTANCE = MetmastWrite

Nacelle

Bases: DomainModel

This represents the reading version of nacelle.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the nacelle.	required
data_record		The data record of the nacelle node.	required
acc_from_back_side_x		The acc from back side x field.	required
acc_from_back_side_y		The acc from back side y field.	required
acc_from_back_side_z		The acc from back side z field.	required
gearbox		The gearbox field.	required
generator		The generator field.	required
high_speed_shaft		The high speed shaft field.	required
main_shaft		The main shaft field.	required
power_inverter		The power inverter field.	required
yaw_direction		The yaw direction field.	required
yaw_error		The yaw error field.	required

Source code in examples/windmill/data_classes/_nacelle.py

class Nacelle(DomainModel):
    """This represents the reading version of nacelle.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the nacelle.
        data_record: The data record of the nacelle node.
        acc_from_back_side_x: The acc from back side x field.
        acc_from_back_side_y: The acc from back side y field.
        acc_from_back_side_z: The acc from back side z field.
        gearbox: The gearbox field.
        generator: The generator field.
        high_speed_shaft: The high speed shaft field.
        main_shaft: The main shaft field.
        power_inverter: The power inverter field.
        yaw_direction: The yaw direction field.
        yaw_error: The yaw error field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Nacelle", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    acc_from_back_side_x: Union[TimeSeries, str, None] = None
    acc_from_back_side_y: Union[TimeSeries, str, None] = None
    acc_from_back_side_z: Union[TimeSeries, str, None] = None
    gearbox: Union[Gearbox, str, dm.NodeId, None] = Field(default=None, repr=False)
    generator: Union[Generator, str, dm.NodeId, None] = Field(default=None, repr=False)
    high_speed_shaft: Union[HighSpeedShaft, str, dm.NodeId, None] = Field(default=None, repr=False)
    main_shaft: Union[MainShaft, str, dm.NodeId, None] = Field(default=None, repr=False)
    power_inverter: Union[PowerInverter, str, dm.NodeId, None] = Field(default=None, repr=False)
    yaw_direction: Union[TimeSeries, str, None] = None
    yaw_error: Union[TimeSeries, str, None] = None

    def as_write(self) -> NacelleWrite:
        """Convert this read version of nacelle to the writing version."""
        return NacelleWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            acc_from_back_side_x=(
                self.acc_from_back_side_x.as_write()
                if isinstance(self.acc_from_back_side_x, CogniteTimeSeries)
                else self.acc_from_back_side_x
            ),
            acc_from_back_side_y=(
                self.acc_from_back_side_y.as_write()
                if isinstance(self.acc_from_back_side_y, CogniteTimeSeries)
                else self.acc_from_back_side_y
            ),
            acc_from_back_side_z=(
                self.acc_from_back_side_z.as_write()
                if isinstance(self.acc_from_back_side_z, CogniteTimeSeries)
                else self.acc_from_back_side_z
            ),
            gearbox=self.gearbox.as_write() if isinstance(self.gearbox, DomainModel) else self.gearbox,
            generator=self.generator.as_write() if isinstance(self.generator, DomainModel) else self.generator,
            high_speed_shaft=(
                self.high_speed_shaft.as_write()
                if isinstance(self.high_speed_shaft, DomainModel)
                else self.high_speed_shaft
            ),
            main_shaft=self.main_shaft.as_write() if isinstance(self.main_shaft, DomainModel) else self.main_shaft,
            power_inverter=(
                self.power_inverter.as_write() if isinstance(self.power_inverter, DomainModel) else self.power_inverter
            ),
            yaw_direction=(
                self.yaw_direction.as_write()
                if isinstance(self.yaw_direction, CogniteTimeSeries)
                else self.yaw_direction
            ),
            yaw_error=self.yaw_error.as_write() if isinstance(self.yaw_error, CogniteTimeSeries) else self.yaw_error,
        )

    def as_apply(self) -> NacelleWrite:
        """Convert this read version of nacelle to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, Nacelle],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._gearbox import Gearbox
        from ._generator import Generator
        from ._high_speed_shaft import HighSpeedShaft
        from ._main_shaft import MainShaft
        from ._power_inverter import PowerInverter

        for instance in instances.values():
            if (
                isinstance(instance.gearbox, (dm.NodeId, str))
                and (gearbox := nodes_by_id.get(instance.gearbox))
                and isinstance(gearbox, Gearbox)
            ):
                instance.gearbox = gearbox
            if (
                isinstance(instance.generator, (dm.NodeId, str))
                and (generator := nodes_by_id.get(instance.generator))
                and isinstance(generator, Generator)
            ):
                instance.generator = generator
            if (
                isinstance(instance.high_speed_shaft, (dm.NodeId, str))
                and (high_speed_shaft := nodes_by_id.get(instance.high_speed_shaft))
                and isinstance(high_speed_shaft, HighSpeedShaft)
            ):
                instance.high_speed_shaft = high_speed_shaft
            if (
                isinstance(instance.main_shaft, (dm.NodeId, str))
                and (main_shaft := nodes_by_id.get(instance.main_shaft))
                and isinstance(main_shaft, MainShaft)
            ):
                instance.main_shaft = main_shaft
            if (
                isinstance(instance.power_inverter, (dm.NodeId, str))
                and (power_inverter := nodes_by_id.get(instance.power_inverter))
                and isinstance(power_inverter, PowerInverter)
            ):
                instance.power_inverter = power_inverter

as_apply()

Convert this read version of nacelle to the writing version.

Source code in examples/windmill/data_classes/_nacelle.py

def as_apply(self) -> NacelleWrite:
    """Convert this read version of nacelle to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of nacelle to the writing version.

Source code in examples/windmill/data_classes/_nacelle.py

def as_write(self) -> NacelleWrite:
    """Convert this read version of nacelle to the writing version."""
    return NacelleWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        acc_from_back_side_x=(
            self.acc_from_back_side_x.as_write()
            if isinstance(self.acc_from_back_side_x, CogniteTimeSeries)
            else self.acc_from_back_side_x
        ),
        acc_from_back_side_y=(
            self.acc_from_back_side_y.as_write()
            if isinstance(self.acc_from_back_side_y, CogniteTimeSeries)
            else self.acc_from_back_side_y
        ),
        acc_from_back_side_z=(
            self.acc_from_back_side_z.as_write()
            if isinstance(self.acc_from_back_side_z, CogniteTimeSeries)
            else self.acc_from_back_side_z
        ),
        gearbox=self.gearbox.as_write() if isinstance(self.gearbox, DomainModel) else self.gearbox,
        generator=self.generator.as_write() if isinstance(self.generator, DomainModel) else self.generator,
        high_speed_shaft=(
            self.high_speed_shaft.as_write()
            if isinstance(self.high_speed_shaft, DomainModel)
            else self.high_speed_shaft
        ),
        main_shaft=self.main_shaft.as_write() if isinstance(self.main_shaft, DomainModel) else self.main_shaft,
        power_inverter=(
            self.power_inverter.as_write() if isinstance(self.power_inverter, DomainModel) else self.power_inverter
        ),
        yaw_direction=(
            self.yaw_direction.as_write()
            if isinstance(self.yaw_direction, CogniteTimeSeries)
            else self.yaw_direction
        ),
        yaw_error=self.yaw_error.as_write() if isinstance(self.yaw_error, CogniteTimeSeries) else self.yaw_error,
    )

NacelleGraphQL

Bases: GraphQLCore

This represents the reading version of nacelle, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the nacelle.	required
data_record		The data record of the nacelle node.	required
acc_from_back_side_x		The acc from back side x field.	required
acc_from_back_side_y		The acc from back side y field.	required
acc_from_back_side_z		The acc from back side z field.	required
gearbox		The gearbox field.	required
generator		The generator field.	required
high_speed_shaft		The high speed shaft field.	required
main_shaft		The main shaft field.	required
power_inverter		The power inverter field.	required
yaw_direction		The yaw direction field.	required
yaw_error		The yaw error field.	required

Source code in examples/windmill/data_classes/_nacelle.py

class NacelleGraphQL(GraphQLCore):
    """This represents the reading version of nacelle, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the nacelle.
        data_record: The data record of the nacelle node.
        acc_from_back_side_x: The acc from back side x field.
        acc_from_back_side_y: The acc from back side y field.
        acc_from_back_side_z: The acc from back side z field.
        gearbox: The gearbox field.
        generator: The generator field.
        high_speed_shaft: The high speed shaft field.
        main_shaft: The main shaft field.
        power_inverter: The power inverter field.
        yaw_direction: The yaw direction field.
        yaw_error: The yaw error field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Nacelle", "1")
    acc_from_back_side_x: Optional[TimeSeriesGraphQL] = None
    acc_from_back_side_y: Optional[TimeSeriesGraphQL] = None
    acc_from_back_side_z: Optional[TimeSeriesGraphQL] = None
    gearbox: Optional[GearboxGraphQL] = Field(default=None, repr=False)
    generator: Optional[GeneratorGraphQL] = Field(default=None, repr=False)
    high_speed_shaft: Optional[HighSpeedShaftGraphQL] = Field(default=None, repr=False)
    main_shaft: Optional[MainShaftGraphQL] = Field(default=None, repr=False)
    power_inverter: Optional[PowerInverterGraphQL] = Field(default=None, repr=False)
    yaw_direction: Optional[TimeSeriesGraphQL] = None
    yaw_error: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("gearbox", "generator", "high_speed_shaft", "main_shaft", "power_inverter", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Nacelle:
        """Convert this GraphQL format of nacelle to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Nacelle(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            acc_from_back_side_x=self.acc_from_back_side_x.as_read() if self.acc_from_back_side_x else None,
            acc_from_back_side_y=self.acc_from_back_side_y.as_read() if self.acc_from_back_side_y else None,
            acc_from_back_side_z=self.acc_from_back_side_z.as_read() if self.acc_from_back_side_z else None,
            gearbox=self.gearbox.as_read() if isinstance(self.gearbox, GraphQLCore) else self.gearbox,
            generator=self.generator.as_read() if isinstance(self.generator, GraphQLCore) else self.generator,
            high_speed_shaft=(
                self.high_speed_shaft.as_read()
                if isinstance(self.high_speed_shaft, GraphQLCore)
                else self.high_speed_shaft
            ),
            main_shaft=self.main_shaft.as_read() if isinstance(self.main_shaft, GraphQLCore) else self.main_shaft,
            power_inverter=(
                self.power_inverter.as_read() if isinstance(self.power_inverter, GraphQLCore) else self.power_inverter
            ),
            yaw_direction=self.yaw_direction.as_read() if self.yaw_direction else None,
            yaw_error=self.yaw_error.as_read() if self.yaw_error else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> NacelleWrite:
        """Convert this GraphQL format of nacelle to the writing format."""
        return NacelleWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            acc_from_back_side_x=self.acc_from_back_side_x.as_write() if self.acc_from_back_side_x else None,
            acc_from_back_side_y=self.acc_from_back_side_y.as_write() if self.acc_from_back_side_y else None,
            acc_from_back_side_z=self.acc_from_back_side_z.as_write() if self.acc_from_back_side_z else None,
            gearbox=self.gearbox.as_write() if isinstance(self.gearbox, GraphQLCore) else self.gearbox,
            generator=self.generator.as_write() if isinstance(self.generator, GraphQLCore) else self.generator,
            high_speed_shaft=(
                self.high_speed_shaft.as_write()
                if isinstance(self.high_speed_shaft, GraphQLCore)
                else self.high_speed_shaft
            ),
            main_shaft=self.main_shaft.as_write() if isinstance(self.main_shaft, GraphQLCore) else self.main_shaft,
            power_inverter=(
                self.power_inverter.as_write() if isinstance(self.power_inverter, GraphQLCore) else self.power_inverter
            ),
            yaw_direction=self.yaw_direction.as_write() if self.yaw_direction else None,
            yaw_error=self.yaw_error.as_write() if self.yaw_error else None,
        )

as_read()

Convert this GraphQL format of nacelle to the reading format.

Source code in examples/windmill/data_classes/_nacelle.py

@no_type_check
def as_read(self) -> Nacelle:
    """Convert this GraphQL format of nacelle to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Nacelle(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        acc_from_back_side_x=self.acc_from_back_side_x.as_read() if self.acc_from_back_side_x else None,
        acc_from_back_side_y=self.acc_from_back_side_y.as_read() if self.acc_from_back_side_y else None,
        acc_from_back_side_z=self.acc_from_back_side_z.as_read() if self.acc_from_back_side_z else None,
        gearbox=self.gearbox.as_read() if isinstance(self.gearbox, GraphQLCore) else self.gearbox,
        generator=self.generator.as_read() if isinstance(self.generator, GraphQLCore) else self.generator,
        high_speed_shaft=(
            self.high_speed_shaft.as_read()
            if isinstance(self.high_speed_shaft, GraphQLCore)
            else self.high_speed_shaft
        ),
        main_shaft=self.main_shaft.as_read() if isinstance(self.main_shaft, GraphQLCore) else self.main_shaft,
        power_inverter=(
            self.power_inverter.as_read() if isinstance(self.power_inverter, GraphQLCore) else self.power_inverter
        ),
        yaw_direction=self.yaw_direction.as_read() if self.yaw_direction else None,
        yaw_error=self.yaw_error.as_read() if self.yaw_error else None,
    )

as_write()

Convert this GraphQL format of nacelle to the writing format.

Source code in examples/windmill/data_classes/_nacelle.py

@no_type_check
def as_write(self) -> NacelleWrite:
    """Convert this GraphQL format of nacelle to the writing format."""
    return NacelleWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        acc_from_back_side_x=self.acc_from_back_side_x.as_write() if self.acc_from_back_side_x else None,
        acc_from_back_side_y=self.acc_from_back_side_y.as_write() if self.acc_from_back_side_y else None,
        acc_from_back_side_z=self.acc_from_back_side_z.as_write() if self.acc_from_back_side_z else None,
        gearbox=self.gearbox.as_write() if isinstance(self.gearbox, GraphQLCore) else self.gearbox,
        generator=self.generator.as_write() if isinstance(self.generator, GraphQLCore) else self.generator,
        high_speed_shaft=(
            self.high_speed_shaft.as_write()
            if isinstance(self.high_speed_shaft, GraphQLCore)
            else self.high_speed_shaft
        ),
        main_shaft=self.main_shaft.as_write() if isinstance(self.main_shaft, GraphQLCore) else self.main_shaft,
        power_inverter=(
            self.power_inverter.as_write() if isinstance(self.power_inverter, GraphQLCore) else self.power_inverter
        ),
        yaw_direction=self.yaw_direction.as_write() if self.yaw_direction else None,
        yaw_error=self.yaw_error.as_write() if self.yaw_error else None,
    )

NacelleList

Bases: DomainModelList[Nacelle]

List of nacelles in the read version.

Source code in examples/windmill/data_classes/_nacelle.py

class NacelleList(DomainModelList[Nacelle]):
    """List of nacelles in the read version."""

    _INSTANCE = Nacelle

    def as_write(self) -> NacelleWriteList:
        """Convert these read versions of nacelle to the writing versions."""
        return NacelleWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> NacelleWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_nacelle.py

def as_apply(self) -> NacelleWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of nacelle to the writing versions.

Source code in examples/windmill/data_classes/_nacelle.py

def as_write(self) -> NacelleWriteList:
    """Convert these read versions of nacelle to the writing versions."""
    return NacelleWriteList([node.as_write() for node in self.data])

NacelleWrite

Bases: DomainModelWrite

This represents the writing version of nacelle.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the nacelle.	required
data_record		The data record of the nacelle node.	required
acc_from_back_side_x		The acc from back side x field.	required
acc_from_back_side_y		The acc from back side y field.	required
acc_from_back_side_z		The acc from back side z field.	required
gearbox		The gearbox field.	required
generator		The generator field.	required
high_speed_shaft		The high speed shaft field.	required
main_shaft		The main shaft field.	required
power_inverter		The power inverter field.	required
yaw_direction		The yaw direction field.	required
yaw_error		The yaw error field.	required

Source code in examples/windmill/data_classes/_nacelle.py

class NacelleWrite(DomainModelWrite):
    """This represents the writing version of nacelle.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the nacelle.
        data_record: The data record of the nacelle node.
        acc_from_back_side_x: The acc from back side x field.
        acc_from_back_side_y: The acc from back side y field.
        acc_from_back_side_z: The acc from back side z field.
        gearbox: The gearbox field.
        generator: The generator field.
        high_speed_shaft: The high speed shaft field.
        main_shaft: The main shaft field.
        power_inverter: The power inverter field.
        yaw_direction: The yaw direction field.
        yaw_error: The yaw error field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Nacelle", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    acc_from_back_side_x: Union[TimeSeriesWrite, str, None] = None
    acc_from_back_side_y: Union[TimeSeriesWrite, str, None] = None
    acc_from_back_side_z: Union[TimeSeriesWrite, str, None] = None
    gearbox: Union[GearboxWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    generator: Union[GeneratorWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    high_speed_shaft: Union[HighSpeedShaftWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    main_shaft: Union[MainShaftWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    power_inverter: Union[PowerInverterWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    yaw_direction: Union[TimeSeriesWrite, str, None] = None
    yaw_error: Union[TimeSeriesWrite, str, None] = None

    @field_validator("gearbox", "generator", "high_speed_shaft", "main_shaft", "power_inverter", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.acc_from_back_side_x is not None or write_none:
            properties["acc_from_back_side_x"] = (
                self.acc_from_back_side_x
                if isinstance(self.acc_from_back_side_x, str) or self.acc_from_back_side_x is None
                else self.acc_from_back_side_x.external_id
            )

        if self.acc_from_back_side_y is not None or write_none:
            properties["acc_from_back_side_y"] = (
                self.acc_from_back_side_y
                if isinstance(self.acc_from_back_side_y, str) or self.acc_from_back_side_y is None
                else self.acc_from_back_side_y.external_id
            )

        if self.acc_from_back_side_z is not None or write_none:
            properties["acc_from_back_side_z"] = (
                self.acc_from_back_side_z
                if isinstance(self.acc_from_back_side_z, str) or self.acc_from_back_side_z is None
                else self.acc_from_back_side_z.external_id
            )

        if self.gearbox is not None:
            properties["gearbox"] = {
                "space": self.space if isinstance(self.gearbox, str) else self.gearbox.space,
                "externalId": self.gearbox if isinstance(self.gearbox, str) else self.gearbox.external_id,
            }

        if self.generator is not None:
            properties["generator"] = {
                "space": self.space if isinstance(self.generator, str) else self.generator.space,
                "externalId": self.generator if isinstance(self.generator, str) else self.generator.external_id,
            }

        if self.high_speed_shaft is not None:
            properties["high_speed_shaft"] = {
                "space": self.space if isinstance(self.high_speed_shaft, str) else self.high_speed_shaft.space,
                "externalId": (
                    self.high_speed_shaft
                    if isinstance(self.high_speed_shaft, str)
                    else self.high_speed_shaft.external_id
                ),
            }

        if self.main_shaft is not None:
            properties["main_shaft"] = {
                "space": self.space if isinstance(self.main_shaft, str) else self.main_shaft.space,
                "externalId": self.main_shaft if isinstance(self.main_shaft, str) else self.main_shaft.external_id,
            }

        if self.power_inverter is not None:
            properties["power_inverter"] = {
                "space": self.space if isinstance(self.power_inverter, str) else self.power_inverter.space,
                "externalId": (
                    self.power_inverter if isinstance(self.power_inverter, str) else self.power_inverter.external_id
                ),
            }

        if self.yaw_direction is not None or write_none:
            properties["yaw_direction"] = (
                self.yaw_direction
                if isinstance(self.yaw_direction, str) or self.yaw_direction is None
                else self.yaw_direction.external_id
            )

        if self.yaw_error is not None or write_none:
            properties["yaw_error"] = (
                self.yaw_error
                if isinstance(self.yaw_error, str) or self.yaw_error is None
                else self.yaw_error.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.gearbox, DomainModelWrite):
            other_resources = self.gearbox._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.generator, DomainModelWrite):
            other_resources = self.generator._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.high_speed_shaft, DomainModelWrite):
            other_resources = self.high_speed_shaft._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.main_shaft, DomainModelWrite):
            other_resources = self.main_shaft._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.power_inverter, DomainModelWrite):
            other_resources = self.power_inverter._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.acc_from_back_side_x, CogniteTimeSeriesWrite):
            resources.time_series.append(self.acc_from_back_side_x)

        if isinstance(self.acc_from_back_side_y, CogniteTimeSeriesWrite):
            resources.time_series.append(self.acc_from_back_side_y)

        if isinstance(self.acc_from_back_side_z, CogniteTimeSeriesWrite):
            resources.time_series.append(self.acc_from_back_side_z)

        if isinstance(self.yaw_direction, CogniteTimeSeriesWrite):
            resources.time_series.append(self.yaw_direction)

        if isinstance(self.yaw_error, CogniteTimeSeriesWrite):
            resources.time_series.append(self.yaw_error)

        return resources

NacelleWriteList

Bases: DomainModelWriteList[NacelleWrite]

List of nacelles in the writing version.

Source code in examples/windmill/data_classes/_nacelle.py

class NacelleWriteList(DomainModelWriteList[NacelleWrite]):
    """List of nacelles in the writing version."""

    _INSTANCE = NacelleWrite

PowerInverter

Bases: DomainModel

This represents the reading version of power inverter.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the power inverter.	required
data_record		The data record of the power inverter node.	required
active_power_total		The active power total field.	required
apparent_power_total		The apparent power total field.	required
reactive_power_total		The reactive power total field.	required

Source code in examples/windmill/data_classes/_power_inverter.py

class PowerInverter(DomainModel):
    """This represents the reading version of power inverter.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the power inverter.
        data_record: The data record of the power inverter node.
        active_power_total: The active power total field.
        apparent_power_total: The apparent power total field.
        reactive_power_total: The reactive power total field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "PowerInverter", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    active_power_total: Union[TimeSeries, str, None] = None
    apparent_power_total: Union[TimeSeries, str, None] = None
    reactive_power_total: Union[TimeSeries, str, None] = None

    def as_write(self) -> PowerInverterWrite:
        """Convert this read version of power inverter to the writing version."""
        return PowerInverterWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            active_power_total=(
                self.active_power_total.as_write()
                if isinstance(self.active_power_total, CogniteTimeSeries)
                else self.active_power_total
            ),
            apparent_power_total=(
                self.apparent_power_total.as_write()
                if isinstance(self.apparent_power_total, CogniteTimeSeries)
                else self.apparent_power_total
            ),
            reactive_power_total=(
                self.reactive_power_total.as_write()
                if isinstance(self.reactive_power_total, CogniteTimeSeries)
                else self.reactive_power_total
            ),
        )

    def as_apply(self) -> PowerInverterWrite:
        """Convert this read version of power inverter to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of power inverter to the writing version.

Source code in examples/windmill/data_classes/_power_inverter.py

def as_apply(self) -> PowerInverterWrite:
    """Convert this read version of power inverter to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of power inverter to the writing version.

Source code in examples/windmill/data_classes/_power_inverter.py

def as_write(self) -> PowerInverterWrite:
    """Convert this read version of power inverter to the writing version."""
    return PowerInverterWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        active_power_total=(
            self.active_power_total.as_write()
            if isinstance(self.active_power_total, CogniteTimeSeries)
            else self.active_power_total
        ),
        apparent_power_total=(
            self.apparent_power_total.as_write()
            if isinstance(self.apparent_power_total, CogniteTimeSeries)
            else self.apparent_power_total
        ),
        reactive_power_total=(
            self.reactive_power_total.as_write()
            if isinstance(self.reactive_power_total, CogniteTimeSeries)
            else self.reactive_power_total
        ),
    )

PowerInverterGraphQL

Bases: GraphQLCore

This represents the reading version of power inverter, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the power inverter.	required
data_record		The data record of the power inverter node.	required
active_power_total		The active power total field.	required
apparent_power_total		The apparent power total field.	required
reactive_power_total		The reactive power total field.	required

Source code in examples/windmill/data_classes/_power_inverter.py

class PowerInverterGraphQL(GraphQLCore):
    """This represents the reading version of power inverter, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the power inverter.
        data_record: The data record of the power inverter node.
        active_power_total: The active power total field.
        apparent_power_total: The apparent power total field.
        reactive_power_total: The reactive power total field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "PowerInverter", "1")
    active_power_total: Optional[TimeSeriesGraphQL] = None
    apparent_power_total: Optional[TimeSeriesGraphQL] = None
    reactive_power_total: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> PowerInverter:
        """Convert this GraphQL format of power inverter to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return PowerInverter(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            active_power_total=self.active_power_total.as_read() if self.active_power_total else None,
            apparent_power_total=self.apparent_power_total.as_read() if self.apparent_power_total else None,
            reactive_power_total=self.reactive_power_total.as_read() if self.reactive_power_total else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> PowerInverterWrite:
        """Convert this GraphQL format of power inverter to the writing format."""
        return PowerInverterWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            active_power_total=self.active_power_total.as_write() if self.active_power_total else None,
            apparent_power_total=self.apparent_power_total.as_write() if self.apparent_power_total else None,
            reactive_power_total=self.reactive_power_total.as_write() if self.reactive_power_total else None,
        )

as_read()

Convert this GraphQL format of power inverter to the reading format.

Source code in examples/windmill/data_classes/_power_inverter.py

@no_type_check
def as_read(self) -> PowerInverter:
    """Convert this GraphQL format of power inverter to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return PowerInverter(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        active_power_total=self.active_power_total.as_read() if self.active_power_total else None,
        apparent_power_total=self.apparent_power_total.as_read() if self.apparent_power_total else None,
        reactive_power_total=self.reactive_power_total.as_read() if self.reactive_power_total else None,
    )

as_write()

Convert this GraphQL format of power inverter to the writing format.

Source code in examples/windmill/data_classes/_power_inverter.py

@no_type_check
def as_write(self) -> PowerInverterWrite:
    """Convert this GraphQL format of power inverter to the writing format."""
    return PowerInverterWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        active_power_total=self.active_power_total.as_write() if self.active_power_total else None,
        apparent_power_total=self.apparent_power_total.as_write() if self.apparent_power_total else None,
        reactive_power_total=self.reactive_power_total.as_write() if self.reactive_power_total else None,
    )

PowerInverterList

Bases: DomainModelList[PowerInverter]

List of power inverters in the read version.

Source code in examples/windmill/data_classes/_power_inverter.py

class PowerInverterList(DomainModelList[PowerInverter]):
    """List of power inverters in the read version."""

    _INSTANCE = PowerInverter

    def as_write(self) -> PowerInverterWriteList:
        """Convert these read versions of power inverter to the writing versions."""
        return PowerInverterWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> PowerInverterWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_power_inverter.py

def as_apply(self) -> PowerInverterWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of power inverter to the writing versions.

Source code in examples/windmill/data_classes/_power_inverter.py

def as_write(self) -> PowerInverterWriteList:
    """Convert these read versions of power inverter to the writing versions."""
    return PowerInverterWriteList([node.as_write() for node in self.data])

PowerInverterWrite

Bases: DomainModelWrite

This represents the writing version of power inverter.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the power inverter.	required
data_record		The data record of the power inverter node.	required
active_power_total		The active power total field.	required
apparent_power_total		The apparent power total field.	required
reactive_power_total		The reactive power total field.	required

Source code in examples/windmill/data_classes/_power_inverter.py

class PowerInverterWrite(DomainModelWrite):
    """This represents the writing version of power inverter.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the power inverter.
        data_record: The data record of the power inverter node.
        active_power_total: The active power total field.
        apparent_power_total: The apparent power total field.
        reactive_power_total: The reactive power total field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "PowerInverter", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    active_power_total: Union[TimeSeriesWrite, str, None] = None
    apparent_power_total: Union[TimeSeriesWrite, str, None] = None
    reactive_power_total: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.active_power_total is not None or write_none:
            properties["active_power_total"] = (
                self.active_power_total
                if isinstance(self.active_power_total, str) or self.active_power_total is None
                else self.active_power_total.external_id
            )

        if self.apparent_power_total is not None or write_none:
            properties["apparent_power_total"] = (
                self.apparent_power_total
                if isinstance(self.apparent_power_total, str) or self.apparent_power_total is None
                else self.apparent_power_total.external_id
            )

        if self.reactive_power_total is not None or write_none:
            properties["reactive_power_total"] = (
                self.reactive_power_total
                if isinstance(self.reactive_power_total, str) or self.reactive_power_total is None
                else self.reactive_power_total.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.active_power_total, CogniteTimeSeriesWrite):
            resources.time_series.append(self.active_power_total)

        if isinstance(self.apparent_power_total, CogniteTimeSeriesWrite):
            resources.time_series.append(self.apparent_power_total)

        if isinstance(self.reactive_power_total, CogniteTimeSeriesWrite):
            resources.time_series.append(self.reactive_power_total)

        return resources

PowerInverterWriteList

Bases: DomainModelWriteList[PowerInverterWrite]

List of power inverters in the writing version.

Source code in examples/windmill/data_classes/_power_inverter.py

class PowerInverterWriteList(DomainModelWriteList[PowerInverterWrite]):
    """List of power inverters in the writing version."""

    _INSTANCE = PowerInverterWrite

Rotor

Bases: DomainModel

This represents the reading version of rotor.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the rotor.	required
data_record		The data record of the rotor node.	required
rotor_speed_controller		The rotor speed controller field.	required
rpm_low_speed_shaft		The rpm low speed shaft field.	required

Source code in examples/windmill/data_classes/_rotor.py

class Rotor(DomainModel):
    """This represents the reading version of rotor.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the rotor.
        data_record: The data record of the rotor node.
        rotor_speed_controller: The rotor speed controller field.
        rpm_low_speed_shaft: The rpm low speed shaft field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Rotor", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    rotor_speed_controller: Union[TimeSeries, str, None] = None
    rpm_low_speed_shaft: Union[TimeSeries, str, None] = None

    def as_write(self) -> RotorWrite:
        """Convert this read version of rotor to the writing version."""
        return RotorWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            rotor_speed_controller=(
                self.rotor_speed_controller.as_write()
                if isinstance(self.rotor_speed_controller, CogniteTimeSeries)
                else self.rotor_speed_controller
            ),
            rpm_low_speed_shaft=(
                self.rpm_low_speed_shaft.as_write()
                if isinstance(self.rpm_low_speed_shaft, CogniteTimeSeries)
                else self.rpm_low_speed_shaft
            ),
        )

    def as_apply(self) -> RotorWrite:
        """Convert this read version of rotor to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of rotor to the writing version.

Source code in examples/windmill/data_classes/_rotor.py

def as_apply(self) -> RotorWrite:
    """Convert this read version of rotor to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of rotor to the writing version.

Source code in examples/windmill/data_classes/_rotor.py

def as_write(self) -> RotorWrite:
    """Convert this read version of rotor to the writing version."""
    return RotorWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        rotor_speed_controller=(
            self.rotor_speed_controller.as_write()
            if isinstance(self.rotor_speed_controller, CogniteTimeSeries)
            else self.rotor_speed_controller
        ),
        rpm_low_speed_shaft=(
            self.rpm_low_speed_shaft.as_write()
            if isinstance(self.rpm_low_speed_shaft, CogniteTimeSeries)
            else self.rpm_low_speed_shaft
        ),
    )

RotorGraphQL

Bases: GraphQLCore

This represents the reading version of rotor, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the rotor.	required
data_record		The data record of the rotor node.	required
rotor_speed_controller		The rotor speed controller field.	required
rpm_low_speed_shaft		The rpm low speed shaft field.	required

Source code in examples/windmill/data_classes/_rotor.py

class RotorGraphQL(GraphQLCore):
    """This represents the reading version of rotor, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the rotor.
        data_record: The data record of the rotor node.
        rotor_speed_controller: The rotor speed controller field.
        rpm_low_speed_shaft: The rpm low speed shaft field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Rotor", "1")
    rotor_speed_controller: Optional[TimeSeriesGraphQL] = None
    rpm_low_speed_shaft: Optional[TimeSeriesGraphQL] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Rotor:
        """Convert this GraphQL format of rotor to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Rotor(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            rotor_speed_controller=self.rotor_speed_controller.as_read() if self.rotor_speed_controller else None,
            rpm_low_speed_shaft=self.rpm_low_speed_shaft.as_read() if self.rpm_low_speed_shaft else None,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> RotorWrite:
        """Convert this GraphQL format of rotor to the writing format."""
        return RotorWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            rotor_speed_controller=self.rotor_speed_controller.as_write() if self.rotor_speed_controller else None,
            rpm_low_speed_shaft=self.rpm_low_speed_shaft.as_write() if self.rpm_low_speed_shaft else None,
        )

as_read()

Convert this GraphQL format of rotor to the reading format.

Source code in examples/windmill/data_classes/_rotor.py

@no_type_check
def as_read(self) -> Rotor:
    """Convert this GraphQL format of rotor to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Rotor(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        rotor_speed_controller=self.rotor_speed_controller.as_read() if self.rotor_speed_controller else None,
        rpm_low_speed_shaft=self.rpm_low_speed_shaft.as_read() if self.rpm_low_speed_shaft else None,
    )

as_write()

Convert this GraphQL format of rotor to the writing format.

Source code in examples/windmill/data_classes/_rotor.py

@no_type_check
def as_write(self) -> RotorWrite:
    """Convert this GraphQL format of rotor to the writing format."""
    return RotorWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        rotor_speed_controller=self.rotor_speed_controller.as_write() if self.rotor_speed_controller else None,
        rpm_low_speed_shaft=self.rpm_low_speed_shaft.as_write() if self.rpm_low_speed_shaft else None,
    )

RotorList

Bases: DomainModelList[Rotor]

List of rotors in the read version.

Source code in examples/windmill/data_classes/_rotor.py

class RotorList(DomainModelList[Rotor]):
    """List of rotors in the read version."""

    _INSTANCE = Rotor

    def as_write(self) -> RotorWriteList:
        """Convert these read versions of rotor to the writing versions."""
        return RotorWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> RotorWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_rotor.py

def as_apply(self) -> RotorWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of rotor to the writing versions.

Source code in examples/windmill/data_classes/_rotor.py

def as_write(self) -> RotorWriteList:
    """Convert these read versions of rotor to the writing versions."""
    return RotorWriteList([node.as_write() for node in self.data])

RotorWrite

Bases: DomainModelWrite

This represents the writing version of rotor.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the rotor.	required
data_record		The data record of the rotor node.	required
rotor_speed_controller		The rotor speed controller field.	required
rpm_low_speed_shaft		The rpm low speed shaft field.	required

Source code in examples/windmill/data_classes/_rotor.py

class RotorWrite(DomainModelWrite):
    """This represents the writing version of rotor.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the rotor.
        data_record: The data record of the rotor node.
        rotor_speed_controller: The rotor speed controller field.
        rpm_low_speed_shaft: The rpm low speed shaft field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Rotor", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    rotor_speed_controller: Union[TimeSeriesWrite, str, None] = None
    rpm_low_speed_shaft: Union[TimeSeriesWrite, str, None] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.rotor_speed_controller is not None or write_none:
            properties["rotor_speed_controller"] = (
                self.rotor_speed_controller
                if isinstance(self.rotor_speed_controller, str) or self.rotor_speed_controller is None
                else self.rotor_speed_controller.external_id
            )

        if self.rpm_low_speed_shaft is not None or write_none:
            properties["rpm_low_speed_shaft"] = (
                self.rpm_low_speed_shaft
                if isinstance(self.rpm_low_speed_shaft, str) or self.rpm_low_speed_shaft is None
                else self.rpm_low_speed_shaft.external_id
            )

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.rotor_speed_controller, CogniteTimeSeriesWrite):
            resources.time_series.append(self.rotor_speed_controller)

        if isinstance(self.rpm_low_speed_shaft, CogniteTimeSeriesWrite):
            resources.time_series.append(self.rpm_low_speed_shaft)

        return resources

RotorWriteList

Bases: DomainModelWriteList[RotorWrite]

List of rotors in the writing version.

Source code in examples/windmill/data_classes/_rotor.py

class RotorWriteList(DomainModelWriteList[RotorWrite]):
    """List of rotors in the writing version."""

    _INSTANCE = RotorWrite

SensorPosition

Bases: DomainModel

This represents the reading version of sensor position.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the sensor position.	required
data_record		The data record of the sensor position node.	required
edgewise_bend_mom_crosstalk_corrected		The edgewise bend mom crosstalk corrected field.	required
edgewise_bend_mom_offset		The edgewise bend mom offset field.	required
edgewise_bend_mom_offset_crosstalk_corrected		The edgewise bend mom offset crosstalk corrected field.	required
edgewisewise_bend_mom		The edgewisewise bend mom field.	required
flapwise_bend_mom		The flapwise bend mom field.	required
flapwise_bend_mom_crosstalk_corrected		The flapwise bend mom crosstalk corrected field.	required
flapwise_bend_mom_offset		The flapwise bend mom offset field.	required
flapwise_bend_mom_offset_crosstalk_corrected		The flapwise bend mom offset crosstalk corrected field.	required
position		The position field.	required

Source code in examples/windmill/data_classes/_sensor_position.py

class SensorPosition(DomainModel):
    """This represents the reading version of sensor position.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the sensor position.
        data_record: The data record of the sensor position node.
        edgewise_bend_mom_crosstalk_corrected: The edgewise bend mom crosstalk corrected field.
        edgewise_bend_mom_offset: The edgewise bend mom offset field.
        edgewise_bend_mom_offset_crosstalk_corrected: The edgewise bend mom offset crosstalk corrected field.
        edgewisewise_bend_mom: The edgewisewise bend mom field.
        flapwise_bend_mom: The flapwise bend mom field.
        flapwise_bend_mom_crosstalk_corrected: The flapwise bend mom crosstalk corrected field.
        flapwise_bend_mom_offset: The flapwise bend mom offset field.
        flapwise_bend_mom_offset_crosstalk_corrected: The flapwise bend mom offset crosstalk corrected field.
        position: The position field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "SensorPosition", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    edgewise_bend_mom_crosstalk_corrected: Union[TimeSeries, str, None] = None
    edgewise_bend_mom_offset: Union[TimeSeries, str, None] = None
    edgewise_bend_mom_offset_crosstalk_corrected: Union[TimeSeries, str, None] = None
    edgewisewise_bend_mom: Union[TimeSeries, str, None] = None
    flapwise_bend_mom: Union[TimeSeries, str, None] = None
    flapwise_bend_mom_crosstalk_corrected: Union[TimeSeries, str, None] = None
    flapwise_bend_mom_offset: Union[TimeSeries, str, None] = None
    flapwise_bend_mom_offset_crosstalk_corrected: Union[TimeSeries, str, None] = None
    position: Optional[float] = None

    def as_write(self) -> SensorPositionWrite:
        """Convert this read version of sensor position to the writing version."""
        return SensorPositionWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            edgewise_bend_mom_crosstalk_corrected=(
                self.edgewise_bend_mom_crosstalk_corrected.as_write()
                if isinstance(self.edgewise_bend_mom_crosstalk_corrected, CogniteTimeSeries)
                else self.edgewise_bend_mom_crosstalk_corrected
            ),
            edgewise_bend_mom_offset=(
                self.edgewise_bend_mom_offset.as_write()
                if isinstance(self.edgewise_bend_mom_offset, CogniteTimeSeries)
                else self.edgewise_bend_mom_offset
            ),
            edgewise_bend_mom_offset_crosstalk_corrected=(
                self.edgewise_bend_mom_offset_crosstalk_corrected.as_write()
                if isinstance(self.edgewise_bend_mom_offset_crosstalk_corrected, CogniteTimeSeries)
                else self.edgewise_bend_mom_offset_crosstalk_corrected
            ),
            edgewisewise_bend_mom=(
                self.edgewisewise_bend_mom.as_write()
                if isinstance(self.edgewisewise_bend_mom, CogniteTimeSeries)
                else self.edgewisewise_bend_mom
            ),
            flapwise_bend_mom=(
                self.flapwise_bend_mom.as_write()
                if isinstance(self.flapwise_bend_mom, CogniteTimeSeries)
                else self.flapwise_bend_mom
            ),
            flapwise_bend_mom_crosstalk_corrected=(
                self.flapwise_bend_mom_crosstalk_corrected.as_write()
                if isinstance(self.flapwise_bend_mom_crosstalk_corrected, CogniteTimeSeries)
                else self.flapwise_bend_mom_crosstalk_corrected
            ),
            flapwise_bend_mom_offset=(
                self.flapwise_bend_mom_offset.as_write()
                if isinstance(self.flapwise_bend_mom_offset, CogniteTimeSeries)
                else self.flapwise_bend_mom_offset
            ),
            flapwise_bend_mom_offset_crosstalk_corrected=(
                self.flapwise_bend_mom_offset_crosstalk_corrected.as_write()
                if isinstance(self.flapwise_bend_mom_offset_crosstalk_corrected, CogniteTimeSeries)
                else self.flapwise_bend_mom_offset_crosstalk_corrected
            ),
            position=self.position,
        )

    def as_apply(self) -> SensorPositionWrite:
        """Convert this read version of sensor position to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert this read version of sensor position to the writing version.

Source code in examples/windmill/data_classes/_sensor_position.py

def as_apply(self) -> SensorPositionWrite:
    """Convert this read version of sensor position to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of sensor position to the writing version.

Source code in examples/windmill/data_classes/_sensor_position.py

def as_write(self) -> SensorPositionWrite:
    """Convert this read version of sensor position to the writing version."""
    return SensorPositionWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        edgewise_bend_mom_crosstalk_corrected=(
            self.edgewise_bend_mom_crosstalk_corrected.as_write()
            if isinstance(self.edgewise_bend_mom_crosstalk_corrected, CogniteTimeSeries)
            else self.edgewise_bend_mom_crosstalk_corrected
        ),
        edgewise_bend_mom_offset=(
            self.edgewise_bend_mom_offset.as_write()
            if isinstance(self.edgewise_bend_mom_offset, CogniteTimeSeries)
            else self.edgewise_bend_mom_offset
        ),
        edgewise_bend_mom_offset_crosstalk_corrected=(
            self.edgewise_bend_mom_offset_crosstalk_corrected.as_write()
            if isinstance(self.edgewise_bend_mom_offset_crosstalk_corrected, CogniteTimeSeries)
            else self.edgewise_bend_mom_offset_crosstalk_corrected
        ),
        edgewisewise_bend_mom=(
            self.edgewisewise_bend_mom.as_write()
            if isinstance(self.edgewisewise_bend_mom, CogniteTimeSeries)
            else self.edgewisewise_bend_mom
        ),
        flapwise_bend_mom=(
            self.flapwise_bend_mom.as_write()
            if isinstance(self.flapwise_bend_mom, CogniteTimeSeries)
            else self.flapwise_bend_mom
        ),
        flapwise_bend_mom_crosstalk_corrected=(
            self.flapwise_bend_mom_crosstalk_corrected.as_write()
            if isinstance(self.flapwise_bend_mom_crosstalk_corrected, CogniteTimeSeries)
            else self.flapwise_bend_mom_crosstalk_corrected
        ),
        flapwise_bend_mom_offset=(
            self.flapwise_bend_mom_offset.as_write()
            if isinstance(self.flapwise_bend_mom_offset, CogniteTimeSeries)
            else self.flapwise_bend_mom_offset
        ),
        flapwise_bend_mom_offset_crosstalk_corrected=(
            self.flapwise_bend_mom_offset_crosstalk_corrected.as_write()
            if isinstance(self.flapwise_bend_mom_offset_crosstalk_corrected, CogniteTimeSeries)
            else self.flapwise_bend_mom_offset_crosstalk_corrected
        ),
        position=self.position,
    )

SensorPositionGraphQL

Bases: GraphQLCore

This represents the reading version of sensor position, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the sensor position.	required
data_record		The data record of the sensor position node.	required
edgewise_bend_mom_crosstalk_corrected		The edgewise bend mom crosstalk corrected field.	required
edgewise_bend_mom_offset		The edgewise bend mom offset field.	required
edgewise_bend_mom_offset_crosstalk_corrected		The edgewise bend mom offset crosstalk corrected field.	required
edgewisewise_bend_mom		The edgewisewise bend mom field.	required
flapwise_bend_mom		The flapwise bend mom field.	required
flapwise_bend_mom_crosstalk_corrected		The flapwise bend mom crosstalk corrected field.	required
flapwise_bend_mom_offset		The flapwise bend mom offset field.	required
flapwise_bend_mom_offset_crosstalk_corrected		The flapwise bend mom offset crosstalk corrected field.	required
position		The position field.	required

Source code in examples/windmill/data_classes/_sensor_position.py

class SensorPositionGraphQL(GraphQLCore):
    """This represents the reading version of sensor position, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the sensor position.
        data_record: The data record of the sensor position node.
        edgewise_bend_mom_crosstalk_corrected: The edgewise bend mom crosstalk corrected field.
        edgewise_bend_mom_offset: The edgewise bend mom offset field.
        edgewise_bend_mom_offset_crosstalk_corrected: The edgewise bend mom offset crosstalk corrected field.
        edgewisewise_bend_mom: The edgewisewise bend mom field.
        flapwise_bend_mom: The flapwise bend mom field.
        flapwise_bend_mom_crosstalk_corrected: The flapwise bend mom crosstalk corrected field.
        flapwise_bend_mom_offset: The flapwise bend mom offset field.
        flapwise_bend_mom_offset_crosstalk_corrected: The flapwise bend mom offset crosstalk corrected field.
        position: The position field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "SensorPosition", "1")
    edgewise_bend_mom_crosstalk_corrected: Optional[TimeSeriesGraphQL] = None
    edgewise_bend_mom_offset: Optional[TimeSeriesGraphQL] = None
    edgewise_bend_mom_offset_crosstalk_corrected: Optional[TimeSeriesGraphQL] = None
    edgewisewise_bend_mom: Optional[TimeSeriesGraphQL] = None
    flapwise_bend_mom: Optional[TimeSeriesGraphQL] = None
    flapwise_bend_mom_crosstalk_corrected: Optional[TimeSeriesGraphQL] = None
    flapwise_bend_mom_offset: Optional[TimeSeriesGraphQL] = None
    flapwise_bend_mom_offset_crosstalk_corrected: Optional[TimeSeriesGraphQL] = None
    position: Optional[float] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> SensorPosition:
        """Convert this GraphQL format of sensor position to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return SensorPosition(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            edgewise_bend_mom_crosstalk_corrected=(
                self.edgewise_bend_mom_crosstalk_corrected.as_read()
                if self.edgewise_bend_mom_crosstalk_corrected
                else None
            ),
            edgewise_bend_mom_offset=self.edgewise_bend_mom_offset.as_read() if self.edgewise_bend_mom_offset else None,
            edgewise_bend_mom_offset_crosstalk_corrected=(
                self.edgewise_bend_mom_offset_crosstalk_corrected.as_read()
                if self.edgewise_bend_mom_offset_crosstalk_corrected
                else None
            ),
            edgewisewise_bend_mom=self.edgewisewise_bend_mom.as_read() if self.edgewisewise_bend_mom else None,
            flapwise_bend_mom=self.flapwise_bend_mom.as_read() if self.flapwise_bend_mom else None,
            flapwise_bend_mom_crosstalk_corrected=(
                self.flapwise_bend_mom_crosstalk_corrected.as_read()
                if self.flapwise_bend_mom_crosstalk_corrected
                else None
            ),
            flapwise_bend_mom_offset=self.flapwise_bend_mom_offset.as_read() if self.flapwise_bend_mom_offset else None,
            flapwise_bend_mom_offset_crosstalk_corrected=(
                self.flapwise_bend_mom_offset_crosstalk_corrected.as_read()
                if self.flapwise_bend_mom_offset_crosstalk_corrected
                else None
            ),
            position=self.position,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> SensorPositionWrite:
        """Convert this GraphQL format of sensor position to the writing format."""
        return SensorPositionWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            edgewise_bend_mom_crosstalk_corrected=(
                self.edgewise_bend_mom_crosstalk_corrected.as_write()
                if self.edgewise_bend_mom_crosstalk_corrected
                else None
            ),
            edgewise_bend_mom_offset=(
                self.edgewise_bend_mom_offset.as_write() if self.edgewise_bend_mom_offset else None
            ),
            edgewise_bend_mom_offset_crosstalk_corrected=(
                self.edgewise_bend_mom_offset_crosstalk_corrected.as_write()
                if self.edgewise_bend_mom_offset_crosstalk_corrected
                else None
            ),
            edgewisewise_bend_mom=self.edgewisewise_bend_mom.as_write() if self.edgewisewise_bend_mom else None,
            flapwise_bend_mom=self.flapwise_bend_mom.as_write() if self.flapwise_bend_mom else None,
            flapwise_bend_mom_crosstalk_corrected=(
                self.flapwise_bend_mom_crosstalk_corrected.as_write()
                if self.flapwise_bend_mom_crosstalk_corrected
                else None
            ),
            flapwise_bend_mom_offset=(
                self.flapwise_bend_mom_offset.as_write() if self.flapwise_bend_mom_offset else None
            ),
            flapwise_bend_mom_offset_crosstalk_corrected=(
                self.flapwise_bend_mom_offset_crosstalk_corrected.as_write()
                if self.flapwise_bend_mom_offset_crosstalk_corrected
                else None
            ),
            position=self.position,
        )

as_read()

Convert this GraphQL format of sensor position to the reading format.

Source code in examples/windmill/data_classes/_sensor_position.py

@no_type_check
def as_read(self) -> SensorPosition:
    """Convert this GraphQL format of sensor position to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return SensorPosition(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        edgewise_bend_mom_crosstalk_corrected=(
            self.edgewise_bend_mom_crosstalk_corrected.as_read()
            if self.edgewise_bend_mom_crosstalk_corrected
            else None
        ),
        edgewise_bend_mom_offset=self.edgewise_bend_mom_offset.as_read() if self.edgewise_bend_mom_offset else None,
        edgewise_bend_mom_offset_crosstalk_corrected=(
            self.edgewise_bend_mom_offset_crosstalk_corrected.as_read()
            if self.edgewise_bend_mom_offset_crosstalk_corrected
            else None
        ),
        edgewisewise_bend_mom=self.edgewisewise_bend_mom.as_read() if self.edgewisewise_bend_mom else None,
        flapwise_bend_mom=self.flapwise_bend_mom.as_read() if self.flapwise_bend_mom else None,
        flapwise_bend_mom_crosstalk_corrected=(
            self.flapwise_bend_mom_crosstalk_corrected.as_read()
            if self.flapwise_bend_mom_crosstalk_corrected
            else None
        ),
        flapwise_bend_mom_offset=self.flapwise_bend_mom_offset.as_read() if self.flapwise_bend_mom_offset else None,
        flapwise_bend_mom_offset_crosstalk_corrected=(
            self.flapwise_bend_mom_offset_crosstalk_corrected.as_read()
            if self.flapwise_bend_mom_offset_crosstalk_corrected
            else None
        ),
        position=self.position,
    )

as_write()

Convert this GraphQL format of sensor position to the writing format.

Source code in examples/windmill/data_classes/_sensor_position.py

@no_type_check
def as_write(self) -> SensorPositionWrite:
    """Convert this GraphQL format of sensor position to the writing format."""
    return SensorPositionWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        edgewise_bend_mom_crosstalk_corrected=(
            self.edgewise_bend_mom_crosstalk_corrected.as_write()
            if self.edgewise_bend_mom_crosstalk_corrected
            else None
        ),
        edgewise_bend_mom_offset=(
            self.edgewise_bend_mom_offset.as_write() if self.edgewise_bend_mom_offset else None
        ),
        edgewise_bend_mom_offset_crosstalk_corrected=(
            self.edgewise_bend_mom_offset_crosstalk_corrected.as_write()
            if self.edgewise_bend_mom_offset_crosstalk_corrected
            else None
        ),
        edgewisewise_bend_mom=self.edgewisewise_bend_mom.as_write() if self.edgewisewise_bend_mom else None,
        flapwise_bend_mom=self.flapwise_bend_mom.as_write() if self.flapwise_bend_mom else None,
        flapwise_bend_mom_crosstalk_corrected=(
            self.flapwise_bend_mom_crosstalk_corrected.as_write()
            if self.flapwise_bend_mom_crosstalk_corrected
            else None
        ),
        flapwise_bend_mom_offset=(
            self.flapwise_bend_mom_offset.as_write() if self.flapwise_bend_mom_offset else None
        ),
        flapwise_bend_mom_offset_crosstalk_corrected=(
            self.flapwise_bend_mom_offset_crosstalk_corrected.as_write()
            if self.flapwise_bend_mom_offset_crosstalk_corrected
            else None
        ),
        position=self.position,
    )

SensorPositionList

Bases: DomainModelList[SensorPosition]

List of sensor positions in the read version.

Source code in examples/windmill/data_classes/_sensor_position.py

class SensorPositionList(DomainModelList[SensorPosition]):
    """List of sensor positions in the read version."""

    _INSTANCE = SensorPosition

    def as_write(self) -> SensorPositionWriteList:
        """Convert these read versions of sensor position to the writing versions."""
        return SensorPositionWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> SensorPositionWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_sensor_position.py

def as_apply(self) -> SensorPositionWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of sensor position to the writing versions.

Source code in examples/windmill/data_classes/_sensor_position.py

def as_write(self) -> SensorPositionWriteList:
    """Convert these read versions of sensor position to the writing versions."""
    return SensorPositionWriteList([node.as_write() for node in self.data])

SensorPositionWrite

Bases: DomainModelWrite

This represents the writing version of sensor position.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the sensor position.	required
data_record		The data record of the sensor position node.	required
edgewise_bend_mom_crosstalk_corrected		The edgewise bend mom crosstalk corrected field.	required
edgewise_bend_mom_offset		The edgewise bend mom offset field.	required
edgewise_bend_mom_offset_crosstalk_corrected		The edgewise bend mom offset crosstalk corrected field.	required
edgewisewise_bend_mom		The edgewisewise bend mom field.	required
flapwise_bend_mom		The flapwise bend mom field.	required
flapwise_bend_mom_crosstalk_corrected		The flapwise bend mom crosstalk corrected field.	required
flapwise_bend_mom_offset		The flapwise bend mom offset field.	required
flapwise_bend_mom_offset_crosstalk_corrected		The flapwise bend mom offset crosstalk corrected field.	required
position		The position field.	required

Source code in examples/windmill/data_classes/_sensor_position.py

class SensorPositionWrite(DomainModelWrite):
    """This represents the writing version of sensor position.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the sensor position.
        data_record: The data record of the sensor position node.
        edgewise_bend_mom_crosstalk_corrected: The edgewise bend mom crosstalk corrected field.
        edgewise_bend_mom_offset: The edgewise bend mom offset field.
        edgewise_bend_mom_offset_crosstalk_corrected: The edgewise bend mom offset crosstalk corrected field.
        edgewisewise_bend_mom: The edgewisewise bend mom field.
        flapwise_bend_mom: The flapwise bend mom field.
        flapwise_bend_mom_crosstalk_corrected: The flapwise bend mom crosstalk corrected field.
        flapwise_bend_mom_offset: The flapwise bend mom offset field.
        flapwise_bend_mom_offset_crosstalk_corrected: The flapwise bend mom offset crosstalk corrected field.
        position: The position field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "SensorPosition", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    edgewise_bend_mom_crosstalk_corrected: Union[TimeSeriesWrite, str, None] = None
    edgewise_bend_mom_offset: Union[TimeSeriesWrite, str, None] = None
    edgewise_bend_mom_offset_crosstalk_corrected: Union[TimeSeriesWrite, str, None] = None
    edgewisewise_bend_mom: Union[TimeSeriesWrite, str, None] = None
    flapwise_bend_mom: Union[TimeSeriesWrite, str, None] = None
    flapwise_bend_mom_crosstalk_corrected: Union[TimeSeriesWrite, str, None] = None
    flapwise_bend_mom_offset: Union[TimeSeriesWrite, str, None] = None
    flapwise_bend_mom_offset_crosstalk_corrected: Union[TimeSeriesWrite, str, None] = None
    position: Optional[float] = None

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.edgewise_bend_mom_crosstalk_corrected is not None or write_none:
            properties["edgewise_bend_mom_crosstalk_corrected"] = (
                self.edgewise_bend_mom_crosstalk_corrected
                if isinstance(self.edgewise_bend_mom_crosstalk_corrected, str)
                or self.edgewise_bend_mom_crosstalk_corrected is None
                else self.edgewise_bend_mom_crosstalk_corrected.external_id
            )

        if self.edgewise_bend_mom_offset is not None or write_none:
            properties["edgewise_bend_mom_offset"] = (
                self.edgewise_bend_mom_offset
                if isinstance(self.edgewise_bend_mom_offset, str) or self.edgewise_bend_mom_offset is None
                else self.edgewise_bend_mom_offset.external_id
            )

        if self.edgewise_bend_mom_offset_crosstalk_corrected is not None or write_none:
            properties["edgewise_bend_mom_offset_crosstalk_corrected"] = (
                self.edgewise_bend_mom_offset_crosstalk_corrected
                if isinstance(self.edgewise_bend_mom_offset_crosstalk_corrected, str)
                or self.edgewise_bend_mom_offset_crosstalk_corrected is None
                else self.edgewise_bend_mom_offset_crosstalk_corrected.external_id
            )

        if self.edgewisewise_bend_mom is not None or write_none:
            properties["edgewisewise_bend_mom"] = (
                self.edgewisewise_bend_mom
                if isinstance(self.edgewisewise_bend_mom, str) or self.edgewisewise_bend_mom is None
                else self.edgewisewise_bend_mom.external_id
            )

        if self.flapwise_bend_mom is not None or write_none:
            properties["flapwise_bend_mom"] = (
                self.flapwise_bend_mom
                if isinstance(self.flapwise_bend_mom, str) or self.flapwise_bend_mom is None
                else self.flapwise_bend_mom.external_id
            )

        if self.flapwise_bend_mom_crosstalk_corrected is not None or write_none:
            properties["flapwise_bend_mom_crosstalk_corrected"] = (
                self.flapwise_bend_mom_crosstalk_corrected
                if isinstance(self.flapwise_bend_mom_crosstalk_corrected, str)
                or self.flapwise_bend_mom_crosstalk_corrected is None
                else self.flapwise_bend_mom_crosstalk_corrected.external_id
            )

        if self.flapwise_bend_mom_offset is not None or write_none:
            properties["flapwise_bend_mom_offset"] = (
                self.flapwise_bend_mom_offset
                if isinstance(self.flapwise_bend_mom_offset, str) or self.flapwise_bend_mom_offset is None
                else self.flapwise_bend_mom_offset.external_id
            )

        if self.flapwise_bend_mom_offset_crosstalk_corrected is not None or write_none:
            properties["flapwise_bend_mom_offset_crosstalk_corrected"] = (
                self.flapwise_bend_mom_offset_crosstalk_corrected
                if isinstance(self.flapwise_bend_mom_offset_crosstalk_corrected, str)
                or self.flapwise_bend_mom_offset_crosstalk_corrected is None
                else self.flapwise_bend_mom_offset_crosstalk_corrected.external_id
            )

        if self.position is not None or write_none:
            properties["position"] = self.position

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        if isinstance(self.edgewise_bend_mom_crosstalk_corrected, CogniteTimeSeriesWrite):
            resources.time_series.append(self.edgewise_bend_mom_crosstalk_corrected)

        if isinstance(self.edgewise_bend_mom_offset, CogniteTimeSeriesWrite):
            resources.time_series.append(self.edgewise_bend_mom_offset)

        if isinstance(self.edgewise_bend_mom_offset_crosstalk_corrected, CogniteTimeSeriesWrite):
            resources.time_series.append(self.edgewise_bend_mom_offset_crosstalk_corrected)

        if isinstance(self.edgewisewise_bend_mom, CogniteTimeSeriesWrite):
            resources.time_series.append(self.edgewisewise_bend_mom)

        if isinstance(self.flapwise_bend_mom, CogniteTimeSeriesWrite):
            resources.time_series.append(self.flapwise_bend_mom)

        if isinstance(self.flapwise_bend_mom_crosstalk_corrected, CogniteTimeSeriesWrite):
            resources.time_series.append(self.flapwise_bend_mom_crosstalk_corrected)

        if isinstance(self.flapwise_bend_mom_offset, CogniteTimeSeriesWrite):
            resources.time_series.append(self.flapwise_bend_mom_offset)

        if isinstance(self.flapwise_bend_mom_offset_crosstalk_corrected, CogniteTimeSeriesWrite):
            resources.time_series.append(self.flapwise_bend_mom_offset_crosstalk_corrected)

        return resources

SensorPositionWriteList

Bases: DomainModelWriteList[SensorPositionWrite]

List of sensor positions in the writing version.

Source code in examples/windmill/data_classes/_sensor_position.py

class SensorPositionWriteList(DomainModelWriteList[SensorPositionWrite]):
    """List of sensor positions in the writing version."""

    _INSTANCE = SensorPositionWrite

Windmill

Bases: DomainModel

This represents the reading version of windmill.

It is used to when data is retrieved from CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the windmill.	required
data_record		The data record of the windmill node.	required
blades		The blade field.	required
capacity		The capacity field.	required
metmast		The metmast field.	required
nacelle		The nacelle field.	required
name		The name field.	required
rotor		The rotor field.	required
windfarm		The windfarm field.	required

Source code in examples/windmill/data_classes/_windmill.py

class Windmill(DomainModel):
    """This represents the reading version of windmill.

    It is used to when data is retrieved from CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the windmill.
        data_record: The data record of the windmill node.
        blades: The blade field.
        capacity: The capacity field.
        metmast: The metmast field.
        nacelle: The nacelle field.
        name: The name field.
        rotor: The rotor field.
        windfarm: The windfarm field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Windmill", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, None] = None
    blades: Optional[list[Union[Blade, str, dm.NodeId]]] = Field(default=None, repr=False)
    capacity: Optional[float] = None
    metmast: Optional[list[Union[Metmast, str, dm.NodeId]]] = Field(default=None, repr=False)
    nacelle: Union[Nacelle, str, dm.NodeId, None] = Field(default=None, repr=False)
    name: Optional[str] = None
    rotor: Union[Rotor, str, dm.NodeId, None] = Field(default=None, repr=False)
    windfarm: Optional[str] = None

    def as_write(self) -> WindmillWrite:
        """Convert this read version of windmill to the writing version."""
        return WindmillWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=self.data_record.version),
            blades=[blade.as_write() if isinstance(blade, DomainModel) else blade for blade in self.blades or []],
            capacity=self.capacity,
            metmast=[
                metmast.as_write() if isinstance(metmast, DomainModel) else metmast for metmast in self.metmast or []
            ],
            nacelle=self.nacelle.as_write() if isinstance(self.nacelle, DomainModel) else self.nacelle,
            name=self.name,
            rotor=self.rotor.as_write() if isinstance(self.rotor, DomainModel) else self.rotor,
            windfarm=self.windfarm,
        )

    def as_apply(self) -> WindmillWrite:
        """Convert this read version of windmill to the writing version."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

    @classmethod
    def _update_connections(
        cls,
        instances: dict[dm.NodeId | str, Windmill],  # type: ignore[override]
        nodes_by_id: dict[dm.NodeId | str, DomainModel],
        edges_by_source_node: dict[dm.NodeId, list[dm.Edge | DomainRelation]],
    ) -> None:
        from ._blade import Blade
        from ._metmast import Metmast
        from ._nacelle import Nacelle
        from ._rotor import Rotor

        for instance in instances.values():
            if (
                isinstance(instance.nacelle, (dm.NodeId, str))
                and (nacelle := nodes_by_id.get(instance.nacelle))
                and isinstance(nacelle, Nacelle)
            ):
                instance.nacelle = nacelle
            if (
                isinstance(instance.rotor, (dm.NodeId, str))
                and (rotor := nodes_by_id.get(instance.rotor))
                and isinstance(rotor, Rotor)
            ):
                instance.rotor = rotor
            if edges := edges_by_source_node.get(instance.as_id()):
                blades: list[Blade | str | dm.NodeId] = []
                metmast: list[Metmast | str | dm.NodeId] = []
                for edge in edges:
                    value: DomainModel | DomainRelation | str | dm.NodeId
                    if isinstance(edge, DomainRelation):
                        value = edge
                    else:
                        other_end: dm.DirectRelationReference = (
                            edge.end_node
                            if edge.start_node.space == instance.space
                            and edge.start_node.external_id == instance.external_id
                            else edge.start_node
                        )
                        destination: dm.NodeId | str = (
                            as_node_id(other_end)
                            if other_end.space != DEFAULT_INSTANCE_SPACE
                            else other_end.external_id
                        )
                        if destination in nodes_by_id:
                            value = nodes_by_id[destination]
                        else:
                            value = destination
                    edge_type = edge.edge_type if isinstance(edge, DomainRelation) else edge.type

                    if edge_type == dm.DirectRelationReference("power-models", "Windmill.blades") and isinstance(
                        value, (Blade, str, dm.NodeId)
                    ):
                        blades.append(value)
                    if edge_type == dm.DirectRelationReference("power-models", "Windmill.metmast") and isinstance(
                        value, (Metmast, str, dm.NodeId)
                    ):
                        metmast.append(value)

                instance.blades = blades or None
                instance.metmast = metmast or None

as_apply()

Convert this read version of windmill to the writing version.

Source code in examples/windmill/data_classes/_windmill.py

def as_apply(self) -> WindmillWrite:
    """Convert this read version of windmill to the writing version."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert this read version of windmill to the writing version.

Source code in examples/windmill/data_classes/_windmill.py

def as_write(self) -> WindmillWrite:
    """Convert this read version of windmill to the writing version."""
    return WindmillWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=self.data_record.version),
        blades=[blade.as_write() if isinstance(blade, DomainModel) else blade for blade in self.blades or []],
        capacity=self.capacity,
        metmast=[
            metmast.as_write() if isinstance(metmast, DomainModel) else metmast for metmast in self.metmast or []
        ],
        nacelle=self.nacelle.as_write() if isinstance(self.nacelle, DomainModel) else self.nacelle,
        name=self.name,
        rotor=self.rotor.as_write() if isinstance(self.rotor, DomainModel) else self.rotor,
        windfarm=self.windfarm,
    )

WindmillGraphQL

Bases: GraphQLCore

This represents the reading version of windmill, used when data is retrieved from CDF using GraphQL.

It is used when retrieving data from CDF using GraphQL.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the windmill.	required
data_record		The data record of the windmill node.	required
blades		The blade field.	required
capacity		The capacity field.	required
metmast		The metmast field.	required
nacelle		The nacelle field.	required
name		The name field.	required
rotor		The rotor field.	required
windfarm		The windfarm field.	required

Source code in examples/windmill/data_classes/_windmill.py

class WindmillGraphQL(GraphQLCore):
    """This represents the reading version of windmill, used
    when data is retrieved from CDF using GraphQL.

    It is used when retrieving data from CDF using GraphQL.

    Args:
        space: The space where the node is located.
        external_id: The external id of the windmill.
        data_record: The data record of the windmill node.
        blades: The blade field.
        capacity: The capacity field.
        metmast: The metmast field.
        nacelle: The nacelle field.
        name: The name field.
        rotor: The rotor field.
        windfarm: The windfarm field.
    """

    view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Windmill", "1")
    blades: Optional[list[BladeGraphQL]] = Field(default=None, repr=False)
    capacity: Optional[float] = None
    metmast: Optional[list[MetmastGraphQL]] = Field(default=None, repr=False)
    nacelle: Optional[NacelleGraphQL] = Field(default=None, repr=False)
    name: Optional[str] = None
    rotor: Optional[RotorGraphQL] = Field(default=None, repr=False)
    windfarm: Optional[str] = None

    @model_validator(mode="before")
    def parse_data_record(cls, values: Any) -> Any:
        if not isinstance(values, dict):
            return values
        if "lastUpdatedTime" in values or "createdTime" in values:
            values["dataRecord"] = DataRecordGraphQL(
                created_time=values.pop("createdTime", None),
                last_updated_time=values.pop("lastUpdatedTime", None),
            )
        return values

    @field_validator("blades", "metmast", "nacelle", "rotor", mode="before")
    def parse_graphql(cls, value: Any) -> Any:
        if not isinstance(value, dict):
            return value
        if "items" in value:
            return value["items"]
        return value

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_read(self) -> Windmill:
        """Convert this GraphQL format of windmill to the reading format."""
        if self.data_record is None:
            raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
        return Windmill(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecord(
                version=0,
                last_updated_time=self.data_record.last_updated_time,
                created_time=self.data_record.created_time,
            ),
            blades=[blade.as_read() for blade in self.blades or []],
            capacity=self.capacity,
            metmast=[metmast.as_read() for metmast in self.metmast or []],
            nacelle=self.nacelle.as_read() if isinstance(self.nacelle, GraphQLCore) else self.nacelle,
            name=self.name,
            rotor=self.rotor.as_read() if isinstance(self.rotor, GraphQLCore) else self.rotor,
            windfarm=self.windfarm,
        )

    # We do the ignore argument type as we let pydantic handle the type checking
    @no_type_check
    def as_write(self) -> WindmillWrite:
        """Convert this GraphQL format of windmill to the writing format."""
        return WindmillWrite(
            space=self.space,
            external_id=self.external_id,
            data_record=DataRecordWrite(existing_version=0),
            blades=[blade.as_write() for blade in self.blades or []],
            capacity=self.capacity,
            metmast=[metmast.as_write() for metmast in self.metmast or []],
            nacelle=self.nacelle.as_write() if isinstance(self.nacelle, GraphQLCore) else self.nacelle,
            name=self.name,
            rotor=self.rotor.as_write() if isinstance(self.rotor, GraphQLCore) else self.rotor,
            windfarm=self.windfarm,
        )

as_read()

Convert this GraphQL format of windmill to the reading format.

Source code in examples/windmill/data_classes/_windmill.py

@no_type_check
def as_read(self) -> Windmill:
    """Convert this GraphQL format of windmill to the reading format."""
    if self.data_record is None:
        raise ValueError("This object cannot be converted to a read format because it lacks a data record.")
    return Windmill(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecord(
            version=0,
            last_updated_time=self.data_record.last_updated_time,
            created_time=self.data_record.created_time,
        ),
        blades=[blade.as_read() for blade in self.blades or []],
        capacity=self.capacity,
        metmast=[metmast.as_read() for metmast in self.metmast or []],
        nacelle=self.nacelle.as_read() if isinstance(self.nacelle, GraphQLCore) else self.nacelle,
        name=self.name,
        rotor=self.rotor.as_read() if isinstance(self.rotor, GraphQLCore) else self.rotor,
        windfarm=self.windfarm,
    )

as_write()

Convert this GraphQL format of windmill to the writing format.

Source code in examples/windmill/data_classes/_windmill.py

@no_type_check
def as_write(self) -> WindmillWrite:
    """Convert this GraphQL format of windmill to the writing format."""
    return WindmillWrite(
        space=self.space,
        external_id=self.external_id,
        data_record=DataRecordWrite(existing_version=0),
        blades=[blade.as_write() for blade in self.blades or []],
        capacity=self.capacity,
        metmast=[metmast.as_write() for metmast in self.metmast or []],
        nacelle=self.nacelle.as_write() if isinstance(self.nacelle, GraphQLCore) else self.nacelle,
        name=self.name,
        rotor=self.rotor.as_write() if isinstance(self.rotor, GraphQLCore) else self.rotor,
        windfarm=self.windfarm,
    )

WindmillList

Bases: DomainModelList[Windmill]

List of windmills in the read version.

Source code in examples/windmill/data_classes/_windmill.py

class WindmillList(DomainModelList[Windmill]):
    """List of windmills in the read version."""

    _INSTANCE = Windmill

    def as_write(self) -> WindmillWriteList:
        """Convert these read versions of windmill to the writing versions."""
        return WindmillWriteList([node.as_write() for node in self.data])

    def as_apply(self) -> WindmillWriteList:
        """Convert these read versions of primitive nullable to the writing versions."""
        warnings.warn(
            "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
            UserWarning,
            stacklevel=2,
        )
        return self.as_write()

as_apply()

Convert these read versions of primitive nullable to the writing versions.

Source code in examples/windmill/data_classes/_windmill.py

def as_apply(self) -> WindmillWriteList:
    """Convert these read versions of primitive nullable to the writing versions."""
    warnings.warn(
        "as_apply is deprecated and will be removed in v1.0. Use as_write instead.",
        UserWarning,
        stacklevel=2,
    )
    return self.as_write()

as_write()

Convert these read versions of windmill to the writing versions.

Source code in examples/windmill/data_classes/_windmill.py

def as_write(self) -> WindmillWriteList:
    """Convert these read versions of windmill to the writing versions."""
    return WindmillWriteList([node.as_write() for node in self.data])

WindmillWrite

Bases: DomainModelWrite

This represents the writing version of windmill.

It is used to when data is sent to CDF.

Parameters:

Name	Type	Description	Default
space		The space where the node is located.	required
external_id		The external id of the windmill.	required
data_record		The data record of the windmill node.	required
blades		The blade field.	required
capacity		The capacity field.	required
metmast		The metmast field.	required
nacelle		The nacelle field.	required
name		The name field.	required
rotor		The rotor field.	required
windfarm		The windfarm field.	required

Source code in examples/windmill/data_classes/_windmill.py

class WindmillWrite(DomainModelWrite):
    """This represents the writing version of windmill.

    It is used to when data is sent to CDF.

    Args:
        space: The space where the node is located.
        external_id: The external id of the windmill.
        data_record: The data record of the windmill node.
        blades: The blade field.
        capacity: The capacity field.
        metmast: The metmast field.
        nacelle: The nacelle field.
        name: The name field.
        rotor: The rotor field.
        windfarm: The windfarm field.
    """

    _view_id: ClassVar[dm.ViewId] = dm.ViewId("power-models", "Windmill", "1")

    space: str = DEFAULT_INSTANCE_SPACE
    node_type: Union[dm.DirectRelationReference, dm.NodeId, tuple[str, str], None] = None
    blades: Optional[list[Union[BladeWrite, str, dm.NodeId]]] = Field(default=None, repr=False)
    capacity: Optional[float] = None
    metmast: Optional[list[Union[MetmastWrite, str, dm.NodeId]]] = Field(default=None, repr=False)
    nacelle: Union[NacelleWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    name: Optional[str] = None
    rotor: Union[RotorWrite, str, dm.NodeId, None] = Field(default=None, repr=False)
    windfarm: Optional[str] = None

    @field_validator("blades", "metmast", "nacelle", "rotor", mode="before")
    def as_node_id(cls, value: Any) -> Any:
        if isinstance(value, dm.DirectRelationReference):
            return dm.NodeId(value.space, value.external_id)
        elif isinstance(value, tuple) and len(value) == 2 and all(isinstance(item, str) for item in value):
            return dm.NodeId(value[0], value[1])
        elif isinstance(value, list):
            return [cls.as_node_id(item) for item in value]
        return value

    def _to_instances_write(
        self,
        cache: set[tuple[str, str]],
        write_none: bool = False,
        allow_version_increase: bool = False,
    ) -> ResourcesWrite:
        resources = ResourcesWrite()
        if self.as_tuple_id() in cache:
            return resources

        properties: dict[str, Any] = {}

        if self.capacity is not None or write_none:
            properties["capacity"] = self.capacity

        if self.nacelle is not None:
            properties["nacelle"] = {
                "space": self.space if isinstance(self.nacelle, str) else self.nacelle.space,
                "externalId": self.nacelle if isinstance(self.nacelle, str) else self.nacelle.external_id,
            }

        if self.name is not None or write_none:
            properties["name"] = self.name

        if self.rotor is not None:
            properties["rotor"] = {
                "space": self.space if isinstance(self.rotor, str) else self.rotor.space,
                "externalId": self.rotor if isinstance(self.rotor, str) else self.rotor.external_id,
            }

        if self.windfarm is not None or write_none:
            properties["windfarm"] = self.windfarm

        if properties:
            this_node = dm.NodeApply(
                space=self.space,
                external_id=self.external_id,
                existing_version=None if allow_version_increase else self.data_record.existing_version,
                type=as_direct_relation_reference(self.node_type),
                sources=[
                    dm.NodeOrEdgeData(
                        source=self._view_id,
                        properties=properties,
                    )
                ],
            )
            resources.nodes.append(this_node)
            cache.add(self.as_tuple_id())

        edge_type = dm.DirectRelationReference("power-models", "Windmill.blades")
        for blade in self.blades or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=blade,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        edge_type = dm.DirectRelationReference("power-models", "Windmill.metmast")
        for metmast in self.metmast or []:
            other_resources = DomainRelationWrite.from_edge_to_resources(
                cache,
                start_node=self,
                end_node=metmast,
                edge_type=edge_type,
                write_none=write_none,
                allow_version_increase=allow_version_increase,
            )
            resources.extend(other_resources)

        if isinstance(self.nacelle, DomainModelWrite):
            other_resources = self.nacelle._to_instances_write(cache)
            resources.extend(other_resources)

        if isinstance(self.rotor, DomainModelWrite):
            other_resources = self.rotor._to_instances_write(cache)
            resources.extend(other_resources)

        return resources

WindmillWriteList

Bases: DomainModelWriteList[WindmillWrite]

List of windmills in the writing version.

Source code in examples/windmill/data_classes/_windmill.py

class WindmillWriteList(DomainModelWriteList[WindmillWrite]):
    """List of windmills in the writing version."""

    _INSTANCE = WindmillWrite