Coverage for pds_crawler/models/pds

1# -*- coding: utf-8 -*-

2# pds-crawler - ETL to index PDS data to pdssp

4# This file is part of pds-crawler <https://github.com/pdssp/pds_crawler>

5# SPDX-License-Identifier: LGPL-3.0-or-later

6"""

7Module Name:

8 pds_models

10Description:

11 PDS3 models

13Classes

15.. uml::

17 class ReferenceModel {

18 REFERENCE_KEY_ID: str

19 REFERENCE_DESC: str

20 }

22 class ReferencesModel {

23 REFERENCES: List[ReferenceModel]

24 }

26 ReferencesModel --> ReferenceModel

28.. uml::

30 class DataSetInformationModel {

31 CONFIDENCE_LEVEL_NOTE: str

32 DATA_SET_COLLECTION_MEMBER_FLG: str

33 DATA_SET_DESC: str

34 DATA_SET_NAME: str

35 DATA_SET_RELEASE_DATE: str

36 DETAILED_CATALOG_FLAG: str

37 PRODUCER_FULL_NAME: Union[str, List[str]]

38 START_TIME: str

39 STOP_TIME: str

40 DATA_OBJECT_TYPE: Optional[str]

41 ABSTRACT_DESC: Optional[str]

42 CITATION_DESC: Optional[str]

43 DATA_SET_TERSE_DESC: Optional[str]

44 }

46 class DataSetTargetModel {

47 TARGET_NAME: str

48 }

50 class DataSetHostModel {

51 INSTRUMENT_HOST_ID: str

52 INSTRUMENT_ID: List[str]

53 }

55 class DataSetMissionModel {

56 MISSION_NAME: str

57 }

59 class DataSetReferenceInformationModel {

60 REFERENCE_KEY_ID: str

61 }

63 class DataSetModel {

64 DATA_SET_ID: str

65 DATA_SET_INFORMATION: DataSetInformationModel

66 DATA_SET_TARGET: List[DataSetTargetModel]

67 DATA_SET_HOST: DataSetHostModel

68 DATA_SET_REFERENCE_INFORMATION: List[DataSetReferenceInformationModel]

69 DATA_SET_MISSION: Optional[DataSetMissionModel]

70 }

72 DataSetModel --> DataSetInformationModel

73 DataSetModel --> DataSetTargetModel

74 DataSetModel --> DataSetHostModel

75 DataSetModel --> DataSetReferenceInformationModel

76 DataSetModel --> DataSetMissionModel

78.. uml::

80 class InstrumentReferenceInfoModel {

81 REFERENCE_KEY_ID: str

82 }

84 class InstrumentInformationModel {

85 INSTRUMENT_DESC: str

86 INSTRUMENT_NAME: str

87 INSTRUMENT_TYPE: str

88 }

90 class InstrumentModel {

91 INSTRUMENT_HOST_ID: str

92 INSTRUMENT_ID: str

93 INSTRUMENT_INFORMATION: InstrumentInformationModel

94 INSTRUMENT_REFERENCE_INFO: List[InstrumentReferenceInfoModel]

95 }

97 class InstrumentHostInformationModel {

98 INSTRUMENT_HOST_DESC: str

99 INSTRUMENT_HOST_NAME: str

100 INSTRUMENT_HOST_TYPE: str

101 }

102

103 class InstrumentHostReferenceInfoModel {

104 REFERENCE_KEY_ID: str

105 }

106

107 class InstrumentHostModel {

108 INSTRUMENT_HOST_ID: str

109 INSTRUMENT_HOST_INFORMATION: InstrumentHostInformationModel

110 INSTRUMENT_HOST_REFERENCE_INFO: List[InstrumentHostReferenceInfoModel]

111 }

112

113 InstrumentHostModel --> InstrumentHostInformationModel

114 InstrumentHostModel --> InstrumentHostReferenceInfoModel

115 InstrumentModel --> InstrumentInformationModel

116 InstrumentModel --> InstrumentReferenceInfoModel

117

118.. uml::

119

120 class MissionInformationModel {

121 MISSION_ALIAS_NAME: str

122 MISSION_DESC: str

123 MISSION_OBJECTIVES_SUMMARY: str

124 MISSION_START_DATE: str

125 MISSION_STOP_DATE: str

126 }

127

128 class MissionTargetModel {

129 TARGET_NAME: str

130 }

131

132 class MissionHostModel {

133 INSTRUMENT_HOST_ID: str

134 MISSION_TARGET: List[MissionTargetModel]

135 }

136

137 class MissionReferenceInformationModel {

138 REFERENCE_KEY_ID: str

139 }

140

141 class MissionModel {

142 MISSION_NAME: str

143 MISSION_HOST: MissionHostModel

144 MISSION_INFORMATION: MissionInformationModel

145 MISSION_REFERENCE_INFORMATION: List[MissionReferenceInformationModel]

146 }

147

148 MissionModel --> MissionHostModel

149 MissionModel --> MissionInformationModel

150 MissionModel --> MissionReferenceInformationModel

151 MissionHostModel --> MissionTargetModel

152

153.. uml::

154

155 class DataSetMapProjectionRefInfoModel {

156 REFERENCE_KEY_ID: str

157 }

158

159 class DataSetMapProjectionInfoModel {

160 MAP_PROJECTION_DESC: str

161 MAP_PROJECTION_TYPE: str

162 ROTATIONAL_ELEMENT_DESC: str

163 DS_MAP_PROJECTION_REF_INFO: List[DataSetMapProjectionRefInfoModel]

164 }

165

166 class DataSetMapProjectionModel {

167 DATA_SET_ID: str

168 DATA_SET_MAP_PROJECTION_INFO: DataSetMapProjectionInfoModel

169 }

170

171 DataSetMapProjectionModel --> DataSetMapProjectionInfoModel

172 DataSetMapProjectionInfoModel --> DataSetMapProjectionRefInfoModel

173

174.. uml::

175

176 class PersonnelInformationModel {

177 ADDRESS_TEXT: str

178 ALTERNATE_TELEPHONE_NUMBER: str

179 FAX_NUMBER: str

180 FULL_NAME: str

181 INSTITUTION_NAME: str

182 LAST_NAME: str

183 NODE_ID: str

184 PDS_AFFILIATION: str

185 REGISTRATION_DATE: str

186 TELEPHONE_NUMBER: str

187 PDS_ADDRESS_BOOK_FLAG: Optional[str]

188 }

189

190 class PersonnelElectronicMailModel {

191 ELECTRONIC_MAIL_ID: str

192 ELECTRONIC_MAIL_TYPE: str

193 PREFERENCE_ID: Optional[str]

194 }

195

196 class PersonnelModel {

197 PDS_USER_ID: str

198 PERSONNEL_ELECTRONIC_MAIL: PersonnelElectronicMailModel

199 PERSONNEL_INFORMATION: PersonnelInformationModel

200 }

201

202 class PersonnelsModel {

203 PERSONNELS: List[PersonnelModel]

204 }

205

206 PersonnelsModel --> PersonnelElectronicMailModel

207 PersonnelsModel --> PersonnelInformationModel

208 PersonnelsModel --> PersonnelModel

209

210.. uml::

211

212 class CatalogModel {

213 DATA_SET_CATALOG: Optional[Union[str, List[str]]]

214 INSTRUMENT_CATALOG: Optional[str]

215 INSTRUMENT_HOST_CATALOG: Optional[str]

216 MISSION_CATALOG: Optional[str]

217 LOGICAL_VOLUME_PATHNAME: Optional[str]

218 LOGICAL_VOLUMES: Optional[str]

219 DATA_SET_ID: Optional[str]

220 DATA_SET_COLLECTION_CATALOG: Optional[str]

221 PERSONNEL_CATALOG: Optional[str]

222 REFERENCE_CATALOG: Optional[str]

223 TARGET_CATALOG: Optional[str]

224 }

225

226 class DataProducerModel {

227 INSTITUTION_NAME: str

228 FACILITY_NAME: str

229 FULL_NAME: str

230 ADDRESS_TEXT: str

231 DISCIPLINE_NAME: Optional[str]

232 NODE_NAME: Optional[str] = field(default=None, repr=False, compare=False)

233 TELEPHONE_NUMBER: Optional[str]

234 ELECTRONIC_MAIL_TYPE: Optional[str]

235 ELECTRONIC_MAIL_ID: Optional[str]

236 }

237

238 class FileModel {

239 RECORD_TYPE: str

240 DESCRIPTION: Optional[str]

241 ENCODING_TYPE: Optional[str]

242 FILE_NAME: Optional[str]

243 FILE_RECORDS: Optional[str]

244 INTERCHANGE_FORMAT: Optional[str]

245 LABEL_RECORDS: Optional[str]

246 RECORD_BYTES: Optional[str]

247 REQUIRED_STORAGE_BYTES: Optional[str]

248 SEQUENCE_NUMBER: Optional[str]

249 UNCOMPRESSED_FILE_NAME: Optional[str]

250 }

251

252 class DirectoryModel {

253 NAME: str

254 FILE: List[FileModel]

255 RECORD_TYPE: Optional[str]

256 SEQUENCE_NUMBER: Optional[str]

257 }

258

259 class DataSupplierModel {

260 INSTITUTION_NAME: str

261 FACILITY_NAME: str

262 FULL_NAME: str

263 ADDRESS_TEXT: str

264 TELEPHONE_NUMBER: str

265 ELECTRONIC_MAIL_TYPE: str

266 ELECTRONIC_MAIL_ID: str

267 DISCIPLINE_NAME: Optional[str]

268 NODE_NAME: Optional[str]

269 }

270

271 class VolumeModel {

272 DATA_SET_ID: str

273 DESCRIPTION: str

274 MEDIUM_TYPE: str

275 PUBLICATION_DATE: str

276 VOLUME_FORMAT: str

277 VOLUME_ID: str

278 VOLUME_NAME: str

279 VOLUME_SERIES_NAME: str

280 VOLUME_SET_NAME: str

281 VOLUME_SET_ID: str

282 VOLUME_VERSION_ID: str

283 VOLUMES: str

284 CATALOG: CatalogModel

285 DATA_PRODUCER: DataProducerModel

286 DIRECTORY: Optional[DirectoryModel]

287 FILE: Optional[FileModel]

288 DATA_SUPPLIER: Optional[DataSupplierModel]

289 BLOCK_BYTES: Optional[str]

290 DATA_SET_COLLECTION_ID: Optional[str]

291 FILES: Optional[str]

292 HARDWARE_MODEL_ID: Optional[str]

293 LOGICAL_VOLUMES: Optional[str]

294 LOGICAL_VOLUME_PATH_NAME: Optional[str]

295 MEDIUM_FORMAT: Optional[str]

296 NOTE: Optional[str]

297 OPERATING_SYSTEM_ID

298 PRODUCT_TYPE: Optional[str]

299 TRANSFER_COMMAND_TEXT: Optional[str]

300 VOLUME_INSERT_TEXT: Optional[str]

301 }

302

303 VolumeModel --> CatalogModel

304 VolumeModel --> DataProducerModel

305 VolumeModel --> DirectoryModel

306 VolumeModel --> FileModel

307 VolumeModel --> DataSupplierModel

308

309"""

