bDrwx

from typing import Optional


class Digit:
    def __init__(self, val: int, acc: Optional[str] = None):
        self.val = val
        self.acc = acc

    def set_val(self, val):
        self.val = val

bDrwx

 # Print n elems in r pow
 ' '.join((str(r**(i)) for i in range(n)))

bDrwx

def sum(arr):
  if not arr:
    return 0
  return arr.pop(0) + sum(arr)
  
def count(arr):
  if not arr:
    return 0
  return 1+count(arr[1:])
  

bDrwx

import re
import itertools as it

def get_file_seq_num(filename: str) -> int:
    ''' CF0001.D00'''
    match = re.search(r"CF(\d{4}).D00", filename)
    if match:
        return int(match.group(1))
    return None

bDrwx

last_cdr: str = ''

def callback(info):
    ls_out = info.split()
    if re.match(r'CF\d{4}.D00', ls_out[-1]):
        td = datetime.now() - datetime.strptime(" ".join(ls_out[:2]), '%H.%M:%S %d.%m.%Y') 
        if td <= timedelta(minutes=15):
            global last_cdr
            last_cdr = ls_out[-1]
    return None

bDrwx

# -*- coding: utf-8 -*-
from typing import List, Tuple, DefaultDict
from collections import deque, defaultdict
import click
import csv
from cdr import Gcdr, Subscriber, Dvo, Interfacez, UserType, CallType, Reg
from sqlalchemy import create_engine
from configparser import ConfigParser, ExtendedInterpolation
from sqlalchemy import (
    Table,

bDrwx

from airflow.contrib.sensors.file_sensor import FileSensor
from airflow.operators.dummy_operator    import DummyOperator
from airflow.contrib.sensors.gcs_sensor  import GoogleCloudStorageObjectSensor
import datetime
from datetime import date, timedelta
import airflow

default_args = {
    "depends_on_past" : False,
    "start_date"      : airflow.utils.dates.days_ago( 1 ),

bDrwx

 <cv-spread-overview :spread-data='[{"title":"\u041c\u043e\u0441\u043a\u0432\u0430","code":"RU-MOW","is_city":true,"coord_x":"13","coord_y":"60","sick":57300,"healed":5766,"died":658,"sick_incr":3561,"healed_incr":631,"died_incr":47},{"title":"\u041c\u043e\u0441\u043a\u043e\u0432\u0441\u043a\u0430\u044f \u043e\u0431\u043b\u0430\u0441\u0442\u044c","code":"RU-MOS","is_city":false,"coord_x":"13","coord_y":"61","sick":12507,"healed":458,"died":109,"sick_incr":797,"healed_incr":28,"died_incr":16},{"title":"\u0421\u0430\u043d\u043a\u0442-\u041f\u0435\u0442\u0435\u0440\u0431\u0443\u0440\u0433","code":"RU-SPE","is_city":true,"coord_x":"13","coord_y":"45","sick":4411,"healed":779,"died":33,"sick_incr":349,"healed_incr":0,"died_incr":4},{"title":"\u041d\u0438\u0436\u0435\u0433\u043e\u0440\u043e\u0434\u0441\u043a\u0430\u044f \u043e\u0431\u043b\u0430\u0441\u0442\u044c","code":"RU-NIZ","is_city":false,"coord_x":"17","coord_y":"63","sick":2123,"healed":289,"died":14,"sick_incr":179,"healed_incr":63,"died_incr":2},{"title":

bDrwx

import pytest
from sorm.cdr import Subscriber, UserType

@pytest.fixture(name="Tetra_users")
def tetra_user():
    return [
        (Subscriber(UserType.inner, '102025000075780046ff', 501, 501), '102025000075780046'),
        (Subscriber(UserType.inner, '0e20250000755995ffffffffff', 65535, 65535), '0e20250000755995'),
        (Subscriber(UserType.inner, '102025000075782418ffffffffff', 506, 65535), '102025000075782418'),
    ]

bDrwx

# pylint: disable=wrong-import-position
if sys.platform == 'cli':
    from serial.serialcli import Serial
else:
    import os
    # chose an implementation, depending on os
    if os.name == 'nt':  # sys.platform == 'win32':
        from serial.serialwin32 import Serial
    elif os.name == 'posix':
        from serial.serialposix import Serial, PosixPollSerial, VTIMESerial  # noqa