KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use source_type method in localstack
Best Python code snippet using localstack_python
data.py
Source:data.py 
1import json, os2import math, copy, time3import numpy as np4from collections import defaultdict5import pandas as pd6from utils.utils import *7import math8from tqdm import tqdm9import seaborn as sb10import matplotlib.pyplot as plt11import matplotlib.cm as cm12import dill13from functools import partial14import multiprocessing as mp15class Graph():16 def __init__(self):17 super(Graph, self).__init__()18 '''19 node_forward and node_backward are only used when building the data. 20 Afterwards will be transformed into node_feature by DataFrame21 22 node_forward: name -> node_id23 node_backward: node_id -> feature_dict24 node_feature: a DataFrame containing all features25 '''26 self.node_forward = defaultdict(lambda: {})27 self.node_backward = defaultdict(lambda: [])28 self.node_feature = defaultdict(lambda: [])29 '''30 edge_list: index the adjacancy matrix (time) by 31 <target_type, source_type, relation_type, target_id, source_id>32 '''33 self.edge_list = defaultdict( #target_type34 lambda: defaultdict( #source_type35 lambda: defaultdict( #relation_type36 lambda: defaultdict( #target_id37 lambda: defaultdict( #source_id(38 lambda: int # time39 )))))40 self.times = {}41 def add_node(self, node):42 nfl = self.node_forward[node['type']]43 if node['id'] not in nfl:44 self.node_backward[node['type']] += [node]45 ser = len(nfl)46 nfl[node['id']] = ser47 return ser48 return nfl[node['id']]49 def add_edge(self, source_node, target_node, time = None, relation_type = None, directed = True):50 edge = [self.add_node(source_node), self.add_node(target_node)]51 '''52 Add bi-directional edges with different relation type53 '''54 self.edge_list[target_node['type']][source_node['type']][relation_type][edge[1]][edge[0]] = time55 if directed:56 self.edge_list[source_node['type']][target_node['type']]['rev_' + relation_type][edge[0]][edge[1]] = time57 else:58 self.edge_list[source_node['type']][target_node['type']][relation_type][edge[0]][edge[1]] = time59 self.times[time] = True60 61 def update_node(self, node):62 nbl = self.node_backward[node['type']]63 ser = self.add_node(node)64 for k in node:65 if k not in nbl[ser]:66 nbl[ser][k] = node[k]67 def get_meta_graph(self):68 types = self.get_types()69 metas = []70 for target_type in self.edge_list:71 for source_type in self.edge_list[target_type]:72 for r_type in self.edge_list[target_type][source_type]:73 metas += [(target_type, source_type, r_type)]74 return metas75 76 def get_types(self):77 return list(self.node_feature.keys())78def sample_subgraph(graph, time_range, sampled_depth = 2, sampled_number = 8, inp = None, feature_extractor = feature_OAG):79 '''80 Sample Sub-Graph based on the connection of other nodes with currently sampled nodes81 We maintain budgets for each node type, indexed by <node_id, time>.82 Currently sampled nodes are stored in layer_data.83 After nodes are sampled, we construct the sampled adjacancy matrix.84 '''85 layer_data = defaultdict( #target_type86 lambda: {} # {target_id: [ser, time]}87 )88 budget = defaultdict( #source_type89 lambda: defaultdict( #source_id90 lambda: [0., 0] #[sampled_score, time]91 ))92 new_layer_adj = defaultdict( #target_type93 lambda: defaultdict( #source_type94 lambda: defaultdict( #relation_type95 lambda: [] #[target_id, source_id]96 )))97 '''98 For each node being sampled, we find out all its neighborhood, 99 adding the degree count of these nodes in the budget.100 Note that there exist some nodes that have many neighborhoods101 (such as fields, venues), for those case, we only consider 102 '''103 def add_budget(te, target_id, target_time, layer_data, budget):104 for source_type in te:105 tes = te[source_type]106 for relation_type in tes:107 if relation_type == 'self' or target_id not in tes[relation_type]:108 continue109 adl = tes[relation_type][target_id]110 if len(adl) < sampled_number:111 sampled_ids = list(adl.keys())112 else:113 sampled_ids = np.random.choice(list(adl.keys()), sampled_number, replace = False)114 for source_id in sampled_ids:115 source_time = adl[source_id]116 if source_time == None:117 source_time = target_time118 if source_time > np.max(list(time_range.keys())) or source_id in layer_data[source_type]:119 continue120 budget[source_type][source_id][0] += 1. / len(sampled_ids)121 budget[source_type][source_id][1] = source_time122 '''123 First adding the sampled nodes then updating budget.124 '''125 for _type in inp:126 for _id, _time in inp[_type]:127 layer_data[_type][_id] = [len(layer_data[_type]), _time]128 for _type in inp:129 te = graph.edge_list[_type]130 for _id, _time in inp[_type]:131 add_budget(te, _id, _time, layer_data, budget)132 '''133 We recursively expand the sampled graph by sampled_depth.134 Each time we sample a fixed number of nodes for each budget,135 based on the accumulated degree.136 '''137 for layer in range(sampled_depth):138 sts = list(budget.keys())139 for source_type in sts:140 te = graph.edge_list[source_type]141 keys = np.array(list(budget[source_type].keys()))142 if sampled_number > len(keys):143 '''144 Directly sample all the nodes145 '''146 sampled_ids = np.arange(len(keys))147 else:148 '''149 Sample based on accumulated degree150 '''151 score = np.array(list(budget[source_type].values()))[:,0] ** 2152 score = score / np.sum(score)153 sampled_ids = np.random.choice(len(score), sampled_number, p = score, replace = False) 154 sampled_keys = keys[sampled_ids]155 '''156 First adding the sampled nodes then updating budget.157 '''158 for k in sampled_keys:159 layer_data[source_type][k] = [len(layer_data[source_type]), budget[source_type][k][1]]160 for k in sampled_keys:161 add_budget(te, k, budget[source_type][k][1], layer_data, budget)162 budget[source_type].pop(k) 163 '''164 Prepare feature, time and adjacency matrix for the sampled graph165 '''166 feature, times, indxs, texts = feature_extractor(layer_data, graph)167 168 edge_list = defaultdict( #target_type169 lambda: defaultdict( #source_type170 lambda: defaultdict( #relation_type171 lambda: [] # [target_id, source_id] 172 )))173 for _type in layer_data:174 for _key in layer_data[_type]:175 _ser = layer_data[_type][_key][0]176 edge_list[_type][_type]['self'] += [[_ser, _ser]]177 '''178 Reconstruct sampled adjacancy matrix by checking whether each179 link exist in the original graph180 '''181 for target_type in graph.edge_list:182 te = graph.edge_list[target_type]183 tld = layer_data[target_type]184 for source_type in te:185 tes = te[source_type]186 sld = layer_data[source_type]187 for relation_type in tes:188 tesr = tes[relation_type]189 for target_key in tld:190 if target_key not in tesr:191 continue192 target_ser = tld[target_key][0]193 for source_key in tesr[target_key]:194 '''195 Check whether each link (target_id, source_id) exist in original adjacancy matrix196 '''197 if source_key in sld:198 source_ser = sld[source_key][0]199 edge_list[target_type][source_type][relation_type] += [[target_ser, source_ser]]200 return feature, times, edge_list, indxs, texts201def to_torch(feature, time, edge_list, graph):202 '''203 Transform a sampled sub-graph into pytorch Tensor204 node_dict: {node_type: <node_number, node_type_ID>} node_number is used to trace back the nodes in original graph.205 edge_dict: {edge_type: edge_type_ID}206 '''207 node_dict = {}208 node_feature = []209 node_type = []210 node_time = []211 edge_index = []212 edge_type = []213 edge_time = []214 215 node_num = 0216 types = graph.get_types()217 for t in types:218 node_dict[t] = [node_num, len(node_dict)]219 node_num += len(feature[t])220 for t in types:221 node_feature += list(feature[t])222 node_time += list(time[t])223 node_type += [node_dict[t][1] for _ in range(len(feature[t]))]224 225 edge_dict = {e[2]: i for i, e in enumerate(graph.get_meta_graph())}226 edge_dict['self'] = len(edge_dict)227 for target_type in edge_list:228 for source_type in edge_list[target_type]:229 for relation_type in edge_list[target_type][source_type]:230 for ii, (ti, si) in enumerate(edge_list[target_type][source_type][relation_type]):231 tid, sid = ti + node_dict[target_type][0], si + node_dict[source_type][0]232 edge_index += [[sid, tid]]233 edge_type += [edge_dict[relation_type]] 234 '''235 Our time ranges from 1900 - 2020, largest span is 120.236 '''237 edge_time += [node_time[tid] - node_time[sid] + 120]238 node_feature = torch.FloatTensor(node_feature)239 node_type = torch.LongTensor(node_type)240 edge_time = torch.LongTensor(edge_time)241 edge_index = torch.LongTensor(edge_index).t()242 edge_type = torch.LongTensor(edge_type)243 return node_feature, node_type, edge_time, edge_index, edge_type, node_dict, edge_dict244 245 246class RenameUnpickler(dill.Unpickler):247 def find_class(self, module, name):248 renamed_module = module249 if module == "GPT_GNN.data" or module == 'data':250 renamed_module = "pyHGT.data"251 return super(RenameUnpickler, self).find_class(renamed_module, name)252def renamed_load(file_obj):...
ma.py
Source:ma.py
1from typing import Union2import numpy as np3from jesse.helpers import get_candle_source, slice_candles4def ma(candles: np.ndarray, period: int = 30, matype: int = 0, source_type: str = "close", sequential: bool = False) -> Union[5 float, np.ndarray]:6 """7 MA - (nearly) All Moving Averages of Jesse8 :param candles: np.ndarray9 :param period: int - default: 3010 :param matype: int - default: 011 :param source_type: str - default: "close"12 :param sequential: bool - default: False13 :return: float | np.ndarray14 """15 candles = slice_candles(candles, sequential)16 if matype <= 8:17 from talib import MA18 if len(candles.shape) != 1:19 candles = get_candle_source(candles, source_type=source_type)20 res = MA(candles, timeperiod=period, matype=matype)21 elif matype == 9:22 from . import fwma23 res = fwma(candles, period, source_type=source_type, sequential=True)24 elif matype == 10:25 from . import hma26 res = hma(candles, period, source_type=source_type, sequential=True)27 elif matype == 11:28 from talib import LINEARREG29 if len(candles.shape) != 1:30 candles = get_candle_source(candles, source_type=source_type)31 res = LINEARREG(candles, period)32 elif matype == 12:33 from . import wilders34 res = wilders(candles, period, source_type=source_type, sequential=True)35 elif matype == 13:36 from . import sinwma37 res = sinwma(candles, period, source_type=source_type, sequential=True)38 elif matype == 14:39 from . import supersmoother40 res = supersmoother(candles, period, source_type=source_type, sequential=True)41 elif matype == 15:42 from . import supersmoother_3_pole43 res = supersmoother_3_pole(candles, period, source_type=source_type, sequential=True)44 elif matype == 16:45 from . import gauss46 res = gauss(candles, period, source_type=source_type, sequential=True)47 elif matype == 17:48 from . import high_pass49 res = high_pass(candles, period, source_type=source_type, sequential=True)50 elif matype == 18:51 from . import high_pass_2_pole52 res = high_pass_2_pole(candles, period, source_type=source_type, sequential=True)53 elif matype == 19:54 from talib import HT_TRENDLINE55 if len(candles.shape) != 1:56 candles = get_candle_source(candles, source_type=source_type)57 res = HT_TRENDLINE(candles)58 elif matype == 20:59 from . import jma60 res = jma(candles, period, source_type=source_type, sequential=True)61 elif matype == 21:62 from . import reflex63 res = reflex(candles, period, source_type=source_type, sequential=True)64 elif matype == 22:65 from . import trendflex66 res = trendflex(candles, period, source_type=source_type, sequential=True)67 elif matype == 23:68 from . import smma69 res = smma(candles, period, source_type=source_type, sequential=True)70 elif matype == 24:71 if len(candles.shape) == 1:72 raise ValueError("vwma only works with normal candles.")73 from . import vwma74 res = vwma(candles, period, source_type=source_type, sequential=True)75 elif matype == 25:76 from . import pwma77 res = pwma(candles, period, source_type=source_type, sequential=True)78 elif matype == 26:79 from . import swma80 res = swma(candles, period, source_type=source_type, sequential=True)81 elif matype == 27:82 from . import alma83 res = alma(candles, period, source_type=source_type, sequential=True)84 elif matype == 28:85 from . import hwma86 res = hwma(candles, source_type=source_type, sequential=True)87 elif matype == 29:88 from . import vwap89 if len(candles.shape) == 1:90 raise ValueError("vwap only works with normal candles.")91 res = vwap(candles, source_type=source_type, sequential=True)92 elif matype == 30:93 from . import nma94 res = nma(candles, period, source_type=source_type, sequential=True)95 elif matype == 31:96 from . import edcf97 res = edcf(candles, period, source_type=source_type, sequential=True)98 elif matype == 32:99 from . import mwdx100 res = mwdx(candles, source_type=source_type, sequential=True)101 elif matype == 33:102 from . import maaq103 res = maaq(candles, period, source_type=source_type, sequential=True)104 elif matype == 34:105 from . import srwma106 res = srwma(candles, period, source_type=source_type, sequential=True)107 elif matype == 35:108 from . import sqwma109 res = sqwma(candles, period, source_type=source_type, sequential=True)110 elif matype == 36:111 from . import vpwma112 res = vpwma(candles, period, source_type=source_type, sequential=True)113 elif matype == 37:114 from . import cwma115 res = cwma(candles, period, source_type=source_type, sequential=True)116 elif matype == 38:117 from . import jsa118 res = jsa(candles, period, source_type=source_type, sequential=True)119 elif matype == 39:120 from . import epma121 res = epma(candles, period, source_type=source_type, sequential=True)...
source_type_test.py
Source:source_type_test.py
1# -*- coding: utf-8 -*-2"""Tests for the source type objects."""3import unittest4from artifacts import errors5from artifacts import source_type6class TestSourceType(source_type.SourceType):7 """Class that implements a test source type."""8 TYPE_INDICATOR = u'test'9 def __init__(self, test=None):10 """Initializes the source type object.11 Args:12 test: optional test string. The default is None.13 Raises:14 FormatError: when test is not set.15 """16 if not test:17 raise errors.FormatError(u'Missing test value.')18 super(TestSourceType, self).__init__()19 self.test = test20 def CopyToDict(self):21 """Copies the source type to a dictionary.22 Returns:23 A dictionary containing the source type attributes.24 """25 return {u'test': self.test}26class SourceTypeTest(unittest.TestCase):27 """Class to test the artifact source type."""28class ArtifactGroupSourceTypeTest(unittest.TestCase):29 """Class to test the artifact group source type."""30 def testInitialize(self):31 """Tests the __init__ function."""32 source_type.ArtifactGroupSourceType(names=[u'test'])33class FileSourceTypeTest(unittest.TestCase):34 """Class to test the files source type."""35 def testInitialize(self):36 """Tests the __init__ function."""37 source_type.FileSourceType(paths=[u'test'])38 source_type.FileSourceType(paths=[u'test'], separator=u'\\')39class PathSourceTypeTest(unittest.TestCase):40 """Class to test the paths source type."""41 def testInitialize(self):42 """Tests the __init__ function."""43 source_type.PathSourceType(paths=[u'test'])44 source_type.PathSourceType(paths=[u'test'], separator=u'\\')45class WindowsRegistryKeySourceTypeTest(unittest.TestCase):46 """Class to test the Windows Registry keys source type."""47 def testInitialize(self):48 """Tests the __init__ function."""49 source_type.WindowsRegistryKeySourceType(keys=[u'HKEY_LOCAL_MACHINE\\test'])50 with self.assertRaises(errors.FormatError):51 source_type.WindowsRegistryKeySourceType(keys=u'HKEY_LOCAL_MACHINE\\test')52class WindowsRegistryValueSourceTypeTest(unittest.TestCase):53 """Class to test the Windows Registry value source type."""54 def testInitialize(self):55 """Tests the __init__ function."""56 key_value_pair = {'key': u'HKEY_LOCAL_MACHINE\\test', 'value': u'test'}57 source_type.WindowsRegistryValueSourceType(key_value_pairs=[key_value_pair])58 key_value_pair = {'bad': u'test', 'value': u'test'}59 with self.assertRaises(errors.FormatError):60 source_type.WindowsRegistryValueSourceType(61 key_value_pairs=[key_value_pair])62 with self.assertRaises(errors.FormatError):63 source_type.WindowsRegistryValueSourceType(key_value_pairs=key_value_pair)64class WMIQuerySourceTypeTest(unittest.TestCase):65 """Class to test the WMI query source type."""66 def testInitialize(self):67 """Tests the __init__ function."""68 source_type.WMIQuerySourceType(query=u'test')69class SourceTypeFactoryTest(unittest.TestCase):70 """Class to test the source type factory."""71 def testCreateSourceType(self):72 """Tests the source type creation."""73 source_type.SourceTypeFactory.RegisterSourceTypes([TestSourceType])74 with self.assertRaises(KeyError):75 source_type.SourceTypeFactory.RegisterSourceTypes([TestSourceType])76 source_object = source_type.SourceTypeFactory.CreateSourceType(77 u'test', {u'test': u'test123'})78 self.assertIsNotNone(source_object)79 self.assertEqual(source_object.test, u'test123')80 with self.assertRaises(errors.FormatError):81 source_object = source_type.SourceTypeFactory.CreateSourceType(u'test',82 {})83 with self.assertRaises(errors.FormatError):84 source_object = source_type.SourceTypeFactory.CreateSourceType(u'bogus',85 {})86 source_type.SourceTypeFactory.DeregisterSourceType(TestSourceType)87 def testRegisterSourceType(self):88 """Tests the source type registration functions."""89 expected_number_of_source_types = len(90 source_type.SourceTypeFactory.GetSourceTypes())91 source_type.SourceTypeFactory.RegisterSourceType(TestSourceType)92 number_of_source_types = len(source_type.SourceTypeFactory.GetSourceTypes())93 self.assertEqual(number_of_source_types,94 expected_number_of_source_types + 1)95 source_type.SourceTypeFactory.DeregisterSourceType(TestSourceType)96 number_of_source_types = len(source_type.SourceTypeFactory.GetSourceTypes())97 self.assertEqual(number_of_source_types, expected_number_of_source_types)98 def testRegisterSourceTypeRaisesWhenAlreadyRegistered(self):99 """Tests the source type registration functions when already registered."""100 source_type.SourceTypeFactory.RegisterSourceType(TestSourceType)101 with self.assertRaises(KeyError):102 source_type.SourceTypeFactory.RegisterSourceType(TestSourceType)103 source_type.SourceTypeFactory.DeregisterSourceType(TestSourceType)104if __name__ == '__main__':...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
