maxfischer2781

import threading, random


class StackContext:
    def __init__(self):
        self._state = threading.local()
        self._state.stack = []

    def __enter__(self):
        this_context = random.random()

maxfischer2781

from itertools import accumulate

a = ['a', '30', '-', 'foot', 'und', 'ete', 'cted']
b = ['a', '30-foot', 'undetected']


def alignment(a: "Iterable[str]", b: "Iterable[str]") -> "List[List[int]]":
    if ''.join(a) != ''.join(b):
        raise ValueError("a and b must be fragments of the same string")
    target_subs = accumulate(b)

maxfischer2781

from usim import Pipe, instant
from usim._primitives.notification import Notification


class MonitoredPipe(Pipe):
    def __init__(self, throughput: float):
        super().__init__(throughput)
        self._monitor = Notification()

    async def load(self):

maxfischer2781

from typing import NamedTuple, Optional, Deque, Any, Dict, AsyncIterable

from usim import Pipe, instant
from collections import deque
from usim._primitives.notification import Notification


class MonitoredPipeInfo(NamedTuple):
    requested_throughput: float
    available_throughput: float

maxfischer2781

#!/usr/bin/env python3
from typing import Dict


def switch(cases: Dict[str, str], default='UNDEFINED', separator='\n'):
    # ["IfThenElse(ac, am", "IfThenElse(bc, bm", "IfThenElse(cc, cm", ...]
    switch_expression = separator.join(
        f"IfThenElse({condition}, {on_match},"
        for condition, on_match in cases.items()
    )

maxfischer2781

#!/usr/bin/env python3
from typing import NamedTuple, Iterable, Dict, List

import argparse
import pathlib
import re


CLI = argparse.ArgumentParser()
CLI.add_argument(

maxfischer2781

#!/bin/bash -l
#
# Wrapper when executing condor jobs
# Run jobs via `bash -l` to force sourcing a login shell including all /etc/profile.d/*.sh scripts
#
exec "$@"

maxfischer2781

Notes on using HTCondor-CE with APEL accounting for WLCG

These are tales of An Admin getting APEL+HTCondor-CE to work at GridKa/FZK Tier 1.

As background information, see

• the HTCondor-CE documentation
• the HtCondorCe Accounting notes

Setting up HTCondor-CE for APEL

maxfischer2781

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import Generic, TypeVar

R = TypeVar('R', bound='Runner')
T = TypeVar('T')  # result type


class Command(ABC, Generic[T, R]):

maxfischer2781

"""
Script to recreate missing APEL blah/batch records from a condor history.d directory
"""
from typing import TextIO, TypeVar, List
import argparse
import pathlib
import subprocess
import os
import threading
import tempfile