Skip to content

Commit

Permalink
utility function to ease the creation of an epc from scratch (#6)
Browse files Browse the repository at this point in the history
  • Loading branch information
valentin-gauthier-geosiris authored Jun 26, 2024
1 parent 60fedfa commit e81186f
Show file tree
Hide file tree
Showing 6 changed files with 522 additions and 74 deletions.
54 changes: 54 additions & 0 deletions energyml-utils/example/main.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# Copyright (c) 2023-2024 Geosiris.
# SPDX-License-Identifier: Apache-2.0
import json
import re
from dataclasses import fields

from energyml.eml.v2_3.commonv2 import *
Expand Down Expand Up @@ -464,6 +465,57 @@ def test_obj_attribs():
print(get_obj_uri(tr, "coucou"))


def test_copy_values():
data_in = {
"a": {"b": "v_0", "c": "v_1"},
"uuid": "215f8219-cabd-4e24-9e4f-e371cabc9622",
"objectVersion": "Resqml 2.0",
"non_existing": 42,
}
data_out = {
"a": None,
"Uuid": "8291afd6-ae01-49f5-bc96-267e7b27450d",
"object_version": "Resqml 2.0",
}
copy_attributes(
obj_in=data_in,
obj_out=data_out,
only_existing_attributes=True,
ignore_case=True,
)


def class_field():
print(get_class_fields(tr)["citation"])
print(get_class_pkg_version(tr))
print(create_energyml_object("resqml22.TriangulatedSetRepresentation"))
ext_20 = create_energyml_object(
"application/x-eml+xml;version=2.0;type=obj_EpcExternalPartReference"
)
print(ext_20)
print(gen_energyml_object_path(ext_20))
print(create_external_part_reference("2.0", "my_h5"))

print(
parse_content_or_qualified_type(
"application/x-eml+xml;version=2.0;type=obj_EpcExternalPartReference"
)
)
print(
get_domain_version_from_content_or_qualified_type(
"application/x-eml+xml;version=2.0;type=obj_EpcExternalPartReference"
)
)
print(
get_domain_version_from_content_or_qualified_type(
"resqml20.obj_EpcExternalPartReference"
)
)

# print(create_external_part_reference("2.2", "myfile.h5"))
# print(create_external_part_reference("2.0", "myfile.h5"))


if __name__ == "__main__":
tests_0()
tests_content_type()
Expand All @@ -479,3 +531,5 @@ def test_obj_attribs():
test_wellbore_marker_frame_representation()

test_obj_attribs()
test_copy_values()
class_field()
63 changes: 51 additions & 12 deletions energyml-utils/src/energyml/utils/constants.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import datetime
import re
import uuid as uuid_mod
from typing import List
from typing import List, Optional

ENERGYML_NAMESPACES = {
"eml": "http://www.energistics.org/energyml/data/commonv2",
Expand Down Expand Up @@ -197,24 +197,59 @@ def flatten_concatenation(matrix) -> List:
return flat_list


def parse_content_type(ct: str):
def parse_content_type(ct: str) -> Optional[re.Match[str]]:
return re.search(RGX_CONTENT_TYPE, ct)


def parse_qualified_type(ct: str):
def parse_qualified_type(ct: str) -> Optional[re.Match[str]]:
return re.search(RGX_QUALIFIED_TYPE, ct)


def now(
time_zone=datetime.timezone(datetime.timedelta(hours=1), "UTC")
) -> float:
def parse_content_or_qualified_type(cqt: str) -> Optional[re.Match[str]]:
"""
Give a re.Match object (or None if failed).
You can access to groups like : "domainVersion", "versionNum", "domain", "type"
:param cqt:
:return:
"""
parsed = None
try:
parsed = parse_content_type(cqt)
except:
try:
parsed = parse_qualified_type(cqt)
except:
pass

return parsed


def get_domain_version_from_content_or_qualified_type(cqt: str) -> str:
"""
return a version number like "2.2" or "2.0"
:param cqt:
:return:
"""
try:
parsed = parse_content_type(cqt)
return parsed.group("domainVersion")
except:
try:
parsed = parse_qualified_type(cqt)
return ".".join(parsed.group("domainVersion"))
except:
pass
return None


def now(time_zone=datetime.timezone.utc) -> float:
"""Return an epoch value"""
return datetime.datetime.timestamp(datetime.datetime.now(time_zone))


def epoch(
time_zone=datetime.timezone(datetime.timedelta(hours=1), "UTC")
) -> int:
def epoch(time_zone=datetime.timezone.utc) -> int:
return int(now(time_zone))


Expand All @@ -228,10 +263,14 @@ def date_to_epoch(date: str) -> int:

def epoch_to_date(
epoch_value: int,
time_zone=datetime.timezone(datetime.timedelta(hours=1), "UTC"),
) -> str:
date = datetime.datetime.fromtimestamp(epoch_value, time_zone)
return date.strftime("%Y-%m-%dT%H:%M:%S%z")
date = datetime.datetime.fromtimestamp(epoch_value, datetime.timezone.utc)
return date.astimezone(datetime.timezone.utc).strftime(
"%Y-%m-%dT%H:%M:%SZ"
)
# date = datetime.datetime.fromtimestamp(epoch_value, datetime.timezone.utc)
# return date.astimezone(datetime.timezone(datetime.timedelta(hours=0), "UTC")).strftime('%Y-%m-%dT%H:%M:%SZ')
# return date.strftime("%Y-%m-%dT%H:%M:%SZ%z")


def gen_uuid() -> str:
Expand Down
Loading

0 comments on commit e81186f

Please sign in to comment.