Commit 68ec0d0f authored by Jiahang Chen's avatar Jiahang Chen
Browse files

add stanford2010 sync

parent bf2d9cac
Pipeline #494593 passed with stages
in 33 seconds
......@@ -341,7 +341,7 @@ def add_function_impl_obj(thing, impl_obj, feature_name, **kwargs):
def create_thing(model, grant_type="password",
secret="", username=None, password=None,
is_broker_rest=False, is_broker=False, is_repo=False):
is_broker_rest=False, is_broker=False, is_repo=False, is_stanford2010=False, stanford2010_path=None):
"""
Creates and launches a thing which connects to the S3I
......@@ -383,7 +383,9 @@ def create_thing(model, grant_type="password",
password=password,
is_broker_rest=is_broker_rest,
is_broker=is_broker,
is_repo=is_repo
is_repo=is_repo,
is_stanford2010=is_stanford2010,
stanford2010_path=stanford2010_path
)
build(thing_ref, model)
......
"""This module implements the thing class which is the core element of
this package."""
import os
from os import listdir
from os.path import isfile, join, splitext
import threading
import json
import uuid
import time
from datetime import datetime
import copy
from s3i import IdentityProvider, TokenType, Directory
from s3i import Repository
import s3i.exception
from s3i.broker import Broker, BrokerREST
from s3i.messages import ServiceReply, GetValueReply, SetValueReply
from ml.tools import find_broker_endpoint
from ml.tools import find_broker_endpoint, XML
from ml.app_logger import APP_LOGGER
......@@ -38,6 +42,8 @@ def __init__(
is_broker=False,
is_repo=False,
is_broker_rest=True,
is_stanford2010=False,
stanford2010_path=None,
username=None,
password=None,
):
......@@ -56,6 +62,10 @@ def __init__(
:type is_broker_rest: bool
:param is_repo: Whether the thing uses the S³I Repository to launch its Digital Twin in the cloud
:type is_repo: bool
:param is_stanford2010: Whether the thing has a extern resource in form of a .hpr data
:type is_stanford2010: bool
:param stanford2010_path: absolute path of stanford2010 (hpr) data
:type stanford2010_path: string
:param username: OAuth 2.0 specified username, registered in the S³I IdentityProvider. If the grant_type is set as password, the username is required
:type username: str
:param password: OAuth 2.0 specified password, registered in the S³I IdentityProvider. If the grant_type is set as password, the password is required
......@@ -73,7 +83,9 @@ def __init__(
self.__is_broker = is_broker
self.__is_broker_rest = is_broker_rest
self.__is_repo = is_repo
self.__is_stanford2010 = is_stanford2010
self.__stanford2010_path = stanford2010_path
self.__access_token = ""
self.__endpoint = ""
......@@ -90,7 +102,6 @@ def __init__(
self.dir_json = dict()
self.dt_json = dict()
attributes = model.get("attributes", None)
self.__name = ""
self.__roles = {}
self.__features = {}
......@@ -98,8 +109,10 @@ def __init__(
# ??? Is this property necessary? Only used as return value in _getValue()
self.__resGetValue = list()
self.__source_obj = None
attributes = model.get("attributes", None)
if attributes:
self.__name = attributes.get("name", "")
self.__stanford2010 = None
@property
def model(self):
......@@ -112,6 +125,14 @@ def model(self):
return self.__model
@property
def stanford2010(self):
return self.__stanford2010
@stanford2010.setter
def stanford2010(self, value):
self.__stanford2010 = value
@property
def ditto_features(self):
return self.__ditto_features
......@@ -237,6 +258,8 @@ def run_forever(self):
threading.Thread(target=self.__json_syn).start()
if self.__is_repo:
threading.Thread(target=self.__repo_syn).start()
if self.__is_stanford2010:
threading.Thread(target=self.__stanford2010_syn, args=(self.__stanford2010_path, True, 10)).start()
@staticmethod
def add_user_def(func):
......@@ -293,6 +316,65 @@ def __repo_syn(self, freq=0.1):
except:
continue
def __stanford2010_syn(self, path, is_period=False, freq=10):
"""
This function finds the .hpr data according to the entered path and read out the content
"""
def get_hpr_files(path):
# get all .hpr files under the specified folder
files = [f for f in listdir(path) if isfile(join(path, f))]
hpr_files = []
for file in files:
filename, file_extension = splitext(file)
if file_extension == ".hpr": # in the folder, there is always one hpr data
hpr_files.append(file)
return hpr_files
def get_timestamp(hpr):
time_node = hpr.find_nodes("HarvestedProductionHeader/CreationDate")[0]
time_str = time_node.text
return time_str.replace(time_str[time_str.rfind("."):time_str.find("+")], "")
def get_last_hpr(hpr_files):
# get the last generated hpr data
if len(hpr_files) == 0:
return None
hpr = XML(path="{0}/{1}".format(path, hpr_files[0]))
os.remove("{0}/{1}".format(path, hpr_files[0]))
hpr_files.pop(0)
for filename in hpr_files:
hpr_temp = XML(path="{0}/{1}".format(path, filename))
os.remove("{0}/{1}".format(path, filename))
hpr_timestamp = get_timestamp(hpr)
hpr_temp_timestamp = get_timestamp(hpr)
hpr_isotime = datetime.fromisoformat(hpr_timestamp)
hpr_temp_isotime = datetime.fromisoformat(hpr_temp_timestamp)
if hpr_isotime < hpr_temp_isotime:
hpr = hpr_temp
return hpr
_stanford2010 = get_last_hpr(get_hpr_files(path))
if _stanford2010 is None:
APP_LOGGER.warn("StanForD2010 Data not found")
return
self.__stanford2010 = _stanford2010
APP_LOGGER.info("Parsing the StanForD2010 Data in DT")
while is_period:
time.sleep(freq)
hpr_temp = get_last_hpr(hpr_files=get_hpr_files(path))
if hpr_temp is None:
continue
isotime_old = datetime.fromisoformat(get_timestamp(self.__stanford2010))
isotime_new = datetime.fromisoformat(get_timestamp(hpr_temp))
if isotime_old == isotime_new:
continue
else:
APP_LOGGER.info("Update the StanForD2010 data in DT")
self.__stanford2010 = hpr_temp
def __connect_with_idp(self):
"""Establishes a connection to the S³I IdentityProvider which guarantees,
that the JSON web token needed to use s3i services.
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment