Skip to content

flight module

Source code in d2spy/models/flight.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class Flight:
    id: UUID
    name: Optional[str]
    acquisition_date: date
    altitude: float
    side_overlap: float
    forward_overlap: float
    sensor: str
    platform: str
    is_active: bool
    deactivated_at: Optional[datetime]
    project_id: UUID
    pilot_id: UUID
    data_products: List[DataProduct]

    def __init__(self, client: APIClient, **kwargs):
        self.client = client
        # Flight attributes returned from API
        self.__dict__.update(kwargs)

    def __repr__(self):
        return (
            f"Flight(acquisition_date={self.acquisition_date!r}, name={self.name!r}, "
            f"altitude={self.altitude!r}, side_overlap={self.side_overlap!r}, "
            f"forward_overlap={self.forward_overlap!r}, sensor={self.sensor!r}, "
            f"platform={self.platform!r})"
        )

    def add_data_product(
        self,
        filepath: str,
        data_type: Union[Literal["dsm", "point_cloud", "ortho"], str],
    ) -> None:
        """Uploads data product to D2S. After the upload finishes, the data product may
        not be available for several minutes while it is processed on the D2S server. It
        will be returned by `Flight.get_data_products` once ready.

        Args:
            filepath (str): Full path to data product on local file system.
            data_type (Union[Literal["dsm", "point_cloud", "ortho"], str]): Data type.
        """
        verify_file_exists(filepath)
        validate_file_extension_and_data_type(filepath, data_type)
        # url for tusd server
        endpoint = f"{self.client.base_url}/files"
        # authorization cookie
        cookies = {"access_token": self.client.session.cookies["access_token"]}
        # project, flight, data type headers
        headers: Dict[str, str] = {
            "X-Project-ID": str(self.project_id),
            "X-Flight-ID": str(self.id),
            "X-Data-Type": data_type,
            "Accept-Language": "en-US,en;q=0.5",
            "Origin": self.client.base_url,
        }
        # metadata about data product file
        metadata = {
            "filename": Path(filepath).name,
            "filetype": get_metadata_filetype(filepath),
            "name": Path(filepath).name,
            "relativePath": "null",
            "type": get_metadata_filetype(filepath),
        }
        # create tus client and set headers and cookies
        tus_client = tusc.TusClient(endpoint)
        tus_client.set_headers(headers)
        tus_client.set_cookies(cookies)
        # create uploader for data product file with metadata
        chunk_size = 10 * 1024 * 1024  # 10 MiB
        tus_uploader = tus_client.uploader(
            filepath, chunk_size=chunk_size, metadata=metadata
        )
        # upload in chunks and print progress
        file_size = tus_uploader.get_file_size()
        while tus_uploader.offset < file_size:
            tus_uploader.upload_chunk()
            progress = (tus_uploader.offset / file_size) * 100
            print(f"Upload progress: {progress:.2f}%", end="\r")

    def add_raw_data(self, filepath: str) -> None:
        """Uploads zipped raw data to D2S. After the upload finishes, the raw data may
        not be available for several minutes while it is processed on the D2S server. It
        will be returned by `Flight.get_raw_data` once ready.

        Args:
            filepath (str): Full path to data product on local file system.
        """
        verify_file_exists(filepath)
        validate_file_extension_for_raw_data(filepath)
        # url for tusd server
        endpoint = f"{self.client.base_url}/files"
        # authorization cookie
        cookies = {"access_token": self.client.session.cookies["access_token"]}
        # project, flight, data type headers
        headers: Dict[str, str] = {
            "X-Project-ID": str(self.project_id),
            "X-Flight-ID": str(self.id),
            "X-Data-Type": "raw",
            "Accept-Language": "en-US,en;q=0.5",
            "Origin": self.client.base_url,
        }
        # metadata about raw data file
        metadata = {
            "filename": Path(filepath).name,
            "filetype": "application/zip",
            "name": Path(filepath).name,
            "relativePath": "null",
            "type": "application/zip",
        }
        # create tus client and set headers and cookies
        tus_client = tusc.TusClient(endpoint)
        tus_client.set_headers(headers)
        tus_client.set_cookies(cookies)
        # create uploader for raw file with metadata
        chunk_size = 10 * 1024 * 1024  # 10 MiB
        tus_uploader = tus_client.uploader(
            filepath, chunk_size=chunk_size, metadata=metadata
        )
        # upload in chunks and print progress
        file_size = tus_uploader.get_file_size()
        while tus_uploader.offset < file_size:
            tus_uploader.upload_chunk()
            progress = (tus_uploader.offset / file_size) * 100
            print(f"Upload progress: {progress:.2f}%", end="\r")

    def get_data_products(self) -> DataProductCollection:
        """Return list of all active data products in a flight.

        Returns:
            DataProductCollection: Collection of data products.
        """
        endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}/data_products"
        response_data = self.client.make_get_request(endpoint)

        data_products = [
            models.DataProduct(
                self.client,
                **schemas.DataProduct.from_dict(data_product).__dict__,
            )
            for data_product in response_data
        ]
        return DataProductCollection(collection=data_products)

    def get_raw_data(self) -> List[models.RawData]:
        """Return list of all active raw data in a flight.

        Returns:
            List[models.RawData]: List of raw data.
        """
        endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}/raw_data"
        response_data = self.client.make_get_request(endpoint)

        all_raw_data = [
            models.RawData(
                self.client,
                **schemas.RawData.from_dict(raw_data).__dict__,
            )
            for raw_data in response_data
        ]
        return all_raw_data

    def move_to_project(self, destination_project_id: UUID) -> None:
        """Moves flight from its current project to different project. You must be an
        owner of both the source project and destination project to make the transfer.

        Args:
            destination_project_id (UUID): ID of project the flight will be moved to.
        """
        endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}"
        endpoint += f"/move_to_project/{destination_project_id}"
        response_data = self.client.make_put_request(endpoint)

        updated_flight = schemas.Flight.from_dict(response_data).__dict__
        for key, value in updated_flight.items():
            if hasattr(self, key):
                setattr(self, key, value)
            else:
                print(f"Warning: Attribute '{key}' not found in Flight class.")
        return None

    def update(self, **kwargs) -> None:
        """Update flight attributes."""
        endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}"
        response_data = self.client.make_put_request(
            endpoint, json=json.loads(json.dumps(kwargs, default=str))
        )

        updated_flight = schemas.Flight.from_dict(response_data).__dict__
        for key, value in updated_flight.items():
            if hasattr(self, key):
                setattr(self, key, value)
            else:
                print(f"Warning: Attribute '{key}' not found in Flight class.")
        return None