310import inspect

311from dataclasses import dataclass

312from dataclasses import field

313from datetime import datetime

314from typing import Any

315from typing import cast

316from typing import Dict

317from typing import Iterable

318from typing import List

319from typing import Optional

320from typing import Union

321

322import pystac

323

324from ..utils import utc_to_iso

325from .common import AbstractModel

326from .pdssp_models import PdsspModel

327

328

329@dataclass

330class Labo:

331 ID = "PDS"

332

333

334@dataclass(frozen=True, eq=True)

335class ReferenceModel(AbstractModel):

336 REFERENCE_KEY_ID: str

337 REFERENCE_DESC: str = field(repr=False, compare=False)

338

339

340@dataclass(frozen=True, eq=True)

341class ReferencesModel(AbstractModel):

342 REFERENCES: List[ReferenceModel]

343

344 @classmethod

345 def from_dict(cls, env: Dict):

346 data = env.copy()

347 if "REFERENCES" in data:

348 data["REFERENCES"] = [

349 ReferenceModel.from_dict(elem) for elem in data["REFERENCES"]

350 ]

351 return cls(**{k: v for k, v in data.items()})

352

353

354@dataclass(frozen=True, eq=True)

355class DataSetInformationModel(AbstractModel):

356 CONFIDENCE_LEVEL_NOTE: str = field(repr=False, compare=False)

357 DATA_SET_COLLECTION_MEMBER_FLG: str = field(repr=False, compare=False)

358 DATA_SET_DESC: str = field(repr=False, compare=False)

359 DATA_SET_NAME: str

360 DATA_SET_RELEASE_DATE: str

361 DETAILED_CATALOG_FLAG: str = field(repr=False, compare=False)

362 PRODUCER_FULL_NAME: Union[str, List[str]] = field(

363 repr=False, compare=False

364 )

365 START_TIME: str = field(repr=False, compare=False)

