KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use update_domain method in tempest

Best Python code snippet using tempest_python

integrator.py

Source:integrator.py

...164                - self.stage1(), self.stage2() etc. depending on the number of165                  stages available.166                - self.compute_accelerations(index=0, update_nnps=True)167                - self.do_post_stage(stage_dt, stage_count_from_1)168                - self.update_domain()169        Please see any of the concrete implementations of the Integrator class170        to study.  By default the Integrator implements a171        predict-evaluate-correct method, the same as PECIntegrator.172        """173        self.initialize()174        # Predict175        self.stage1()176        self.update_domain()177        # Call any post-stage functions.178        self.do_post_stage(0.5*dt, 1)179        self.compute_accelerations()180        # Correct181        self.stage2()182        self.update_domain()183        # Call any post-stage functions.184        self.do_post_stage(dt, 2)185    def set_compiled_object(self, c_integrator):186        """Set the high-performance compiled object to call internally.187        """188        self.c_integrator = c_integrator189    def set_parallel_manager(self, pm):190        self.parallel_manager = pm191        self.c_integrator.set_parallel_manager(pm)192    def set_post_stage_callback(self, callback):193        """This callback is called when the particles are moved, i.e194        one stage of the integration is done.195        This callback is passed the current time value, the timestep and the196        stage.197        The current time value is  t + stage_dt, for example this would be198        0.5*dt for a two stage predictor corrector integrator.199        """200        self.c_integrator.set_post_stage_callback(callback)201    def step(self, time, dt):202        """This function is called by the solver.203        To implement the integration step please override the204        ``one_timestep`` method.205        """206        self.c_integrator.step(time, dt)207    def compute_accelerations(self, index=0, update_nnps=True):208        if update_nnps:209            # update NNPS since particles have moved210            if self.parallel_manager:211                self.parallel_manager.update()212            self.nnps.update()213        # Evaluate214        c_integrator = self.c_integrator215        a_eval = self.acceleration_evals[index]216        a_eval.compute(c_integrator.t, c_integrator.dt)217    def initial_acceleration(self, t, dt):218        """Compute the initial accelerations if needed before the iterations start.219        The default implementation only does this for the first acceleration220        evaluator. So if you have multiple evaluators, you must override this221        method in a subclass.222        """223        self.acceleration_evals[0].compute(t, dt)224    def update_domain(self):225        """Update the domain of the simulation.226        This is to be called when particles move so the ghost particles227        (periodicity, mirror boundary conditions) can be reset. Further, this228        also recalculates the appropriate cell size based on the particle229        kernel radius, `h`. This should be called explicitly when desired but230        usually this is done when the particles are moved or the `h` is231        changed.232        The integrator should explicitly call this when needed in the233        `one_timestep` method.234        """235        self.nnps.update_domain()236###############################################################################237# `EulerIntegrator` class238###############################################################################239class EulerIntegrator(Integrator):240    def one_timestep(self, t, dt):241        self.compute_accelerations()242        self.stage1()243        self.update_domain()244        self.do_post_stage(dt, 1)245###############################################################################246# `PECIntegrator` class247###############################################################################248class PECIntegrator(Integrator):249    r"""250    In the Predict-Evaluate-Correct (PEC) mode, the system is advanced using:251    .. math::252        y^{n+\frac{1}{2}} = y^n + \frac{\Delta t}{2}F(y^{n-\frac{1}{2}})253        --> Predict254        F(y^{n+\frac{1}{2}}) --> Evaluate255        y^{n + 1} = y^n + \Delta t F(y^{n+\frac{1}{2}})256    """257    def one_timestep(self, t, dt):258        self.initialize()259        # Predict260        self.stage1()261        self.update_domain()262        # Call any post-stage functions.263        self.do_post_stage(0.5*dt, 1)264        self.compute_accelerations()265        # Correct266        self.stage2()267        self.update_domain()268        # Call any post-stage functions.269        self.do_post_stage(dt, 2)270###############################################################################271# `EPECIntegrator` class272###############################################################################273class EPECIntegrator(Integrator):274    r"""275    Predictor corrector integrators can have two modes of276    operation.277    In the Evaluate-Predict-Evaluate-Correct (EPEC) mode, the278    system is advanced using:279    .. math::280        F(y^n) --> Evaluate281        y^{n+\frac{1}{2}} = y^n + F(y^n) --> Predict282        F(y^{n+\frac{1}{2}}) --> Evaluate283        y^{n+1} = y^n + \Delta t F(y^{n+\frac{1}{2}}) --> Correct284    Notes:285    The Evaluate stage of the integrator forces a function286    evaluation. Therefore, the PEC mode is much faster but relies on287    old accelertions for the Prediction stage.288    In the EPEC mode, the final corrector can be modified to:289    :math:`y^{n+1} = y^n + \frac{\Delta t}{2}\left( F(y^n) +290                                F(y^{n+\frac{1}{2}}) \right)`291    This would require additional storage for the accelerations.292    """293    def one_timestep(self, t, dt):294        self.initialize()295        self.compute_accelerations()296        # Predict297        self.stage1()298        self.update_domain()299        # Call any post-stage functions.300        self.do_post_stage(0.5*dt, 1)301        self.compute_accelerations()302        # Correct303        self.stage2()304        self.update_domain()305        # Call any post-stage functions.306        self.do_post_stage(dt, 2)307###############################################################################308# `TVDRK3Integrator` class309###############################################################################310class TVDRK3Integrator(Integrator):311    r"""312    In the TVD-RK3 integrator, the system is advanced using:313    .. math::314        y^{n + \frac{1}{3}} = y^n + \Delta t F( y^n )315        y^{n + \frac{2}{3}} = \frac{3}{4}y^n +316        \frac{1}{4}(y^{n + \frac{1}{3}} + \Delta t F(y^{n + \frac{1}{3}}))317        y^{n + 1} = \frac{1}{3}y^n + \frac{2}{3}(y^{n + \frac{2}{3}}318        + \Delta t F(y^{n + \frac{2}{3}}))319    """320    def one_timestep(self, t, dt):321        self.initialize()322        # stage 1323        self.compute_accelerations()324        self.stage1()325        self.update_domain()326        self.do_post_stage(1./3*dt, 1)327        # stage 2328        self.compute_accelerations()329        self.stage2()330        self.update_domain()331        self.do_post_stage(2./3*dt, 2)332        # stage 3 and end333        self.compute_accelerations()334        self.stage3()335        self.update_domain()336        self.do_post_stage(dt, 3)337###############################################################################338class LeapFrogIntegrator(PECIntegrator):339    r"""A leap-frog integrator.340    """341    def one_timestep(self, t, dt):342        self.stage1()343        self.update_domain()344        self.do_post_stage(0.5*dt, 1)345        self.compute_accelerations()346        self.stage2()347        self.update_domain()348        self.do_post_stage(dt, 2)349###############################################################################350class PEFRLIntegrator(Integrator):351    r"""A Position-Extended Forest-Ruth-Like integrator [Omeylan2002]_352    References353    ----------354    .. [Omeylan2002] I.M. Omelyan, I.M. Mryglod and R. Folk, "Optimized355       Forest-Ruth- and Suzuki-like algorithms for integration of motion356       in many-body systems", Computer Physics Communications 146, 188 (2002)357       http://arxiv.org/abs/cond-mat/0110585358    """359    def one_timestep(self, t, dt):360        self.stage1()361        self.update_domain()362        self.do_post_stage(0.1786178958448091*dt, 1)363        self.compute_accelerations()364        self.stage2()365        self.update_domain()366        self.do_post_stage(0.1123533131749906*dt, 2)367        self.compute_accelerations()368        self.stage3()369        self.update_domain()370        self.do_post_stage(0.8876466868250094*dt, 3)371        self.compute_accelerations()372        self.stage4()373        self.update_domain()374        self.do_post_stage(0.8213821041551909*dt, 4)375        self.compute_accelerations()376        self.stage5()377        self.update_domain()...