add_data_product(filepath, data_type)

Uploads data product to D2S. After the upload finishes, the data product may not be available for several minutes while it is processed on the D2S server. It will be returned by Flight.get_data_products once ready.

Parameters:

Name Type Description Default
filepath str

Full path to data product on local file system.

 required
data_type Union[Literal['dsm', 'point_cloud', 'ortho'], str]

Data type.

 required
Source code in d2spy/models/flight.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def add_data_product(
    self,
    filepath: str,
    data_type: Union[Literal["dsm", "point_cloud", "ortho"], str],
) -> None:
    """Uploads data product to D2S. After the upload finishes, the data product may
    not be available for several minutes while it is processed on the D2S server. It
    will be returned by `Flight.get_data_products` once ready.

    Args:
        filepath (str): Full path to data product on local file system.
        data_type (Union[Literal["dsm", "point_cloud", "ortho"], str]): Data type.
    """
    verify_file_exists(filepath)
    validate_file_extension_and_data_type(filepath, data_type)
    # url for tusd server
    endpoint = f"{self.client.base_url}/files"
    # authorization cookie
    cookies = {"access_token": self.client.session.cookies["access_token"]}
    # project, flight, data type headers
    headers: Dict[str, str] = {
        "X-Project-ID": str(self.project_id),
        "X-Flight-ID": str(self.id),
        "X-Data-Type": data_type,
        "Accept-Language": "en-US,en;q=0.5",
        "Origin": self.client.base_url,
    }
    # metadata about data product file
    metadata = {
        "filename": Path(filepath).name,
        "filetype": get_metadata_filetype(filepath),
        "name": Path(filepath).name,
        "relativePath": "null",
        "type": get_metadata_filetype(filepath),
    }
    # create tus client and set headers and cookies
    tus_client = tusc.TusClient(endpoint)
    tus_client.set_headers(headers)
    tus_client.set_cookies(cookies)
    # create uploader for data product file with metadata
    chunk_size = 10 * 1024 * 1024  # 10 MiB
    tus_uploader = tus_client.uploader(
        filepath, chunk_size=chunk_size, metadata=metadata
    )
    # upload in chunks and print progress
    file_size = tus_uploader.get_file_size()
    while tus_uploader.offset < file_size:
        tus_uploader.upload_chunk()
        progress = (tus_uploader.offset / file_size) * 100
        print(f"Upload progress: {progress:.2f}%", end="\r")

add_raw_data(filepath)

Uploads zipped raw data to D2S. After the upload finishes, the raw data may not be available for several minutes while it is processed on the D2S server. It will be returned by Flight.get_raw_data once ready.

Parameters:

Name Type Description Default
filepath str

Full path to data product on local file system.

 required
Source code in d2spy/models/flight.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def add_raw_data(self, filepath: str) -> None:
    """Uploads zipped raw data to D2S. After the upload finishes, the raw data may
    not be available for several minutes while it is processed on the D2S server. It
    will be returned by `Flight.get_raw_data` once ready.

    Args:
        filepath (str): Full path to data product on local file system.
    """
    verify_file_exists(filepath)
    validate_file_extension_for_raw_data(filepath)
    # url for tusd server
    endpoint = f"{self.client.base_url}/files"
    # authorization cookie
    cookies = {"access_token": self.client.session.cookies["access_token"]}
    # project, flight, data type headers
    headers: Dict[str, str] = {
        "X-Project-ID": str(self.project_id),
        "X-Flight-ID": str(self.id),
        "X-Data-Type": "raw",
        "Accept-Language": "en-US,en;q=0.5",
        "Origin": self.client.base_url,
    }
    # metadata about raw data file
    metadata = {
        "filename": Path(filepath).name,
        "filetype": "application/zip",
        "name": Path(filepath).name,
        "relativePath": "null",
        "type": "application/zip",
    }
    # create tus client and set headers and cookies
    tus_client = tusc.TusClient(endpoint)
    tus_client.set_headers(headers)
    tus_client.set_cookies(cookies)
    # create uploader for raw file with metadata
    chunk_size = 10 * 1024 * 1024  # 10 MiB
    tus_uploader = tus_client.uploader(
        filepath, chunk_size=chunk_size, metadata=metadata
    )
    # upload in chunks and print progress
    file_size = tus_uploader.get_file_size()
    while tus_uploader.offset < file_size:
        tus_uploader.upload_chunk()
        progress = (tus_uploader.offset / file_size) * 100
        print(f"Upload progress: {progress:.2f}%", end="\r")

get_data_products()

Return list of all active data products in a flight.

Returns:

Name Type Description
DataProductCollection DataProductCollection

Collection of data products.
Source code in d2spy/models/flight.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
def get_data_products(self) -> DataProductCollection:
    """Return list of all active data products in a flight.

    Returns:
        DataProductCollection: Collection of data products.
    """
    endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}/data_products"
    response_data = self.client.make_get_request(endpoint)

    data_products = [
        models.DataProduct(
            self.client,
            **schemas.DataProduct.from_dict(data_product).__dict__,
        )
        for data_product in response_data
    ]
    return DataProductCollection(collection=data_products)

get_raw_data()

Return list of all active raw data in a flight.

Returns:

Type Description
List[RawData]

List[models.RawData]: List of raw data.
Source code in d2spy/models/flight.py 
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
def get_raw_data(self) -> List[models.RawData]:
    """Return list of all active raw data in a flight.

    Returns:
        List[models.RawData]: List of raw data.
    """
    endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}/raw_data"
    response_data = self.client.make_get_request(endpoint)

    all_raw_data = [
        models.RawData(
            self.client,
            **schemas.RawData.from_dict(raw_data).__dict__,
        )
        for raw_data in response_data
    ]
    return all_raw_data

move_to_project(destination_project_id)

Moves flight from its current project to different project. You must be an owner of both the source project and destination project to make the transfer.

Parameters:

Name Type Description Default
destination_project_id UUID

ID of project the flight will be moved to.

 required
Source code in d2spy/models/flight.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def move_to_project(self, destination_project_id: UUID) -> None:
    """Moves flight from its current project to different project. You must be an
    owner of both the source project and destination project to make the transfer.

    Args:
        destination_project_id (UUID): ID of project the flight will be moved to.
    """
    endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}"
    endpoint += f"/move_to_project/{destination_project_id}"
    response_data = self.client.make_put_request(endpoint)

    updated_flight = schemas.Flight.from_dict(response_data).__dict__
    for key, value in updated_flight.items():
        if hasattr(self, key):
            setattr(self, key, value)
        else:
            print(f"Warning: Attribute '{key}' not found in Flight class.")
    return None

update(**kwargs)

Update flight attributes.

Source code in d2spy/models/flight.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def update(self, **kwargs) -> None:
    """Update flight attributes."""
    endpoint = f"/api/v1/projects/{self.project_id}/flights/{self.id}"
    response_data = self.client.make_put_request(
        endpoint, json=json.loads(json.dumps(kwargs, default=str))
    )

    updated_flight = schemas.Flight.from_dict(response_data).__dict__
    for key, value in updated_flight.items():
        if hasattr(self, key):
            setattr(self, key, value)
        else:
            print(f"Warning: Attribute '{key}' not found in Flight class.")
    return None