366 STOP_TIME: str = field(repr=False, compare=False)

367 DATA_OBJECT_TYPE: Optional[str] = field(

368 default=None, repr=False, compare=False

369 )

370 ABSTRACT_DESC: Optional[str] = field(

371 default=None, repr=False, compare=False

372 )

373 CITATION_DESC: Optional[str] = field(

374 default=None, repr=False, compare=False

375 )

376 DATA_SET_TERSE_DESC: Optional[str] = field(

377 default=None, repr=False, compare=False

378 )

379

380 def _get_title(self) -> str:

381 return self.DATA_SET_NAME

382

383 def _get_description(self) -> Optional[str]:

384 description: Optional[str] = None

385 if self.ABSTRACT_DESC:

386 description = self.ABSTRACT_DESC

387 elif self.DATA_SET_DESC:

388 description = self.DATA_SET_DESC

389 return description

390

391 def _add_providers(self, stac_collection: pystac.Collection):

392 if isinstance(self.PRODUCER_FULL_NAME, list):

393 stac_collection.providers = [

394 pystac.Provider(

395 name=provider, roles=[pystac.ProviderRole.PRODUCER]

396 )

397 for provider in self.PRODUCER_FULL_NAME

398 ]

399 else:

400 stac_collection.providers = [

401 pystac.Provider(

402 name=self.PRODUCER_FULL_NAME,

403 roles=[pystac.ProviderRole.PRODUCER],

404 )

405 ]

406

407 def _add_citations(self, stac_collection: pystac.Collection):

408 if self.CITATION_DESC:

409 stac_collection.summaries.add(

410 "sci:publications", {"sci:citation": self.CITATION_DESC}

411 )

412 stac_collection.stac_extensions.append(

413 "https://stac-extensions.github.io/scientific/v1.0.0/schema.json"

414 )

415

416 def _get_start_date(self) -> Optional[datetime]:

417 start_date = Optional[datetime]

418 try:

419 start_date = datetime.strptime(

420 utc_to_iso(self.START_TIME), "%Y-%m-%dT%H:%M:%S.%f%z"

421 )

422 except: # noqa: E722

423 start_date = None

424 return start_date

425

426 def _get_stop_date(self) -> Optional[datetime]:

427 stop_date = Optional[datetime]

428 try:

429 stop_date = datetime.strptime(

430 utc_to_iso(self.STOP_TIME), "%Y-%m-%dT%H:%M:%S.%f%z"

431 )

432 except: # noqa: E722

433 stop_date = None

434 return stop_date

435

436 def _get_range_time(self) -> Optional[pystac.RangeSummary]:

437 range: Optional[pystac.RangeSummary] = None

438 start_date = self._get_start_date()

439 stop_date = self._get_stop_date()

440 if start_date is not None and stop_date is not None:

441 range = pystac.RangeSummary(minimum=start_date, maximum=stop_date) # type: ignore

442 return range

443

444 def _set_extent(self, stac_collection: pystac.Collection):

445 stac_collection.extent = pystac.Extent(

446 pystac.SpatialExtent(bboxes=[[]]),

447 pystac.TemporalExtent(

448 intervals=[[self._get_start_date(), self._get_stop_date()]]

449 ),

450 )

451

452 def _add_summaries(self, stac_collection: pystac.Collection):

453 summaries: Dict[str, Any] = dict()

454 range_time = self._get_range_time()

455

456 if stac_collection.summaries and range_time is not None:

457 stac_collection.summaries.add("observation_time", range_time)

458 elif stac_collection.summaries is None and range_time is not None:

459 stac_collection.summaries = pystac.Summaries(summaries=summaries)

460

461 def _add_extra_kwds(self, stac_collection: pystac.Collection):

462 extra_kws = {

463 attribute.lower(): self.__dict__[attribute]

464 for attribute in self.__dict__.keys()

465 if attribute

466 in [

467 "CONFIDENCE_LEVEL_NOTE",

468 "DATA_SET_COLLECTION_MEMBER_FLG",

469 "DETAILED_CATALOG_FLAG",

470 "DATA_OBJECT_TYPE",

471 "DATA_SET_RELEASE_DATE" "CITATION_DESC",

472 ]

473 and attribute is not None

474 }

475 if stac_collection.extra_fields:

476 stac_collection.extra_fields.update(extra_kws)

477 else:

478 stac_collection.extra_fields = extra_kws

479

480 @classmethod

481 def from_dict(cls, env):

482 parameters = inspect.signature(cls).parameters

483 return cls(**{k: v for k, v in env.items() if k in parameters})

484

485 def update_stac(self, stac_collection: pystac.Collection):

486 stac_collection.title = self._get_title()

487 if self._get_description():

488 stac_collection.description = cast(str, self._get_description())

489

490 self._set_extent(stac_collection)

491 self._add_providers(stac_collection)

492 self._add_citations(stac_collection)

493 self._add_summaries(stac_collection)

494 self._add_extra_kwds(stac_collection)

495

496

497@dataclass(frozen=True, eq=True)

498class DataSetTargetModel(AbstractModel):

499 TARGET_NAME: str

500

501 @classmethod

502 def from_dict(cls, env):

503 parameters = inspect.signature(cls).parameters

504 return cls(**{k: v for k, v in env.items() if k in parameters})

505

506 def update_stac(self, stac_collection: pystac.Collection):

507 if not stac_collection.extra_fields:

508 stac_collection.extra_fields = dict()

509 stac_collection.extra_fields["ssys:targets"] = self.TARGET_NAME

510 stac_collection.stac_extensions.append(

511 "https://github.com/thareUSGS/ssys/blob/main/json-schema/schema.json"

512 )

513

514

515@dataclass(frozen=True, eq=True)

516class DataSetHostModel(AbstractModel):

517 INSTRUMENT_HOST_ID: str

518 INSTRUMENT_ID: Union[str, List[str]]

519

520 def get_instrument_id(self) -> Union[str, List[str]]:

521 result: Union[str, List[str]]

522 if isinstance(self.INSTRUMENT_ID, str):

523 result = PdsspModel.create_instru_id(Labo.ID, self.INSTRUMENT_ID)

524 else:

525 result = [

526 PdsspModel.create_instru_id(Labo.ID, instrument)

527 for instrument in self.INSTRUMENT_ID

528 ]

529 return result

530

531 def get_plateform_id(self):

532 return PdsspModel.create_platform_id(Labo.ID, self.INSTRUMENT_HOST_ID)

533

534 def update_stac(self, stac_collection: pystac.Collection):

535 if not stac_collection.extra_fields:

536 stac_collection.extra_fields = dict()

537 stac_collection.extra_fields["plateform_id"] = self.INSTRUMENT_HOST_ID

538 stac_collection.extra_fields["instrument_id"] = self.INSTRUMENT_ID

539

540

541@dataclass(frozen=True, eq=True)

542class DataSetMissionModel(AbstractModel):

543 MISSION_NAME: str

544

545 def update_stac(self, stac_collection: pystac.Collection):

546 if not stac_collection.extra_fields:

547 stac_collection.extra_fields = dict()

548 stac_collection.extra_fields["mission"] = self.MISSION_NAME

549

550

551@dataclass(frozen=True, eq=True)

552class DataSetReferenceInformationModel(AbstractModel):

553 REFERENCE_KEY_ID: str

554

555

556@dataclass(frozen=True, eq=True)

557class DataSetModel(AbstractModel):

558 DATA_SET_ID: str

559 DATA_SET_INFORMATION: DataSetInformationModel = field(

560 repr=False, compare=False

561 )

562 DATA_SET_TARGET: List[DataSetTargetModel] = field(

563 repr=False, compare=False

564 )

565 DATA_SET_HOST: DataSetHostModel = field(repr=False, compare=False)

566 DATA_SET_REFERENCE_INFORMATION: List[

567 DataSetReferenceInformationModel

568 ] = field(repr=False, compare=False)

569 DATA_SET_MISSION: Optional[DataSetMissionModel] = field(

570 default=None, repr=False, compare=False

571 )

572

573 def get_collection_id(self) -> str:

574 return PdsspModel.create_collection_id(Labo.ID, self.DATA_SET_ID)

575

576 def _add_citations(

577 self,

578 stac_collection: pystac.Collection,

579 citations: Optional[ReferencesModel],

580 ):

581 if citations:

582 references = {

583 citation.REFERENCE_KEY_ID: citation.REFERENCE_DESC

584 for citation in citations.REFERENCES

585 if citation.REFERENCE_KEY_ID is not None

586 }

587 publi_list = [

588 references.get(citation.REFERENCE_KEY_ID)

589 for citation in self.DATA_SET_REFERENCE_INFORMATION

590 if references.get(citation.REFERENCE_KEY_ID) is not None

591 ]

592 if stac_collection.extra_fields:

593 stac_collection.extra_fields.update({})

594 stac_collection.extra_fields = {"sci:publications": publi_list}

595

596 def _add_providers(

597 self,

598 stac_collection: pystac.Collection,

599 data_supplier: Optional["DataSupplierModel"],

600 data_producer: Optional["DataProducerModel"],

601 ):

602 providers: List[pystac.Provider] = list()

603 if data_supplier:

604 providers.append(data_supplier.create_stac_data_provider())

605 if data_producer:

606 providers.append(data_producer.create_stac_data_provider())

607 stac_collection.providers = providers

608

609 @classmethod

610 def from_dict(cls, env: Dict):

611 data = env.copy()

612 if "DATA_SET_INFORMATION" in data:

613 data["DATA_SET_INFORMATION"] = DataSetInformationModel.from_dict(

614 data["DATA_SET_INFORMATION"]

615 )

616 if "DATA_SET_TARGET" in data:

617 data["DATA_SET_TARGET"] = [

618 DataSetTargetModel.from_dict(elem)

619 for elem in data["DATA_SET_TARGET"]

620 ]

621 if "DATA_SET_HOST" in data:

622 data["DATA_SET_HOST"] = DataSetHostModel.from_dict(

623 data["DATA_SET_HOST"]

624 )

625

626 if "DATA_SET_MISSION" in data:

627 data["DATA_SET_MISSION"] = DataSetMissionModel.from_dict(

628 data["DATA_SET_MISSION"]

629 )

630 if "DATA_SET_REFERENCE_INFORMATION" in data:

631 data["DATA_SET_REFERENCE_INFORMATION"] = [

632 DataSetReferenceInformationModel.from_dict(elem)

633 for elem in data["DATA_SET_REFERENCE_INFORMATION"]

634 ]

635 if (

636 isinstance(data["DATA_SET_ID"], list)

637 and len(data["DATA_SET_ID"]) > 1

638 ):

639 raise ValueError(

640 f"An array of DATA_SET_ID is fobidden : {data['DATA_SET_ID']}"

641 )

642 elif (

643 isinstance(data["DATA_SET_ID"], list)

644 and len(data["DATA_SET_ID"]) == 1

645 ):

646 data["DATA_SET_ID"] = data["DATA_SET_ID"][0]

647

648 return cls(**{k: v for k, v in data.items()})

649

650 def create_stac_collection(

651 self,

652 body_id: str,

653 citations: Optional[ReferencesModel],

654 data_supplier: Optional["DataSupplierModel"],

655 data_producer: Optional["DataProducerModel"],

656 ) -> pystac.Collection:

657 extension: Dict = PdsspModel.create_ssys_extension(body_id)

658 stac_collection = pystac.Collection(

659 id=self.get_collection_id(),

660 description="",

661 extent=pystac.Extent(

662 pystac.SpatialExtent(bboxes=[[]]),

663 pystac.TemporalExtent(intervals=[[None, None]]),

664 ),

665 stac_extensions=list(),

666 extra_fields=dict(),

667 )

668 stac_collection.stac_extensions.extend(extension["stac_extensions"])

669 stac_collection.extra_fields.update(extension["extra_fields"])

670 self.DATA_SET_INFORMATION.update_stac(stac_collection)

671 self.DATA_SET_TARGET[0].update_stac(stac_collection)

672 self.DATA_SET_HOST.update_stac(stac_collection)

673

674 if self.DATA_SET_MISSION:

675 self.DATA_SET_MISSION.update_stac(stac_collection)

676

677 self._add_citations(stac_collection, citations)

678 self._add_providers(stac_collection, data_supplier, data_producer)

679 stac_collection.license = "CC0-1.0"

680

681 return stac_collection

682

683

684@dataclass(frozen=True, eq=True)

685class InstrumentReferenceInfoModel(AbstractModel):

686 REFERENCE_KEY_ID: str

687

688

689@dataclass(frozen=True, eq=True)

690class InstrumentInformationModel(AbstractModel):

691 INSTRUMENT_DESC: str = field(repr=False, compare=False)

692 INSTRUMENT_NAME: str

693 INSTRUMENT_TYPE: str

694

695 def update_stac(self, stac_catalog: pystac.Catalog):

696 stac_catalog.description = self.INSTRUMENT_DESC

697 stac_catalog.title = self.INSTRUMENT_NAME

698 if not stac_catalog.extra_fields:

699 stac_catalog.extra_fields = dict()

700 stac_catalog.extra_fields["instrument_type"] = self.INSTRUMENT_TYPE

701

702

703@dataclass(frozen=True, eq=True)

704class InstrumentModel(AbstractModel):

705 INSTRUMENT_HOST_ID: str

706 INSTRUMENT_ID: str

707 INSTRUMENT_INFORMATION: InstrumentInformationModel = field(

708 repr=False, compare=False

709 )

710 INSTRUMENT_REFERENCE_INFO: List[InstrumentReferenceInfoModel] = field(

711 repr=False, compare=False

712 )

713

714 def get_plateform_id(self):

715 return PdsspModel.create_platform_id(Labo.ID, self.INSTRUMENT_HOST_ID)

716

717 def get_instrument_id(self):

718 return PdsspModel.create_instru_id(Labo.ID, self.INSTRUMENT_ID)

719

720 def _add_citations(

721 self,

722 stac_catalog: pystac.Catalog,

723 citations: Optional[ReferencesModel],

724 ):

725 if citations:

726 references = {

727 citation.REFERENCE_KEY_ID: citation.REFERENCE_DESC

728 for citation in citations.REFERENCES

729 if citation.REFERENCE_KEY_ID is not None

730 }

731 publi_list = [

732 references.get(citation.REFERENCE_KEY_ID)

733 for citation in self.INSTRUMENT_REFERENCE_INFO

734 if references.get(citation.REFERENCE_KEY_ID) is not None

735 ]

736 if not stac_catalog.extra_fields:

737 stac_catalog.extra_fields = dict()

738 stac_catalog.extra_fields["publications"] = publi_list

739

740 @classmethod

741 def from_dict(cls, env: Dict):

742 data = env.copy()

743 if "INSTRUMENT_INFORMATION" in data:

744 data[

745 "INSTRUMENT_INFORMATION"

746 ] = InstrumentInformationModel.from_dict(

747 data["INSTRUMENT_INFORMATION"]

748 )

749 if "INSTRUMENT_REFERENCE_INFO" in data:

750 data["INSTRUMENT_REFERENCE_INFO"] = [

751 InstrumentReferenceInfoModel.from_dict(elem)

752 for elem in data["INSTRUMENT_REFERENCE_INFO"]

753 ]

754 return cls(**{k: v for k, v in data.items()})

755

756 def create_stac_catalog(

757 self, body_id: str, citations: Optional[ReferencesModel] = None

758 ) -> pystac.Catalog:

759 extension: Dict = PdsspModel.create_ssys_extension(body_id)

760 stac_catalog = pystac.Catalog(

761 title=self.INSTRUMENT_ID,

762 id=self.get_instrument_id(),

763 description="",

764 stac_extensions=list(),

765 extra_fields=dict(),

766 )

767 stac_catalog.stac_extensions.extend(extension["stac_extensions"])

768 stac_catalog.extra_fields.update(extension["extra_fields"])

769 self._add_citations(stac_catalog, citations)

770 self.INSTRUMENT_INFORMATION.update_stac(stac_catalog)

771

772 return stac_catalog

773

774

775@dataclass(frozen=True, eq=True)

776class InstrumentHostInformationModel(AbstractModel):

777 INSTRUMENT_HOST_DESC: str = field(repr=False, compare=False)

778 INSTRUMENT_HOST_NAME: str

779 INSTRUMENT_HOST_TYPE: str

780

781 def update_stac(self, stac_catalog: pystac.Catalog):

782 stac_catalog.description = self.INSTRUMENT_HOST_DESC

783 stac_catalog.title = self.INSTRUMENT_HOST_NAME

784 if not stac_catalog.extra_fields:

785 stac_catalog.extra_fields = dict()

786 stac_catalog.extra_fields["plateform"] = self.INSTRUMENT_HOST_TYPE

787

788

789@dataclass(frozen=True, eq=True)

790class InstrumentHostReferenceInfoModel(AbstractModel):

791 REFERENCE_KEY_ID: str

792

793

794@dataclass(frozen=True, eq=True)

795class InstrumentHostModel(AbstractModel):

796 INSTRUMENT_HOST_ID: str

797 INSTRUMENT_HOST_INFORMATION: InstrumentHostInformationModel = field(

798 repr=False, compare=False

799 )

800 INSTRUMENT_HOST_REFERENCE_INFO: List[

801 InstrumentHostReferenceInfoModel

802 ] = field(repr=False, compare=False)

803

804 def get_plateform_id(self) -> str:

805 return PdsspModel.create_platform_id(Labo.ID, self.INSTRUMENT_HOST_ID)

806

807 def _add_citations(

808 self,

809 stac_catalog: pystac.Catalog,

810 citations: Optional[ReferencesModel],

811 ):

812 if citations:

813 references = {

814 citation.REFERENCE_KEY_ID: citation.REFERENCE_DESC

815 for citation in citations.REFERENCES

816 if citation.REFERENCE_KEY_ID is not None

817 }

818 publi_list = [

819 references.get(citation.REFERENCE_KEY_ID)

820 for citation in self.INSTRUMENT_HOST_REFERENCE_INFO

821 if references.get(citation.REFERENCE_KEY_ID) is not None

822 ]

823 if not stac_catalog.extra_fields:

824 stac_catalog.extra_fields = dict()

825 stac_catalog.extra_fields["publications"] = publi_list

826

827 @classmethod

828 def from_dict(cls, env: Dict):

829 data = env.copy()

830 if "INSTRUMENT_HOST_INFORMATION" in data:

831 data[

832 "INSTRUMENT_HOST_INFORMATION"

833 ] = InstrumentHostInformationModel.from_dict(

834 data["INSTRUMENT_HOST_INFORMATION"]

835 )

836 if "INSTRUMENT_HOST_REFERENCE_INFO" in data:

837 data["INSTRUMENT_HOST_REFERENCE_INFO"] = [

838 InstrumentHostReferenceInfoModel.from_dict(elem)

839 for elem in data["INSTRUMENT_HOST_REFERENCE_INFO"]

840 ]

841 return cls(**{k: v for k, v in data.items()})

842

843 def create_stac_catalog(

844 self, body_id: str, citations: Optional[ReferencesModel] = None

845 ) -> pystac.Catalog:

846 extension: Dict = PdsspModel.create_ssys_extension(body_id)

847 stac_catalog = pystac.Catalog(

848 title=self.INSTRUMENT_HOST_ID,

849 id=self.get_plateform_id(),

850 description="",

851 stac_extensions=list(),

852 extra_fields=dict(),

853 )

854 stac_catalog.stac_extensions.extend(extension["stac_extensions"])

855 stac_catalog.extra_fields.update(extension["extra_fields"])

856 self._add_citations(stac_catalog, citations)

857 self.INSTRUMENT_HOST_INFORMATION.update_stac(stac_catalog)

858

859 return stac_catalog

860

861

862@dataclass(frozen=True, eq=True)

863class MissionInformationModel(AbstractModel):

864 MISSION_ALIAS_NAME: str

865 MISSION_DESC: str = field(repr=False, compare=False)

866 MISSION_OBJECTIVES_SUMMARY: str = field(repr=False, compare=False)

867 MISSION_START_DATE: str

868 MISSION_STOP_DATE: str

869

870 def get_mission_id(self):

871 return PdsspModel.create_mission_id(Labo.ID, self.MISSION_ALIAS_NAME)

872

873 @classmethod

874 def from_dict(cls, env):

875 data = env.copy()

876 # Fix problem

877 if data["MISSION_ALIAS_NAME"] == "N/A":

878 data["MISSION_ALIAS_NAME"] = "MGS"

879 return cls(**{k: v for k, v in data.items()})

880

881 def update_stac(self, stac_catalog: pystac.Catalog):

882 stac_catalog.description = self.MISSION_DESC

883 stac_catalog.id = self.get_mission_id()

884 if not stac_catalog.extra_fields:

885 stac_catalog.extra_fields = dict()

886 stac_catalog.extra_fields[

887 "mission_objectives_summary"

888 ] = self.MISSION_OBJECTIVES_SUMMARY

889 stac_catalog.extra_fields[

890 "mission_start_date"

891 ] = self.MISSION_START_DATE

892 stac_catalog.extra_fields["mission_stop_date"] = self.MISSION_STOP_DATE

893

894

895@dataclass(frozen=True, eq=True)

896class MissionTargetModel(AbstractModel):

897 TARGET_NAME: str

898

899 @classmethod

900 def from_dict(cls, env: Dict):

901 data = env.copy()

902 if "MISSION_TARGET" in data:

903 data["MISSION_TARGET"] = [

904 MissionTargetModel.from_dict(elem)

905 for elem in data["MISSION_TARGET"]

906 ]

907 return cls(**{k: v for k, v in data.items()})

908

909

910@dataclass(frozen=True, eq=True)

911class MissionHostModel(AbstractModel):

912 INSTRUMENT_HOST_ID: str

913 MISSION_TARGET: List[MissionTargetModel]

914

915 @classmethod

916 def from_dict(cls, env: Dict):

917 data = env.copy()

918 if "MISSION_TARGET" in data:

919 data["MISSION_TARGET"] = [

920 MissionTargetModel.from_dict(elem)

921 for elem in data["MISSION_TARGET"]

922 ]

923 return cls(**{k: v for k, v in data.items()})

924

925 def update_stac(self, stac_catalog: pystac.Catalog):

926 if not stac_catalog.extra_fields:

927 stac_catalog.extra_fields = dict()

928 stac_catalog.extra_fields["plateform_id"] = self.INSTRUMENT_HOST_ID

929 stac_catalog.extra_fields["mission_targets"] = [

930 mission_target.TARGET_NAME

931 for mission_target in self.MISSION_TARGET

932 ]

933

934

935@dataclass(frozen=True, eq=True)

936class MissionReferenceInformationModel(AbstractModel):

937 REFERENCE_KEY_ID: str

938

939

940@dataclass(frozen=True, eq=True)

941class DataSetMapProjectionRefInfoModel(AbstractModel):

942 REFERENCE_KEY_ID: str

943

944

945@dataclass(frozen=True, eq=True)

946class DataSetMapProjectionInfoModel(AbstractModel):

947 MAP_PROJECTION_DESC: str = field(repr=False, compare=False)

948 MAP_PROJECTION_TYPE: str

949 ROTATIONAL_ELEMENT_DESC: str = field(repr=False, compare=False)

950 DS_MAP_PROJECTION_REF_INFO: List[DataSetMapProjectionRefInfoModel] = field(

951 repr=False, compare=False

952 )

953

954

955@dataclass(frozen=True, eq=True)

956class DataSetMapProjectionModel(AbstractModel):

957 DATA_SET_ID: str

958 DATA_SET_MAP_PROJECTION_INFO: DataSetMapProjectionInfoModel = field(

959 repr=False, compare=False

960 )

961

962

963@dataclass(frozen=True, eq=True)

964class MissionModel(AbstractModel):

965 MISSION_NAME: str

966 MISSION_HOST: MissionHostModel

967 MISSION_INFORMATION: MissionInformationModel = field(

968 repr=False, compare=False

969 )

970 MISSION_REFERENCE_INFORMATION: Iterable[

971 MissionReferenceInformationModel

972 ] = field(repr=False, compare=False)

973

974 def _add_citations(

975 self,

976 stac_catalog: pystac.Catalog,

977 citations: Optional[ReferencesModel],

978 ):

979 if citations:

980 references = {

981 citation.REFERENCE_KEY_ID: citation.REFERENCE_DESC

982 for citation in citations.REFERENCES

983 if citation.REFERENCE_KEY_ID is not None

984 }

985 publi_list = [

986 references.get(citation.REFERENCE_KEY_ID)

987 for citation in self.MISSION_REFERENCE_INFORMATION

988 if references.get(citation.REFERENCE_KEY_ID) is not None

989 ]

990 if not stac_catalog.extra_fields:

991 stac_catalog.extra_fields = dict()

992

993 if len(publi_list) > 0:

994 stac_catalog.extra_fields["publications"]: publi_list # type: ignore

995

996 @classmethod

997 def from_dict(cls, env: Dict):

998 data = env.copy()

999 if "MISSION_HOST" in data:

1000 data["MISSION_HOST"] = MissionHostModel.from_dict(

1001 data["MISSION_HOST"]

1002 )

1003 if "MISSION_INFORMATION" in data:

1004 data["MISSION_INFORMATION"] = MissionInformationModel.from_dict(

1005 data["MISSION_INFORMATION"]

1006 )

1007 if "MISSION_REFERENCE_INFORMATION" in data:

1008 data["MISSION_REFERENCE_INFORMATION"] = [

1009 MissionReferenceInformationModel.from_dict(elem)

1010 for elem in data["MISSION_REFERENCE_INFORMATION"]

1011 ]

1012 parameters = inspect.signature(cls).parameters

1013 return cls(**{k: v for k, v in data.items() if k in parameters})

1014

1015 def create_stac_catalog(

1016 self, body: str, citations: Optional[ReferencesModel] = None

1017 ) -> pystac.Catalog:

1018 extension = PdsspModel.create_ssys_extension(body)

1019 stac_catalog = pystac.Catalog(

1020 title=self.MISSION_NAME,

1021 id="",

1022 description="",

1023 stac_extensions=list(),

1024 extra_fields=dict(),

1025 )

1026 stac_catalog.stac_extensions.extend(extension["stac_extensions"])

1027 stac_catalog.extra_fields.update(extension["extra_fields"])

1028 self.MISSION_HOST.update_stac(stac_catalog)

1029 self.MISSION_INFORMATION.update_stac(stac_catalog)

1030

1031 return stac_catalog

1032

1033

1034@dataclass(frozen=True, eq=True)

1035class PersonnelInformationModel(AbstractModel):

1036 ADDRESS_TEXT: str = field(repr=False, compare=False)

1037 ALTERNATE_TELEPHONE_NUMBER: str = field(repr=False, compare=False)

1038 FAX_NUMBER: str = field(repr=False, compare=False)

1039 FULL_NAME: str

1040 INSTITUTION_NAME: str

1041 LAST_NAME: str

1042 NODE_ID: str = field(repr=False)

1043 PDS_AFFILIATION: str = field(repr=False, compare=False)

1044 REGISTRATION_DATE: str = field(repr=False, compare=False)

1045 TELEPHONE_NUMBER: str = field(repr=False, compare=False)

1046 PDS_ADDRESS_BOOK_FLAG: Optional[str] = field(

1047 default=None, repr=False, compare=False

1048 )

1049

1050

1051@dataclass(frozen=True, eq=True)

1052class PersonnelElectronicMailModel(AbstractModel):

1053 ELECTRONIC_MAIL_ID: str

1054 ELECTRONIC_MAIL_TYPE: str = field(repr=False, compare=False)

1055 PREFERENCE_ID: Optional[str] = field(

1056 default=None, repr=False, compare=False

1057 )

1058

1059

1060@dataclass(frozen=True, eq=True)

1061class PersonnelModel(AbstractModel):

1062 PDS_USER_ID: str

1063 PERSONNEL_ELECTRONIC_MAIL: PersonnelElectronicMailModel = field(

1064 repr=False, compare=False

1065 )

1066 PERSONNEL_INFORMATION: PersonnelInformationModel = field(

1067 repr=False, compare=False

1068 )

1069

1070 @classmethod

1071 def from_dict(cls, env: Dict):

1072 data = env.copy()

1073 if "PERSONNEL_ELECTRONIC_MAIL" in data:

1074 data[

1075 "PERSONNEL_ELECTRONIC_MAIL"

1076 ] = PersonnelElectronicMailModel.from_dict(

1077 data["PERSONNEL_ELECTRONIC_MAIL"]

1078 )

1079 if "PERSONNEL_INFORMATION" in data:

1080 data[

1081 "PERSONNEL_INFORMATION"

1082 ] = PersonnelInformationModel.from_dict(

1083 data["PERSONNEL_INFORMATION"]

1084 )

1085 return cls(**{k: v for k, v in data.items()})

1086

1087

1088@dataclass(frozen=True, eq=True)

1089class PersonnelsModel(AbstractModel):

1090 PERSONNELS: List[PersonnelModel]

1091

1092 @classmethod

1093 def from_dict(cls, env: Dict):

1094 data = env.copy()

1095 if "PERSONNELS" in data:

1096 data["PERSONNELS"] = [

1097 PersonnelModel.from_dict(elem) for elem in data["PERSONNELS"]

1098 ]

1099 return cls(**{k: v for k, v in data.items()})

1100

1101

1102@dataclass(frozen=True, eq=True)

1103class CatalogModel(AbstractModel):

1104 DATA_SET_CATALOG: Optional[Union[str, List[str]]] = field(

1105 default=None, repr=False, compare=False

1106 )

1107 INSTRUMENT_CATALOG: Optional[str] = field(

1108 default=None, repr=False, compare=False

1109 )

1110 INSTRUMENT_HOST_CATALOG: Optional[str] = field(

1111 default=None, repr=False, compare=False

1112 )

1113 MISSION_CATALOG: Optional[str] = field(

1114 default=None, repr=False, compare=False

1115 )

1116 LOGICAL_VOLUME_PATHNAME: Optional[str] = field(

1117 default=None, repr=False, compare=False

1118 )

1119 LOGICAL_VOLUMES: Optional[str] = field(

1120 default=None, repr=False, compare=False

1121 )

1122 DATA_SET_ID: Optional[str] = field(default=None, repr=False, compare=False)

1123 DATA_SET_COLLECTION_CATALOG: Optional[str] = field(

1124 default=None, repr=False, compare=False

1125 )

1126 PERSONNEL_CATALOG: Optional[str] = field(

1127 default=None, repr=False, compare=False

1128 )

1129 REFERENCE_CATALOG: Optional[str] = field(

1130 default=None, repr=False, compare=False

1131 )

1132 TARGET_CATALOG: Optional[str] = field(

1133 default=None, repr=False, compare=False

1134 )

1135

1136 @classmethod

1137 def from_dict(cls, env: Dict):

1138 parameters = inspect.signature(cls).parameters

1139 return cls(**{k: v for k, v in env.items() if k in parameters})

1140

1141

1142@dataclass(frozen=True, eq=True)

1143class DataProducerModel(AbstractModel):

1144 INSTITUTION_NAME: str

1145 FACILITY_NAME: str

1146 FULL_NAME: str

1147 ADDRESS_TEXT: str = field(repr=False, compare=False)

1148 DISCIPLINE_NAME: Optional[str] = field(

1149 default=None, repr=False, compare=False

1150 )

1151 NODE_NAME: Optional[str] = field(default=None, repr=False, compare=False)

1152 TELEPHONE_NUMBER: Optional[str] = field(

1153 default=None, repr=False, compare=False

1154 )

1155 ELECTRONIC_MAIL_TYPE: Optional[str] = field(

1156 default=None, repr=False, compare=False

1157 )

1158 ELECTRONIC_MAIL_ID: Optional[str] = field(

1159 default=None, repr=False, compare=False

1160 )

1161

1162 def create_stac_data_provider(self) -> pystac.Provider:

1163 producer = pystac.Provider(

1164 name=self.FULL_NAME, roles=[pystac.ProviderRole.HOST]

1165 )

1166 producer.extra_fields = {

1167 key.lower(): self.__dict__[key]

1168 for key in self.__dict__.keys()

1169 if key is not None and key not in ["FULL_NAME"]

1170 }

1171 return producer

1172

1173

1174@dataclass(frozen=True, eq=True)