app.py

Source:app.py

1import subprocess2import requests3import shlex4import yaml5import os6import sys7import socket8from datetime import datetime9import dns.resolver10def configParse(config_file):11    api_key = None12    tld = None13    update_domain = None14    15    with open(config_file, 'r') as file:16        config = yaml.full_load(file)17    for key,value in config.items():18        if key == "API_KEY":19            api_key = value20        if key == "DOMAIN":21            tld = value22        if key == "FQDN":23            update_domain = value24    # Validate that we find an API key, tld and domain name25    if api_key is None or tld is None or update_domain is None:26        log("Config invalid. Correct the config or pass options as variables.")27        sys.exit(1)28    return(api_key, tld, update_domain)29def log(message):30    print("{} {}".format(datetime.now(), message))31def getExternalCurrentIP():32    domain = 'myip.opendns.com'33    resolver = dns.resolver.Resolver()34    resolver.nameservers = ['208.67.222.222', '208.67.220.220'] # opendns nameservers35    answer = resolver.query(domain)36    current_ip = str(answer[0])37    try:38        # Validate IP with builtin 39        socket.inet_aton(current_ip)40    except OSError:41        log("Unable to lookup current external IP. Answer returned {}".format(current_ip))42        sys.exit(1)43    return current_ip44def updateIP(currentExternalIP, api_key, tld, update_domain):45    46    headers = {'Content-Type':'application/json','Authorization': 'Bearer ' + api_key}47    get_url = "https://api.digitalocean.com/v2/domains/{}/records?name={}".format(tld, update_domain)48    try:49        r = requests.get(get_url, headers=headers, timeout=5)50    except TimeoutError as err:51        log("Timed out accessing DigitalOcean")52        print(err)53        sys.exit(1)54    if r.status_code == 200:55        try:56            content = r.json()57            if content['meta']['total'] == 0:58                log("No current DNS entry found for {}. Add a manual A record and then retry".format(update_domain))59                sys.exit(1)60            elif content['meta']['total'] > 1:61                log("Returned more than 1 IP for current hostname. Exiting as assume something is wrong.")62                sys.exit(1)63            else:64                record_id = content['domain_records'][0]['id']65                current_DNS_IP = content['domain_records'][0]['data']66                if content['domain_records'][0]['type'] == 'CNAME':67                    log("\nCurrent DNS record is set as a CNAME but we cannot change record type.\n\nPlease delete the record and retry.")68                    sys.exit(1)69        except ValueError as err:70            log("Unknown content returned when looking up existing IP")71            print(err)72            sys.exit(1)73    else:74        log("Failed to get current ip: {}".format(r.reason))75        print(r.text)76        sys.exit(1)77        78    if currentExternalIP == current_DNS_IP:79        # Current IP matches DNS IP80        log("No change in IP found.")81        sys.exit(0)82    else:83        put_url = "https://api.digitalocean.com/v2/domains/{}/records/{}".format(tld, record_id)84        payload = {"data": currentExternalIP}85        r = requests.put(put_url, headers=headers, json = payload)86        log("{} {}".format(r.text, r.status_code))87def main():88    config_file='/config/config.yaml'89    90    if not os.path.exists(config_file):91        if os.getenv("API_KEY"):92            api_key = os.environ['API_KEY']93        else:94            print("API_KEY environment is not set.")95            sys.exit(1)96            97        if os.getenv("DOMAIN"):98            tld = os.environ['DOMAIN']99        else:100            print("DOMAIN environment is not set.")101            sys.exit(1)102            103        if os.getenv("FQDN"):104            update_domain = os.environ['FQDN']105        else:106            print("FQDN environment is not set.")107            sys.exit(1)108    else:109        api_key, tld, update_domain = configParse(config_file)110    currentExternalIP = getExternalCurrentIP()111    updateIP(currentExternalIP, api_key, tld, update_domain)112if __name__ == "__main__":...