1175class FileModel(AbstractModel):

1176 RECORD_TYPE: str

1177 DESCRIPTION: Optional[str] = field(default=None, repr=False, compare=False)

1178 ENCODING_TYPE: Optional[str] = field(

1179 default=None, repr=False, compare=False

1180 )

1181 FILE_NAME: Optional[str] = field(default=None, repr=False)

1182 FILE_RECORDS: Optional[str] = field(default=None, repr=False)

1183 INTERCHANGE_FORMAT: Optional[str] = field(

1184 default=None, repr=False, compare=False

1185 )

1186 LABEL_RECORDS: Optional[str] = field(

1187 default=None, repr=False, compare=False

1188 )

1189 RECORD_BYTES: Optional[str] = field(default=None, repr=False)

1190 REQUIRED_STORAGE_BYTES: Optional[str] = field(

1191 default=None, repr=False, compare=False

1192 )

1193 SEQUENCE_NUMBER: Optional[str] = field(default=None, repr=False)

1194 UNCOMPRESSED_FILE_NAME: Optional[str] = field(

1195 default=None, repr=False, compare=False

1196 )

1197

1198

1199@dataclass(frozen=True, eq=True)

1200class DirectoryModel(AbstractModel):

1201 NAME: str

1202 FILE: List[FileModel] = field(repr=False)

1203 RECORD_TYPE: Optional[str] = field(default=None, repr=False, compare=False)

1204 SEQUENCE_NUMBER: Optional[str] = field(

1205 default=None, repr=False, compare=False

1206 )

1207

1208 @classmethod

1209 def from_dict(cls, env: Dict):

1210 data = env.copy()

1211 if "FILE" in data:

1212 data["FILE"] = [FileModel.from_dict(elem) for elem in data["FILE"]]

1213 return cls(**{k: v for k, v in data.items()})

1214

1215

1216@dataclass(frozen=True, eq=True)

1217class DataSupplierModel(AbstractModel):

1218 INSTITUTION_NAME: str

1219 FACILITY_NAME: str

1220 FULL_NAME: str

1221 ADDRESS_TEXT: str = field(repr=False, compare=False)

1222 TELEPHONE_NUMBER: str = field(repr=False, compare=False)

1223 ELECTRONIC_MAIL_TYPE: str = field(repr=False, compare=False)

1224 ELECTRONIC_MAIL_ID: str

1225 DISCIPLINE_NAME: Optional[str] = field(

1226 default=None, repr=False, compare=False

1227 )

1228 NODE_NAME: Optional[str] = field(default=None, repr=False, compare=False)

1229

1230 def create_stac_data_provider(self) -> pystac.Provider:

1231 supplier = pystac.Provider(

1232 name=self.FULL_NAME, roles=[pystac.ProviderRole.HOST]

1233 )

1234 supplier.extra_fields = {

1235 key.lower(): self.__dict__[key]

1236 for key in self.__dict__.keys()

1237 if key is not None and key not in ["FULL_NAME"]

1238 }

1239 return supplier

1240

1241

1242@dataclass(frozen=True, eq=True)

1243class VolumeModel(AbstractModel):

1244 DATA_SET_ID: str

1245 DESCRIPTION: str = field(repr=False, compare=False)

1246 MEDIUM_TYPE: str = field(repr=False, compare=False)

1247 PUBLICATION_DATE: str = field(repr=False, compare=False)

1248 VOLUME_FORMAT: str

1249 VOLUME_ID: str

1250 VOLUME_NAME: str

1251 VOLUME_SERIES_NAME: str

1252 VOLUME_SET_NAME: str

1253 VOLUME_SET_ID: str

1254 VOLUME_VERSION_ID: str

1255 VOLUMES: str = field(repr=False, compare=False)

1256 CATALOG: CatalogModel

1257 DATA_PRODUCER: DataProducerModel

1258 DIRECTORY: Optional[DirectoryModel] = field(

1259 default=None, repr=False, compare=False

1260 )

1261 FILE: Optional[FileModel] = field(default=None, repr=False, compare=False)

1262 DATA_SUPPLIER: Optional[DataSupplierModel] = field(

1263 default=None, repr=False, compare=False

1264 )

1265 BLOCK_BYTES: Optional[str] = field(default=None, repr=False, compare=False)

1266 DATA_SET_COLLECTION_ID: Optional[str] = field(

1267 default=None, repr=False, compare=False

1268 )

1269 FILES: Optional[str] = field(default=None, repr=False, compare=False)

1270 HARDWARE_MODEL_ID: Optional[str] = field(

1271 default=None, repr=False, compare=False

1272 )

1273 LOGICAL_VOLUMES: Optional[str] = field(

1274 default=None, repr=False, compare=False

1275 )

1276 LOGICAL_VOLUME_PATH_NAME: Optional[str] = field(

1277 default=None, repr=False, compare=False

1278 )

1279 MEDIUM_FORMAT: Optional[str] = field(

1280 default=None, repr=False, compare=False

1281 )

1282 NOTE: Optional[str] = field(default=None, repr=False, compare=False)

1283 OPERATING_SYSTEM_ID: Optional[str] = field(

1284 default=None, repr=False, compare=False

1285 )

1286 PRODUCT_TYPE: Optional[str] = field(

1287 default=None, repr=False, compare=False

1288 )

1289 TRANSFER_COMMAND_TEXT: Optional[str] = field(

1290 default=None, repr=False, compare=False

1291 )

1292 VOLUME_INSERT_TEXT: Optional[str] = field(

1293 default=None, repr=False, compare=False

1294 )

1295

1296 @classmethod

1297 def from_dict(cls, env: Dict[str, Any]):

1298 data = env.copy()

1299

1300 if "CATALOG" in data:

1301 data["CATALOG"] = CatalogModel.from_dict(data["CATALOG"])

1302 if "DATA_PRODUCER" in data:

1303 data["DATA_PRODUCER"] = DataProducerModel.from_dict(

1304 data["DATA_PRODUCER"]

1305 )

1306 if "DATA_SUPPLIER" in data:

1307 data["DATA_SUPPLIER"] = DataSupplierModel.from_dict(

1308 data["DATA_SUPPLIER"]

1309 )

1310 if "FILE" in data:

1311 data["FILE"] = FileModel.from_dict(data["FILE"])

1312 if "DIRECTORY" in data:

1313 data["DIRECTORY"] = DirectoryModel.from_dict(data["DIRECTORY"])

1314

1315 parameters = inspect.signature(cls).parameters

1316 return cls(**{k: v for k, v in data.items() if k in parameters})

Coverage for pds_crawler/models/pds_models.py: 93%

551 statements