unary.py

Source:unary.py

...12		'''Sets infinite positive integer values for all L variables.'''13		domain = {"min": spec["minl"], "max": float("inf")}14		for var in X:15			if var[0] == "L":16				csp.update_domain(var, copy.deepcopy(domain))17	def __establish_holes_space(self, csp, spec):18		'''Makes L4, L5, and L6 consistent W.R.T. minimum node length.19		20		Nodes 4, 5, and 6 contain holes. Between the holes and nodes'21		junction with adjacent nodes must be a minumum space. Also,22		between the holes themselves must be a minimum space.23		'''		24		d = csp.get_domain("L4")25		new_d = copy.deepcopy(d)26		new_d["min"] = spec["hmarg"] * 3 + spec["holed"] * 227		csp.update_domain("L4", new_d)28		29		d = csp.get_domain("L5")30		new_d = copy.deepcopy(d)31		new_d["min"] =  spec["hmarg"] * 4 + spec["holed"] * 332		csp.update_domain("L5", new_d)33		34		d = csp.get_domain("L6")35		new_d = copy.deepcopy(d)36		new_d["min"] = spec["hmarg"] * 2 + spec["holed"] * 137		csp.update_domain("L6", new_d)38				39	def __establish_RTD(self, csp, catalog, X):40		diams = catalog.values("D")41		thicks = catalog.values("T")42		rounds = catalog.values("R")43		for var in X:44			if var[0] == "D":45				csp.update_domain(var, copy.deepcopy(diams))46			elif var[0] == "R":47				csp.update_domain(var, copy.deepcopy(rounds))48			elif var[0] == "T":49				csp.update_domain(var, copy.deepcopy(thicks))50	def __establish_D1(self, csp, spec):51		'''Makes D1 consistent W.R.T. minimum d1 minimum diameter.'''52		d1 = csp.get_domain("D1")53		legal_values = set([])54		for diam in d1:55			if diam >= spec["topd"]["min"] and diam <= spec["topd"]["max"]:56				legal_values.add(diam)...

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.