text:
<repo_name>prev-reorganize/_lib_contextme_ner-mapper<file_sep>/pinkerton/providers/base.py
from aiohttp import ClientSession
from pinkerton.settings import DATA_PROVIDER_CLIENT_HEADERS

class BaseDataProvider:

    ALLOWED_ENTITY_TYPES = []

    @property
    def session(self):
        return ClientSession(headers=DATA_PROVIDER_CLIENT_HEADERS)

    async def search(self, query: str) -> list:
        '''
        Query method takes an entity (in most cases: string) and returns list
        of matches from external knowlegde database, like Wikipedia.org
        '''
        raise NotImplementedError

    def accepts(self, entity: object) -> bool:
        '''
        Returns true if data provider can return information about given entity
        '''
        entity_type = (
            getattr(entity, 'type', None) or entity.get('type', None)
        )
        return entity_type in self.ALLOWED_ENTITY_TYPES
<file_sep>/pinkerton/__init__.py
from pinkerton.settings import PROJECT_VERSION

__version__ = PROJECT_VERSION
<file_sep>/pinkerton/similarity/tests/test_lda.py
import pytest

from pinkerton.similarity.lda import LDASimilarity
from pinkerton.providers.wikipedia import WikipediaProvider
from pinkerton.providers.yandex_maps import YandexMapsProvider

@pytest.fixture
def wiki_provider():
    return WikipediaProvider()

@pytest.fixture
def ya_maps_provider():
    return YandexMapsProvider()

@pytest.fixture
def person_context():
    return '<NAME>, царь руси, по историческим данным - был тираном'

@pytest.fixture
def address_context():
    return 'Приезжаю в Россию в пятницу, встретимся в Санкт-Петербурге, на улице рубинштейна, 21'

@pytest.fixture
def organisation_context():
    return '''
    Меркель назвала санкции единственным способом надавить на Россию
    Канцлер считает, что действия Москвы нельзя оставить без последствий, но решить конфликт военным путем нельзя.
    Санкции являются единственным средством нажать на Россию, считает канцлер Германии А<NAME>, пишут «Ведомости». Как передает Deutsche Welle, она заявила это в интервью радиостанции rbb-Inforadio, которое вышло в эфир 9 сентября. По словам канцлера, непосредственное участие России в украинском конфликте «очень-очень очевидно». Она считает, что действия Москвы нельзя оставить без последствий, но решить конфликт военным путем нельзя. Как только ситуация на Украине нормализуется, санкции должны быть отменены, добавила Меркель.
    Канцлер также заявила, что правительство не намерено наращивать бюджет министерства обороны, несмотря на планы НАТО создать новые силы быстрого реагирования. По словам Меркель, дополнительные финансовые средства на армию появятся после вывода контингента ВС ФРГ из Афганистана.
    8 сентября Меркель заявила на закрытом заседании в правительстве, что санкции против России необходимо ввести, несмотря на достигнутую договоренность о прекращении огня. Об этом рассказали несколько участников заседания. По их словам, канцлер напомнила, что Россия уже неоднократно обманывала Запад, не выполняя обещания, а также добавила, что российские войска по-прежнему находятся на Украине. ЕС должен действовать более решительно, процитировали ее участники заседания.
    '''

@pytest.fixture
@pytest.mark.asyncio
async def person_search_results(wiki_provider):
    for e in [
        '<NAME>',
    ]:
        yield (await wiki_provider.search(query=e))

@pytest.fixture
@pytest.mark.asyncio
async def address_search_results(ya_maps_provider):
    for e in [
        'рубинштейна 21',
    ]:
        yield (await ya_maps_provider.search(query=e))

@pytest.fixture
@pytest.mark.asyncio
async def organisation_search_results(wiki_provider):
    for e in [
        'министерство обороны',
    ]:
        yield (await wiki_provider.search(query=e))

@pytest.fixture
def lda_comparator():
    return LDASimilarity()

@pytest.mark.asyncio
async def test_lda_score_for_person_entities(
    person_context,
    person_search_results,
    lda_comparator
):
    async for entities in person_search_results:
        if entities:
            scores = lda_comparator.score(entities, person_context)
            correct_query = entities[0]
            correct_query_score = scores[0][1]
            assert correct_query['title'] == '<NAME>'
            assert correct_query['context'].startswith(
                '<NAME>, прозванный Гро́зным'
            )
            assert round(correct_query_score) >= 0.87

@pytest.mark.asyncio
async def test_lda_score_for_address_entities(
    address_context,
    address_search_results,
    lda_comparator
):
    async for entities in address_search_results:
        if entities:
            scores = lda_comparator.score(entities, address_context)
            first_result, second_result = scores[0:2]
            assert first_result[0]['title'] == 'Россия, Санкт-Петербург, улица Рубинштейна, 21'
            assert first_result[1] > second_result[1]

@pytest.mark.asyncio
async def test_lda_score_for_organisation_entities(
    organisation_context,
    organisation_search_results,
    lda_comparator
):
    async for entities in organisation_search_results:
        if entities:
            scores = lda_comparator.score(entities, organisation_context)
            top_3_titles = [s[0]['title'] for s in scores[:3]]

            assert len(scores) > 3
            assert set(top_3_titles) & {
                'Министерство обороны Украины',
                'Министерство обороны Российской Федерации',
            }
<file_sep>/pinkerton/similarity/base.py
class BaseSimilarityComparator:

    def score(self, entities: list, context: str) -> list:
        '''
        Returns list with entities sorted by similarity score
        '''
        raise NotImplementedError
<file_sep>/pinkerton/settings.py
PROJECT_NAME = 'pinkerton'
PROJECT_NAME_CAPITALIZED = 'Pinkerton'
PROJECT_SHORT_DESCRIPTION = 'Named entity linking and disambiguation service'
PROJECT_VERSION = '0.0.1'

BIND_HOST = '127.0.0.1'
BIND_PORT = 4000

DATA_PROVIDER_CLIENT_USER_AGENT = (
    f'{PROJECT_NAME_CAPITALIZED}/{PROJECT_VERSION} ({PROJECT_SHORT_DESCRIPTION})'
)

DATA_PROVIDER_CLIENT_HEADERS = {
    'User-Agent': DATA_PROVIDER_CLIENT_USER_AGENT,
}

MAX_ENTITIES_PER_OBJECT = 3
<file_sep>/pinkerton/linker.py
from operator import itemgetter

from pinkerton.base import EntityType

class EntityLinker:

    '''
    Entity linker is a layer between objects extractor and data providers
    '''

    def __init__(self, extractor, comparator, providers):
        self.extractor = extractor
        self.comparator = comparator
        self.providers = providers

    def get_queryset(self, obj: dict) -> set:
        '''
        Returns list of strings, that can be used in search with data providers
        '''
        type = obj['type']
        fields = obj['fields']

        # default queryset is normalized forms of all entity spans
        queryset = set(
            span['normalized'] for span in obj['spans']
        )

        if type == EntityType.Person:
            # add person-specific search queries to queryset
            # mostly because a lot of wiki articles have titles
            # in different form (like, Lastname_Firstname and Firstname_Lastname)
            firstname, lastname = (
                fields.get('firstname', None),
                fields.get('lastname', None),
            )
            if firstname and lastname:
                queryset |= {
                    ' '.join((firstname, lastname)),
                    ' '.join((lastname, firstname))
                }

        return queryset

    async def lookup(self, obj: dict, context: str) -> list:
        '''
        Returns all search results from data providers as generator
        '''
        providers = (p for p in self.providers if p.accepts(obj))
        for query in self.get_queryset(obj):
            for provider in providers:
                entities = await provider.search(query=query)
                for match in self.comparator.score(entities, context=context):
                    yield match

    async def process(self, text: str) -> list:
        objects, spans = await self.extractor.extract(text)
        for obj in objects:
            yield obj, sorted(
                [
                    match async for match in self.lookup(obj, context=text)
                ],
                key=itemgetter(1),
                reverse=True,
            )
<file_sep>/pinkerton/app.py
from aiohttp import web

from pinkerton import views
from pinkerton.linker import EntityLinker
from pinkerton.similarity import LDASimilarity
from pinkerton.extractor import EntityExtractor
from pinkerton.providers import WikipediaProvider, YandexMapsProvider

app = web.Application()

'''
Entity linking setup
'''
app.linker = EntityLinker(
    extractor=EntityExtractor(
        api_base_url='https://natasha.b-labs.pro/api/',
    ),
    comparator=LDASimilarity(),
    providers=[
        WikipediaProvider(),
        YandexMapsProvider(),
    ],
)

'''
Routes setup
'''

app.router.add_get('/version', views.version)
app.router.add_post('/annotate', views.annotate)
<file_sep>/pinkerton/views.py
from aiohttp import web

from pinkerton.settings import (
    PROJECT_VERSION,
    MAX_ENTITIES_PER_OBJECT,
)
from pinkerton.utils import serialize_object

async def version(request):
    return web.json_response(data={
        'version': PROJECT_VERSION,
    })

async def annotate(request):
    form = await request.post()
    entities = [
        (serialize_object(obj), entities[:MAX_ENTITIES_PER_OBJECT])
        async for (obj, entities) in request.app.linker.process(form['text'])
    ]
    return web.json_response(data=entities)
<file_sep>/setup.py
from setuptools import setup, find_packages

REQUIREMENTS = [
    'aiohttp==2.0.7',
    'pymorphy2[fast]==0.8',
    'gensim==2.0.0',
    'regex>=2017.4.29',
    'b-labs-models>=2017.8.19',
]

setup(
    name='pinkerton',
    version='0.0.1',
    description='Named entity linking and disambiguation service for russian language',
    url='https://github.com/bureaucratic-labs/pinkerton',
    author='<NAME>',
    author_email='<EMAIL>',
    license='MIT',
    classifiers=[
        'Development Status :: 3 - Alpha',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: MIT License',
        'Programming Language :: Python :: 3.6',
    ],
    keywords='natural language processing, named entity linking, wikification',
    packages=find_packages(),
    install_requires=REQUIREMENTS,
)
<file_sep>/pinkerton/tests/test_utils.py
import pytest

from pinkerton.utils import (
    remove_unnecessary_chars,
    tokenize,
    normalize,
)

@pytest.fixture
def person_entity():
    return '<NAME>'

def test_remove_unnecessary_chars(person_entity):
    assert remove_unnecessary_chars(person_entity) == 'пётр первый'

def test_tokenize():
    assert tokenize('Петр родился в 1470 году') == [
        'пётр',
        'родиться',
        '1470',
        'год',
    ]

def test_normalize():
    tokens = tokenize('пётр первый вступил на престол в раннем возрасте')
    assert list(normalize(tokens)) == [
        'пётр',
        'один',
        'вступить',
        'престол',
        'ранний',
        'возраст',
    ]
<file_sep>/pinkerton/base.py
from enum import Enum

class EntityType(Enum):

    Person = 0
    Location = 1
    Organisation = 2
    Address = 3

class Entity:

    def __init__(self, type: EntityType, title: str, content: object):
        self.type = type
        self.title = title
        self.content = content

    @property
    def serialized(self):
        return {
            'title': self.title,
            'content': self.content,
        }
<file_sep>/pinkerton/similarity/lda.py
from gensim.corpora import Dictionary
from gensim.models import LdaModel

from pinkerton.utils import tokenize
from pinkerton.similarity.base import BaseSimilarityComparator

class LDASimilarity(BaseSimilarityComparator):

    def __init__(self, **kwargs):
        self.model_kwargs = kwargs or {
            'passes': 20,
            'num_topics': 100,
        }

    def score(self, entities: list, context: str) -> list:

        queries = [
            (i, q['context']) for i, q in enumerate(entities) if q['context']
        ]

        context = tokenize(context)

        dictionary = Dictionary([context])

        vectors = [
            dictionary.doc2bow(
                tokenize(q)
            ) for _, q in queries
        ]

        model = LdaModel(id2word=dictionary, **self.model_kwargs)

        ents = (
            entities[i] for i, _ in queries
        )

        scores = (
            model[vec][-1][1] for vec in vectors if model[vec]
        )

        results = zip(ents, scores)

        def sort_by_score(item):
            return item[1]

        return sorted(results, key=sort_by_score, reverse=True)
<file_sep>/pinkerton/tests/test_extractor.py
import pytest

from pinkerton.base import EntityType
from pinkerton.extractor import EntityExtractor

@pytest.fixture
def extractor():
    return EntityExtractor('https://natasha.b-labs.pro/api/')

@pytest.fixture
def test_text():
    return '''
    <NAME>, или просто Ваню, застал в расплох <NAME>
    '''

def test_correct_api_urls(extractor):
    assert extractor.api_extract_url == 'https://natasha.b-labs.pro/api/extract'
    assert extractor.api_version_url == 'https://natasha.b-labs.pro/api/version'

@pytest.mark.asyncio
async def test_that_api_returns_some_results(extractor, test_text):
    objects, spans = await extractor.extract(test_text)

    assert objects
    assert spans

    assert objects[0]['type'] == EntityType.Person
    assert objects[0]['fields']['firstname'] == 'Иван'
    assert objects[0]['fields']['middlename'] == 'Василиевич'
    assert objects[0]['fields']['lastname'] == 'Иванов'
<file_sep>/README.md
# Pinkerton [![Build Status](https://travis-ci.org/bureaucratic-labs/pinkerton.svg?branch=master)](https://travis-ci.org/bureaucratic-labs/pinkerton)

<img align="right" src="http://i.imgur.com/gRXgf48.png">

`Pinkerton` связывает именованные сущности, упоминаемые в тексте на естественном языке, с данными в внешних хранилищах данных (например, Wikipedia, DaData.ru и др.).

При разборе результатов, `Pinkerton` определяет контекст, в котом упоминается сущность и выбирает наиболее подходящий вариант ответа - так, например, можно быть уверенным, что для персоны `<NAME>`, которая упоминается в контексте художественного фильма, будет выбрана правильная статья из Wikipedia.

Общая цель этого проекта - сделать открытую и расширяемую альтернативу [Яндекс.Карточкам](https://yandex.ru/promo/yobject/)
<file_sep>/pinkerton/providers/tests/test_wikipedia.py
import pytest

@pytest.mark.asyncio
async def test_query_person(wiki_provider):
    results = await wiki_provider.search(query='пётр 1')
    assert {
        'context': 'Пётр I Алексе́евич, прозванный Вели́кий (30 мая [9 июня] '
        '1672 год — 28 января [8 февраля] 1725 год) — последний царь '
        'всея Руси (с 1682 года) и первый Император Всероссийский (с '
        '1721 года).',
        'source': 'https://ru.wikipedia.org/wiki/%D0%9F%D1%91%D1%82%D1%80_I',
        'title': 'Пётр I',
    } in results

@pytest.mark.asyncio
async def test_query_org(wiki_provider):
    results = await wiki_provider.search(query='петропавловская крепость')
    assert {
        'context': 'Петропа́вловская кре́пость — крепость в Санкт-Петербурге, '
        'расположенная на Заячьем острове, историческое ядро города.',
        'source': 'https://ru.wikipedia.org/wiki/%D0%9F%D0%B5%D1%82%D1%80%D0%BE%D0%BF%D0%B0%D0%B2%D0%BB%D0%BE%D0%B2%D1%81%D0%BA%D0%B0%D1%8F_%D0%BA%D1%80%D0%B5%D0%BF%D0%BE%D1%81%D1%82%D1%8C',
        'title': 'Петропавловская крепость'
    } in results

@pytest.mark.asyncio
async def test_query_with_limit(wiki_provider):
    results = await wiki_provider.search(query='пётр 1', limit=1)
    assert results == [
        {
            'context': 'Пётр I Алексе́евич, прозванный Вели́кий (30 мая [9 июня] 1672 '
            'год — 28 января [8 февраля] 1725 год) — последний царь всея Руси '
            '(с 1682 года) и первый Император Всероссийский (с 1721 года).',
            'source': 'https://ru.wikipedia.org/wiki/%D0%9F%D1%91%D1%82%D1%80_I',
            'title': 'Пётр I',
        }
    ]

def test_not_accepts_method(wiki_provider, address_entity):
    assert not wiki_provider.accepts(address_entity)

def test_accepts_method(wiki_provider, person_entity):
    assert wiki_provider.accepts(person_entity)
<file_sep>/pinkerton/tests/test_linker.py
import pytest

from pinkerton.linker import EntityLinker
from pinkerton.extractor import EntityExtractor
from pinkerton.similarity import LDASimilarity
from pinkerton.providers import WikipediaProvider

@pytest.fixture
def history_text():
    return '''
    <NAME>, царь всея руси, по историческим данным, был тираном
    '''

@pytest.fixture
def linker():
    return EntityLinker(
        extractor=EntityExtractor(
            api_base_url='https://natasha.b-labs.pro/api/'
        ),
        comparator=LDASimilarity(),
        providers=[
            WikipediaProvider(),
        ],
    )

@pytest.mark.asyncio
async def test_link_person(linker, history_text):
    async for obj, entities in linker.process(history_text):
        assert obj['fields']['firstname'] == 'Иван'
        assert obj['fields']['middlename'] == 'Василиевич'

        entity, score = entities[0]

        assert entity['title'] == '<NAME>'
        assert entity['source'] == 'https://ru.wikipedia.org/wiki/%D0%98%D0%B2%D0%B0%D0%BD_%D0%93%D1%80%D0%BE%D0%B7%D0%BD%D1%8B%D0%B9'
<file_sep>/pinkerton/providers/tests/conftest.py
import pytest

from pinkerton.base import Entity, EntityType
from pinkerton.providers.wikipedia import WikipediaProvider
from pinkerton.providers.yandex_maps import YandexMapsProvider

@pytest.fixture
def wiki_provider():
    return WikipediaProvider()

@pytest.fixture
def ya_maps_provider():
    return YandexMapsProvider()

@pytest.fixture
def address_entity():
    return Entity(EntityType.Address, 'спб, цветочная 21', {
        'house': 21,
        'street': 'цветочная',
        'street_descriptor': 'улица',
    })

@pytest.fixture
def person_entity():
    return Entity(EntityType.Person, '<NAME>', {
        'firstname': 'василий',
        'middlename': 'иванович',
    })
<file_sep>/pinkerton/similarity/__init__.py
from pinkerton.similarity.lda import LDASimilarity
<file_sep>/pinkerton/providers/yandex_maps.py
from pinkerton.base import EntityType
from pinkerton.providers.base import BaseDataProvider

class YandexMapsProvider(BaseDataProvider):

    ALLOWED_ENTITY_TYPES = [
        EntityType.Address,
    ]

    def __init__(self):
        self.api_base_url = 'https://geocode-maps.yandex.ru/1.x/'

    async def search(self, query: str, limit=100) -> list:

        params = {
            'geocode': query,
            'results': limit,
            'format': 'json',
        }

        async with self.session as session:
            async with session.get(
                self.api_base_url, params=params
            ) as response:
                response = await response.json()
                response = (
                    response
                    .get('response', {})
                    .get('GeoObjectCollection', {})
                    .get('featureMember', [])
                )
                items = (
                    item.get('GeoObject', {})
                    for item in response
                )
                response = [
                    {
                        'title': (
                            item
                            .get('metaDataProperty')
                            .get('GeocoderMetaData')
                            .get('text')
                        ),
                        'context': item['description'],
                        'source': item,
                    } for item in items
                ]
                return response
<file_sep>/pinkerton/providers/tests/test_yandex_maps.py
import pytest

@pytest.mark.asyncio
async def test_query_address(ya_maps_provider, address_entity):
    results = await ya_maps_provider.search(query=address_entity.title)
    items = [
        {
            'title': item['title'],
            'context': item['context'],
        } for item in results
    ]
    assert {
        'title': 'Россия, Санкт-Петербург, Цветочная улица, 21',
        'context': 'Санкт-Петербург, Россия',
    } in items
<file_sep>/requirements.txt
aiohttp==2.0.7
pymorphy2[fast]==0.8
gensim==2.0.0
regex>=2017.4.29
b-labs-models>=2017.8.19
pytest==3.0.7
pytest-asyncio==0.5.0
<file_sep>/pinkerton/extractor.py
from urllib.parse import urljoin
from aiohttp import ClientSession

from pinkerton.base import EntityType
from pinkerton.settings import DATA_PROVIDER_CLIENT_HEADERS

class EntityExtractor:

    '''
    Provides API to Natasha playground
    '''

    ENTITIES_CONVERT_MAP = {
        'person': EntityType.Person,
        'organisation': EntityType.Organisation,
        'location': EntityType.Location,
        'street': EntityType.Location,
        'address': EntityType.Address,
    }

    def __init__(self, api_base_url: str):
        self.api_base_url = api_base_url
        self.api_extract_url = urljoin(self.api_base_url, 'extract')
        self.api_version_url = urljoin(self.api_base_url, 'version')

    @property
    def session(self):
        return ClientSession(headers=DATA_PROVIDER_CLIENT_HEADERS)

    async def extract(self, text: str) -> (list, list):
        '''
        Returns two lists: with extracted objects (\w solved coreference)
        and a list with extracted spans (with raw morphological info)
        '''
        async with self.session as session:
            async with session.post(self.api_extract_url, data={
                'text': text,
            }) as response:
                response = await response.json()
                objects, spans = response['objects'], response['spans']

                for obj in objects:
                    obj['type'] = self.ENTITIES_CONVERT_MAP[obj['type']]

                return objects, spans
<file_sep>/pinkerton/providers/__init__.py
from pinkerton.providers.wikipedia import WikipediaProvider
from pinkerton.providers.yandex_maps import YandexMapsProvider
<file_sep>/pinkerton/__main__.py
from aiohttp import web

from pinkerton.app import app
from pinkerton.settings import BIND_HOST, BIND_PORT

if __name__ == '__main__':
    web.run_app(app, host=BIND_HOST, port=BIND_PORT)
<file_sep>/pinkerton/providers/wikipedia.py
from pinkerton.base import EntityType
from pinkerton.providers.base import BaseDataProvider

class WikipediaProvider(BaseDataProvider):

    ALLOWED_ENTITY_TYPES = [
        EntityType.Person,
        EntityType.Location,
        EntityType.Organisation,
    ]

    def __init__(self, language='ru'):
        self.api_base_url = f'https://{language}.wikipedia.org/w/api.php'

    async def search(self, query: str, limit=100) -> list:

        params = {
            'utf8': 'true',
            'format': 'json',
            'action': 'opensearch',
            'redirects': 'resolve',
            'limit': limit,
            'search': query,
        }

        async with self.session as session:
            async with session.get(
                self.api_base_url, params=params
            ) as response:
                _, titles, descriptions, links = await response.json()
                return [
                    {
                        'title': t,
                        'context': d,
                        'source': l,
                    } for (t, d, l) in zip(titles, descriptions, links)
                ]
<file_sep>/pinkerton/utils.py
import regex as re
import pymorphy2
import unicodedata
import b_labs_models as models

morph = pymorphy2.MorphAnalyzer()
tokenizer = models.Tokenizer()

def remove_unnecessary_chars(string):
    '''
    Removes some characters, like accent diacritics
    '''
    source = unicodedata.normalize('NFD', string)
    result = ''.join(
        c for c in source if c not in {'´', '́'}
    )
    return unicodedata.normalize('NFC', result)

def normalize(tokens, analyzer=morph):
    '''
    Replaces tokens with its first normal form returned by pymorphy2
    '''

    def get_normal_form(token):
        forms = morph.parse(token)
        return forms[0].normal_form

    return (get_normal_form(t) for t in tokens)

def tokenize(text):
    tokens = tokenizer.tokenize(text)
    tokens = (token for token in tokens if len(token) >= 3)
    tokens = (
        remove_unnecessary_chars(token.lower()) for token in tokens
    )
    return list(
        normalize(tokens)
    )

def serialize_object(obj):
    obj['type'] = obj['type'].value
    return obj

languages: Markdown Text Python


text:
<file_sep># sm-ui

[![Gitter](https://badges.gitter.im/Some-Group/sm-ui.svg)](https://gitter.im/Some-Group/sm-ui?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)

branch_name: refs/heads/main

repo_language: Markdown

repo_name: Some-Group/sm-ui


text:
<file_sep># ADF
Hello world

branch_name: refs/heads/master


text:
<repo_name>simka-for/statementParser<file_sep>/src/main/java/Main.java
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.*;

public class Main {

    public static void main(String[] args) {

        List<Movement> movementsList = loadMovementList();

        System.out.println("\nОбщий приход денежных средств в период <<01.03.17 - 31.05.17>> : " + movementsList.stream()
                .mapToDouble(Movement::getComing).sum() + " RUB");

        System.out.println("Общий расход денежных средств в период <<01.03.17 - 31.05.17>> : " + movementsList.stream()
                .mapToDouble(Movement::getConsumption).sum() + " RUB");

        ArrayList<Movement> coming = new ArrayList<>();
        for (Movement com : movementsList){
            if (com.getComing() != 0){
                coming.add(com);
            }
        }

        ArrayList<Movement> consumption = new ArrayList<>();
        for (Movement con : movementsList){
            if (con.getConsumption() != 0){
                consumption.add(con);
            }
        }

        System.out.println("\nОтчет о приходе: \n" +
                "==================================");
        for (String key : getPayment(coming).keySet()) {
            System.out.println(key + " : " + getPayment(coming).get(key) + " RUB");
        }

        System.out.println("\nОтчет о расходе: \n" +
                "==================================");
        for (String key : getPayment(consumption).keySet()) {
            System.out.println(key + " : " + getPayment(consumption).get(key) + " RUB");
        }

    }

    private static ArrayList<Movement> loadMovementList() {
        ArrayList<Movement> list = new ArrayList<>();

        try {
            String movementList = "data/movementList.csv";
            List<String> lines = Files.readAllLines(Paths.get(movementList));
            for (int i = 1; i < lines.size(); i++) {
                String[] fragments = lines.get(i).
                        replaceAll("\"([^\"]*),([^\"]*)\"", "$1.$2").
                        split(",", 8);
                String operation = clearOperation(fragments[5]).replaceAll("\\s+", "/");
                list.add(new Movement(operation,
                        Double.parseDouble(fragments[6]),
                        Double.parseDouble(fragments[7])));
            }
            list.remove(0);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return list;
    }

    private static String clearOperation(String operation) {
        if (!"Описание операции".equals(operation)) {
            int last = operation.indexOf("       ");
            return operation.substring(17, last).trim();
        }
        return operation;
    }

    private static TreeMap<String, Double> getPayment(List<Movement> operations) {
        TreeMap<String, Double> payment = new TreeMap<>();
        for (Movement operation : operations) {
            String operate = operation.getOperation();
            double amount = 0;
            for (Movement o : operations) {
                if (operate.equals(o.getOperation())) {
                    amount += o.getConsumption();
                    amount += o.getComing();
                }
                payment.put(operate, amount);
            }
        }
        return payment;
    }

}
<file_sep>/src/main/java/Movement.java
public class Movement {

    private String operation;
    private double coming;
    private double consumption;

    public Movement(String operation, double coming, double consumption){
        this.operation = operation;
        this.coming = coming;
        this.consumption = consumption;
    }

    public String getOperation() {
        return operation;
    }

    public double getComing() {
        return coming;
    }

    public double getConsumption() {
        return consumption;
    }

    @Override
    public String toString() {
        return  operation + "\t" + coming + "\t" + consumption;
    }
}

repo_language: Java

repo_name: simka-for/statementParser


text:
<file_sep># Repositorio Test
Para clase de páginas web
<file_sep>import math

def seno(num):
	res= math.sin(num)
	return res

def coseno(num):
    res = math.cos(num)
    return res

num = int(input("Ingrese un numero: "))
base = int(input("Ingrese el valor de la base: "))

print("La base mide %s" %base)
print(coseno(num))

<file_sep>nombre = input("Igresa tu nombre: ")
edad = input("Ingresa tu edad: ")

print("Tu nombre es %s y edad %d ya estas bien pinche viejo carnal" %(nombre, edad))

languages: Markdown Python


text:
<repo_name>jpcasa29/jpcasa29<file_sep>/README.md
# <NAME>

## Presentación 📋

Soy desarrollador web Full Stack y experto en las áreas Hard de Recursos Humanos, actualmente trabajo en Cencosud Argentina como Gerente de Payroll.

Estudié Desarrollo Web en Digital House y avancé en herramientas de programación como Javascript, HTML, CSS, REACT, NodeJS.

Contactame por [LinkedIn](www.linkedin.com/in/juanpacasa2009) ⌨


text:
<repo_name>MathiasPoe/RPi-read-lab-status<file_sep>/Readme.md
# RPi-read-lab-status
Read status infos from lab equiptment and display them.
This program runs on a Raspberry Pi 3b
<file_sep>/config_LT.py
import collections
import pickle as pk
from Adafruit_MAX31856 import MAX31856 as MAX31856

MACHINE='LT'

HELIUM_CHECK = '/pressure-logs/measure-helium-LT'
HELIUM_LOG = '/pressure-logs/helium-LT-%Y.log'
PRESSURE_LOGS = '/pressure-logs/%Y/pressure-LT-%Y-%m-%d.log'

COM_PORT_MAXIGAUGE = '/dev/serial/by-id/usb-FTDI_FT232R_USB_UART_A505YB2G-if00-port0'
COM_PORT_MVC_GAUGE_PREP = None
COM_PORT_MVC_GAUGE_STM = None
COM_PORT_ION_PREP = None
COM_PORT_ION_CRYO = None
COM_PORT_ION_STM = None

HELIUM=True

FPS_SHOW=False

GRADIENT = 30  # calc gradient from last n seconds
GRADIENT_RUNEVERY = 5  # save past values every n seconds
GRADIENT_SHOW = 60  # show gradient per n seconds

# temperature chip
SPI0_DEV = 0
SPI0_CS0 = 0
SPI0_CS0_temp_type = MAX31856.MAX31856_K_TYPE
SPI0_CS1 = 1
SPI0_CS1_temp_type = MAX31856.MAX31856_K_TYPE

SPI1_DEV = None
SPI1_CS0 = None
SPI0_CS0_temp_type = MAX31856.MAX31856_K_TYPE
SPI1_CS1 = None
SPI0_CS1_temp_type = MAX31856.MAX31856_K_TYPE

data = collections.OrderedDict()
# unit: unit in which the values are measure (mbar, K, C, A)
# color: color which is used in the GUI
# sensor type: sensor with which values are recorded (maxigauges, ADC_diods, ADC_resistor, SPI0, SPI1, mvc_stm, mvc_prep, ser_ion_stm, ser_ion_cryo, ser_ion_prep)
# sensor: sensor number (channel)
# value: current value (initial value -4000 - Not found)
# limit_max: if value > limit_max -> # WARNING:
# limit_max_warning: warning, which appears if value > limit_max
# format: format of value displayed in GUI (check https://www.programiz.com/python-programming/methods/string/format )
# format_gradient: format of gradient displayed in GUI
# log_to_file: shuld value be logged? Yes -> True, No -> False
# gui_size: size of value in GUI (1 or 2)
# gui_order: set order for appearance in GUI
# used_sensor: sensor assigned to value (None at beginning)
data['PSTM'] = {'unit': 'mbar',
                     'color': '#837C00',
                     'sensor_type': 'maxigauges',
                     'sensor': 1,
                     'status': 5,
                     'value': -4000,
                     'limit_max': 1e-7,
                     'limit_max_warning': 'Cryo chamber pressure is high',
                     'format': '.2e',
                     'format_gradient': '.0e',
                     'log_to_file': True,
                     'gui_size': 2,
                     'gui_order': 0,
                     'used_sensor':None}
data['PROU'] = {'unit': 'mbar',
                     'color': '#606060',
                     'sensor_type': 'maxigauges',
                     'sensor': 2,
                     'status': 5,
                     'value': -4000,
                     'limit_max': 2e1,
                     'limit_max_warning': 'Rough pump pressure is high',
                     'format': '.2e',
                     'log_to_file': True,
                     'gui_size': 1,
                     'gui_order': 6,
                     'used_sensor':None}
data['PPRP'] = {'unit': 'mbar',
                     'color': '#E6DD23',
                     'sensor_type': 'maxigauges',
                     'sensor': 3,
                     'status': 5,
                     'value': -4000,
                     'limit_max': 1e-4,
                     'limit_max_warning': 'Prep chamber pressure is high',
                     'format': '.2e',
                     'format_gradient': '.0e',
                     'log_to_file': True,
                     'gui_size': 2,
                     'gui_order': 1,
                     'used_sensor':None}
data['TSTM'] = {'unit': 'K',
                     'color': '#AC0D2F',
                     'sensor_type': 'ADC_diods',
                     'sensor': 0,
                     'status': 5,
                     'limit_max': 40,
                     'limit_max_warning': 'STM temperature is high',
                     'value': -4000,
                     'format': '.3f',
                     'format_gradient': '.2f',
                     'log_to_file': True,
                     'gui_size': 2,
                     'gui_order': 2,
                     'used_sensor':None}
data['TCRY'] = {'unit': 'K',
                     'color': '#606060',
                     'sensor_type': 'ADC_diods',
                     'sensor': 2,
                     'status': 5,
                     'value': -4000,
                     'limit_max': 10,
                     'limit_max_warning': 'Cryo temperature is high',
                     'format': '.3f',
                     'format_gradient': '.2f',
                     'log_to_file': True,
                     'gui_size': 1,
                     'gui_order': 8,
                     'used_sensor':None}
data['TSAM'] = {'unit': 'C',
                     'color': '#ABDA21',
                     'sensor_type': 'SPI0',
                     'sensor': 'CS0',
                     'status': 5,
                     'value': -4000,
                     'format': '.2f',
                     'format_gradient': '.1f',
                     'log_to_file': True,
                     'gui_size': 2,
                     'gui_order': 4,
                     'used_sensor':None}
data['TMAN'] = {'unit': 'C',
                     'color': '#606060',
                     'sensor_type': 'ADC_resistor',
                     'sensor': 1,
                     'status': 5,
                     'value': -4000,
                     'limit_max': 80,
                     'limit_max_warning': 'Manipulator temperature is high',
                     'format': '.2f',
                     'format_gradient': '.1f',
                     'log_to_file': True,
                     'gui_size': 2,
                     'gui_order': 5,
                     'used_sensor':None}
data['TLAB'] = {'unit': 'C',
                     'color': '#606060',
                     'sensor_type': 'SPI0',
                     'sensor': 'CS0',
                     'status': 5,
                     'value': -4000,
                     'format': '.2f',
                     'log_to_file': True,
                     'gui_size': 1,
                     'gui_order': 7,
                     'used_sensor':None}

# variuos
kel_cel = -273.15
date_fmt = '%Y-%m-%d_%H:%M:%S'
date_fmt_day = '%Y-%m-%d'
date_fmt_display = '%b %d, %H:%M:%S'
date_fmt_display_he = '%b %d, %H:%M'
date_fmt_display_he_short = '%H:%M'

# wireless outlet
A_ON = '010101100001000001001010111111111'
A_OFF = '010110100001000001001010111111111'
delay_shorten = 0.00005  # delays wil be made shorter to account for command execution time
short_delay = 0.00058 - delay_shorten
long_delay = 0.00116 - delay_shorten
extended_delay = 0.00716
TRANSMIT_PIN = 26  # adc

# temperature chip
SPI_DEV=0
SPI_PORT=0

#adc
ADC_ADDR_DIODS = 0x48
ADC_ADDR_VARIOUS = 0x49

# decoding
conv_to_decode = {0: 0, 1: 1e-12, 2: -1000, 3: -2000, 4: -3000, 5: -4000, 6: -5000}
decoding_dict = {1e-12: 'Underrange', -1000: 'Overrange', -2000: 'Error', -3000: 'Off', -4000: 'Not found', -5000: 'ID error'}

# temperature calibrations
# voltage to temperature calibration for diode measuerement
temp_calib_diode = pk.load(open("tempdiode.pickle", "rb"))
# voltage to temperature calibration for type-k thermocouple
temp_calib_type_K = pk.load(open("temptypek.pickle", "rb"))
# resistance to temperature calibration for pt100 sensor
temp_calib_resistor = pk.load(open("tempresistor.pickle", "rb"))

# GUI
FONT_FAMILY = 'Liberation Mono'
COLOR_BACKGROUND_WINDOW = '#030919'
COLOR_BACKGROUND = '#030919'
COLOR_TIME = "#4A63A1"
COLOR_HELIUM = "#606060"
COLOR_status_gui_inactive = '#303030'
COLOR_status_gui_active = '#a0a0a0'
COLOR_status_gui_warning = '#f06060'
COLOR_button_helium_bg = '#303030'
COLOR_button_helium_fg = '#f06020'

COLOR_gradient_brightness_factor = 0.4
COLOR_gradient_unit_brightness_factor = 0.3
FONT_SCALING = 2.0

height_ratio_time = 4
height_ratio_normal = 6
height_ratio_small = 3
height_ratio_helium = 4

font_ratio_time = 0.75
font_ratio_small = 0.5
font_ratio_gradient = 0.40
font_ratio_gradient_unit = 0.35
font_ratio_helium = 0.75
font_size_gui = 15

measure_animations = [
    [u"\u25DC", u"\u25DD", u"\u25DE", u"\u25DF"],
    ["⠋","⠙","⠹","⠸","⠼","⠴","⠦","⠧","⠇","⠏"],
    ["⣾","⣽","⣻","⢿","⡿","⣟","⣯","⣷"],
    ["⠋","⠙","⠚","⠞","⠖","⠦","⠴","⠲","⠳","⠓"],
    ["⠄","⠆","⠇","⠋","⠙","⠸","⠰","⠠","⠰","⠸","⠙","⠋","⠇","⠆"],
    ["⠋","⠙","⠚","⠒","⠂","⠂","⠒","⠲","⠴","⠦","⠖","⠒","⠐","⠐","⠒","⠓","⠋"],
    ["⠁","⠉","⠙","⠚","⠒","⠂","⠂","⠒","⠲","⠴","⠤","⠄","⠄","⠤","⠴","⠲","⠒","⠂","⠂","⠒","⠚","⠙","⠉","⠁"],
    ["⠁","⠁","⠉","⠙","⠚","⠒","⠂","⠂","⠒","⠲","⠴","⠤","⠄","⠄","⠤","⠠","⠠","⠤","⠦","⠖","⠒","⠐","⠐","⠒","⠓","⠋","⠉","⠈","⠈"],
    ["⢹","⢺","⢼","⣸","⣇","⡧","⡗","⡏"],
    ["⢄","⢂","⢁","⡁","⡈","⡐","⡠"],
    ["⠁","⠂","⠄","⡀","⢀","⠠","⠐","⠈"],
    ["_","_","_","-","`","`","'","´","-","_","_","_"],
    ["☱","☲","☴"],
    #["🙈","🙈","🙉","🙊"],
    #["😄","😝"],
    #["🕛","🕐","🕑","🕒","🕓","🕔","🕕","🕖","🕗","🕘","🕙","🕚"],
    #["🌍","🌎","🌏"],
	#["🌑","🌒","🌓","🌔","🌕","🌖","🌗","🌘"],
    #["🚶","🏃"],
    ["◐","◓","◑","◒"],
    ["▖","▘","▝","▗"],
]
measure_animations_speed = [
    80,
    80,
    80,
    80,
    80,
    80,
    80,
    80,
    80,
    80,
    100,
    70,
    100,
    #300,
    #200,
    #100,
    #180,
    #80,
    #140,
    50,
    120,
]
<file_sep>/GUI.py
# tkinter based GUI for raspberry pi status
#
# info on tkinter as thread:
# https://stackoverflow.com/questions/459083/how-do-you-run-your-own-code-alongside-tkinters-event-loop
#

import colorsys
import datetime as dt
import numpy as np
import os
import time
import tkinter as tk
import threading
from google_speech import Speech

import config as CFG

def color_brightness(color, amount=0.5):
    crgb = tuple(int(color[i+1:i+3], 16) for i in (0, 2 ,4))
    c = colorsys.rgb_to_hls(*crgb)
    crgb = colorsys.hls_to_rgb(c[0], 1 - amount * (1 - c[1]), c[2])
    crgb = [int(c) for c in crgb]
    return "#{0:02x}{1:02x}{2:02x}".format(*crgb)

class MainWindow(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.start()

        self.root = tk.Tk()
        self.root.title("{} Status".format(CFG.MACHINE))
        self.root.tk.call('wm', 'iconphoto', self.root._w, tk.PhotoImage(file='icons8-dove-96.png'))
        self.root.geometry('800x600')
        self.root.configure(background=CFG.COLOR_BACKGROUND_WINDOW)
        # self.root.configure(padx=20)
        # self.root.configure(pady=20)
        # self.root.state('zoomed')
        # self.root.wm_state('zoomed')
        self.root.lift()

        self.font = (CFG.FONT_FAMILY, 50, 'bold')
        self.font_time = (CFG.FONT_FAMILY, 30)
        self.font_helium = (CFG.FONT_FAMILY, 24)
        self.font_gui = (CFG.FONT_FAMILY, CFG.font_size_gui)

        self.topmost = False
        self.geometry = '800x600'

        self.root.bind("<Configure>", self.resize)
        self.root.bind("<F12>", self.toggle_fullscreen)
        self.root.bind("<F11>", self.toggle_zoomed)
        self.root.bind("<F10>", self.toggle_topmost)
        self.root.protocol("WM_DELETE_WINDOW", self.endApp)

        self.labels_names = {}
        self.labels_values = {}
        self.label_time = None
        self.font_scaling_factor = 10

        self.warnings = {}
        self.warning_notification_output = False

        self.measure_helium_symbol = u"\u21bb"
        self.measure_animation_current = 0
        self.measure_animation = []
        self.measure_animation_speed = 100
        self.measure_helium_animation_timer_id = 0

        # w, h = root.winfo_screenwidth(), root.winfo_screenheight()
        # root.geometry("%dx%d+0+0" % (w, h))

    def startApp(self):
        # start in fullscreen mode
        if not self.is_fullscreen():
            self.toggle_fullscreen(None)

        self.root.mainloop()

    def endApp(self):
        global APP_RUNNING
        APP_RUNNING = False
        self.root.quit()

    def init_labels(self, labels, colors, sizes):
        """initialiyes layout with labels"""

        self.menu_top = tk.Frame(self.root, height=12, borderwidth=0, highlightthickness=0, bg=CFG.COLOR_BACKGROUND_WINDOW)
        self.main_area = tk.Frame(self.root, borderwidth=0, highlightthickness=0, bg=CFG.COLOR_BACKGROUND_WINDOW)
        self.menu_top.pack(fill=tk.X)
        self.main_area.pack(expand=True, fill=tk.BOTH)

        self.label_gui_ontop = tk.Label(
            self.menu_top, text=u"\u25C9", font=self.font_gui, fg=CFG.COLOR_status_gui_inactive, bg=CFG.COLOR_BACKGROUND,
            borderwidth=0, highlightthickness=0
        )
        self.label_gui_ontop.pack(side=tk.RIGHT)
        self.label_gui_ontop.bind("<Button-1>", self.toggle_topmost)

        self.label_gui_spacers = {}
        self.label_gui_warnings = {}
        for l in labels:
            self.label_gui_spacers[l] = tk.Label(
                self.menu_top, text=u" ", font=self.font_gui, fg=CFG.COLOR_status_gui_inactive, bg=CFG.COLOR_BACKGROUND,
                borderwidth=0, highlightthickness=0
            )
            self.label_gui_warnings[l] = tk.Label(
                self.menu_top, text="", font=self.font_gui, fg=CFG.COLOR_status_gui_inactive, bg=CFG.COLOR_BACKGROUND,
                borderwidth=0, highlightthickness=0
            )
            self.label_gui_warnings[l].bind("<Button-1>", self.warning_remove)

        row_num = 0
        row_height_relative = []
        self.label_time = tk.Label(self.main_area, text="", font=self.font_time, fg=CFG.COLOR_TIME, bg=CFG.COLOR_BACKGROUND)
        self.label_time.grid(column=0, row=row_num, columnspan=4, padx=0, pady=0, sticky=tk.N + tk.E + tk.W)
        tk.Grid.rowconfigure(self.main_area, row_num, weight=CFG.height_ratio_time)
        row_height_relative.append(CFG.height_ratio_time)

        self.labels_names = {}
        self.labels_values_container = {}
        self.labels_values = {}
        self.labels_values_gradient = {}
        self.labels_values_gradient_unit = {}
        self.labels_small = {}
        total_size = 0
        # TODO: so far only sizes 1 and 2 are supported
        for l, val in labels.items():
            extra_column = total_size % 2
            if sizes[l] >=2:
                if extra_column:
                    total_size += total_size % 2
                    extra_column = 0
            else:
                self.labels_small[l] = True
            if extra_column == 0:
                row_num += 1
            self.labels_values_container[l] = tk.Frame(self.main_area, borderwidth=0, highlightthickness=0, bg=CFG.COLOR_BACKGROUND_WINDOW)
            self.labels_names[l] = tk.Label(self.main_area, text=val, font=self.font, fg=colors[l], bg=CFG.COLOR_BACKGROUND)
            self.labels_values[l] = tk.Label(self.labels_values_container[l], text="", font=self.font, fg=colors[l], bg=CFG.COLOR_BACKGROUND)
            if sizes[l] >= 2:  # no gradients for the small labels
                self.labels_values_gradient[l] = tk.Label(self.labels_values_container[l], text="      ", font=self.font, fg=color_brightness(colors[l], CFG.COLOR_gradient_brightness_factor), bg=CFG.COLOR_BACKGROUND)
                self.labels_values_gradient_unit[l] = tk.Label(self.labels_values_container[l], text="/{}s".format(CFG.GRADIENT_SHOW), font=self.font, fg=color_brightness(colors[l], CFG.COLOR_gradient_unit_brightness_factor), bg=CFG.COLOR_BACKGROUND)
            self.labels_values[l].grid(column=0, row=0, rowspan=2, padx=0, pady=0, sticky=tk.N + tk.S + tk.E)
            if sizes[l] >= 2:
                self.labels_values_gradient[l].grid(column=1, row=0, padx=0, pady=0, sticky=tk.N + tk.S + tk.W)
                self.labels_values_gradient_unit[l].grid(column=1, row=1, padx=0, pady=0, sticky=tk.N + tk.E + tk.W)
                self.labels_values_container[l].columnconfigure(1, weight=1)
            self.labels_values_container[l].rowconfigure(0, weight = 1)
            self.labels_values_container[l].rowconfigure(1, weight = 1)
            self.labels_values_container[l].columnconfigure(0, weight=1)
            if sizes[l] >= 2:
                c_label = 0
                c_value = 2
            else:
                c_label = extra_column * 2
                c_value = extra_column * 2 + 1
            self.labels_names[l].grid(column=c_label, row=row_num, columnspan=sizes[l], padx=0, pady=0, sticky=tk.N + tk.S + tk.E)
            self.labels_values_container[l].grid(column=c_value, row=row_num, columnspan=sizes[l], padx=2, pady=0, sticky=tk.N + tk.S + tk.W)
            total_size += sizes[l]
            self.main_area.rowconfigure(row_num, weight=CFG.height_ratio_normal)
            row_height_relative.append(CFG.height_ratio_normal)

        row_num += 1

        if CFG.HELIUM != None:
            self.frame_helium = tk.Frame(self.main_area, borderwidth=0, highlightthickness=0, bg=CFG.COLOR_BACKGROUND_WINDOW)
            self.frame_helium.grid(column=0, row=row_num, columnspan=4, padx=0, pady=0, sticky=tk.N + tk.S + tk.E + tk.W)
            self.button_measure_helium = tk.Button(
                self.frame_helium, text=self.measure_helium_symbol, command=self.measure_helium,
                font=self.font_helium, fg=CFG.COLOR_button_helium_fg, bg=CFG.COLOR_button_helium_bg,
                borderwidth=0, highlightthickness=0
            )
            self.button_measure_helium.pack(side=tk.LEFT)

            self.label_helium = tk.Label(self.frame_helium, text="", font=self.font_helium, fg=CFG.COLOR_HELIUM, bg=CFG.COLOR_BACKGROUND)
            self.label_helium.pack()

            tk.Grid.rowconfigure(self.main_area, row_num, weight=CFG.height_ratio_helium)
            row_height_relative.append(CFG.height_ratio_helium)

        row_height_relative = np.array(row_height_relative)
        self.font_scaling_factor = np.sum(row_height_relative / np.max(row_height_relative))

        for i in range(4):
            tk.Grid.columnconfigure(self.main_area, i, weight=1)

    def update_values(self, values, str_time):
        '''updates values'''
        for k, v in values.items():
            self.labels_values[k]['text'] = v
        self.label_time['text'] = str_time

    def update_values_gradient(self, values):
        '''updates values'''
        for k, v in values.items():
            if k in self.labels_values_gradient:
                self.labels_values_gradient[k]['text'] = v

    def update_helium(self, str_helium):
        self.label_helium['text'] = str_helium

        if self.measure_helium_animation_timer_id:
            self.root.after_cancel(self.measure_helium_animation_timer_id)
        self.measure_animation_current = 0
        self.button_measure_helium['text'] = self.measure_helium_symbol
        self.button_measure_helium.config(state="normal")

    def measure_helium(self):
        with open(os.getcwd() + CFG.HELIUM_CHECK, 'a'):
            os.utime(os.getcwd() + CFG.HELIUM_CHECK)

        self.button_measure_helium.config(state="disabled")
        num = np.random.randint(len(CFG.measure_animations))
        self.measure_animation = CFG.measure_animations[num]
        self.measure_animation_speed = CFG.measure_animations_speed[num]
        self.measure_helium_animation()

    def measure_helium_animation(self):
        self.button_measure_helium['text'] = self.measure_animation[self.measure_animation_current].encode('utf-8').decode()
        self.measure_animation_current += 1
        if self.measure_animation_current >= len(self.measure_animation):
            self.measure_animation_current = 0
        self.measure_helium_animation_timer_id = self.root.after(self.measure_animation_speed, self.measure_helium_animation)

    def is_fullscreen(self):
        return self.root.attributes('-fullscreen')

    def toggle_fullscreen(self, event):
        if self.is_fullscreen():
            self.root.wm_attributes('-fullscreen', False)
            self._set_topmost(self.topmost)
        else:
            self.root.wm_attributes('-fullscreen', True)
            self._set_topmost(False)

    def is_zoomed(self):
        if self.root.state() == 'zoomed':
            return True
        if self.root.wm_state() == 'zoomed':
            return True
        zoomed = False
        try:
            geometry_size = [int(x) for x in self.geometry.split('+')[0].split('x')]
            actual_geometry_size = [int(x) for x in self.root.geometry().split('+')[0].split('x')]
            if actual_geometry_size[0] > geometry_size[0] and actual_geometry_size[1] > geometry_size[1]:
                zoomed = True
        except ValueError:
            pass
        return zoomed

    def toggle_zoomed(self, event):
        if not self.is_zoomed():
            self.geometry = self.root.geometry()
            try:
                self.root.wm_state('zoomed')
            except Exception:
                pass
            try:
                self.root.wm_attributes('-zoomed', True)
            except Exception:
                pass
            self._set_topmost(False)
        else:
            self.root.wm_state('normal')
            try:
                self.root.wm_attributes('-zoomed', False)
            except Exception:
                pass
            self._set_topmost(self.topmost)
            self.root.geometry(self.geometry)

    def toggle_topmost(self, event):
        if self.root.wm_attributes("-topmost"):
            self._set_topmost(False)
            self.topmost = False
        else:
            if not self.is_zoomed() and not self.is_fullscreen():
                self._set_topmost(True)
            self.topmost = True

    def _set_topmost(self, active=True):
        self.root.wm_attributes("-topmost", active)
        if active:
            self.label_gui_ontop.config(fg=CFG.COLOR_status_gui_active)
        else:
            self.label_gui_ontop.config(fg=CFG.COLOR_status_gui_inactive)

    def resize(self, event):
        if (event.widget == self.root):
            size = min(event.height / 3, event.width / 4)
            s = int(size / self.font_scaling_factor * CFG.FONT_SCALING)
            ssmall = int(s * CFG.font_ratio_small)
            stime = int(s * CFG.font_ratio_time)
            sgradient = int(s * CFG.font_ratio_gradient)
            sgradientsmall = int(ssmall * CFG.font_ratio_gradient)
            sgradientunit = int(s * CFG.font_ratio_gradient_unit)
            sgradientunitsmall = int(ssmall * CFG.font_ratio_gradient_unit)
            if CFG.HELIUM != None:
                shelium = int(s * CFG.font_ratio_helium)
            self.font = (CFG.FONT_FAMILY, s, 'bold')
            self.font_small = (CFG.FONT_FAMILY, ssmall, 'bold')
            self.font_time = (CFG.FONT_FAMILY, stime)
            self.font_gradient = (CFG.FONT_FAMILY, sgradient)
            self.font_gradientsmall = (CFG.FONT_FAMILY, sgradientsmall)
            self.font_gradientunit = (CFG.FONT_FAMILY, sgradientunit)
            self.font_gradientunitsmall = (CFG.FONT_FAMILY, sgradientunitsmall)
            if CFG.HELIUM != None:
                self.font_helium = (CFG.FONT_FAMILY, shelium)
            for k in self.labels_values:
                if k in self.labels_small:
                    self.labels_values[k].config(font=self.font_small)
                    # self.labels_values_gradient[k].config(font=self.font_gradientsmall)
                    # self.labels_values_gradient_unit[k].config(font=self.font_gradientunitsmall)
                    self.labels_names[k].config(font=self.font_small)
                else:
                    self.labels_values[k].config(font=self.font)
                    self.labels_values_gradient[k].config(font=self.font_gradient)
                    self.labels_values_gradient_unit[k].config(font=self.font_gradientunit)
                    self.labels_names[k].config(font=self.font)
            if self.label_time is not None:
                self.label_time.config(font=self.font_time)
            if CFG.HELIUM != None:
                if self.label_helium is not None:
                    self.label_helium.config(font=self.font_helium)
            # print("New size is: {}x{}".format(event.width, event.height))

    def warning_notification(self, text):
        print(dt.datetime.now().strftime('%b %d, %H:%M:%S'), text)
        while self.warning_notification_output:
            time.sleep(0.5)
        self.warning_notification_output = True
        speech = Speech(text, 'en')
        speech.play()
        time.sleep(3)
        self.warning_notification_output = False

    def warning(self, key, text, text_short):
        if key not in self.warnings:
            self.warnings[key] = {}
            self.warnings[key]['last_sound'] = dt.datetime.now()-dt.timedelta(hours=24*365*10)
            self.warnings[key]['num_warnings'] = 0
            self.warning_notification(text)
        if (dt.datetime.now() - self.warnings[key]['last_sound']).seconds > 3 + 2 ** self.warnings[key]['num_warnings']:
            self.warning_notification(text)
            self.warnings[key]['num_warnings'] += 1
            self.warnings[key]['last_sound'] = dt.datetime.now()
        self.label_gui_warnings[key]['text']=text_short
        self.label_gui_warnings[key].pack(side=tk.LEFT)
        self.label_gui_spacers[key].pack(side=tk.LEFT)
        self.label_gui_warnings[key].config(fg=CFG.COLOR_status_gui_warning)

    def dewarning(self, key):
        if key in self.warnings:
            self.label_gui_warnings[key].config(fg=CFG.COLOR_status_gui_inactive)
            self.warnings[key]['num_warnings'] = 0

    def warning_remove(self, event):
        for key, label in self.label_gui_warnings.items():
            if event.widget == label:
                label.pack_forget()
                self.label_gui_spacers[key].pack_forget()
                self.warnings[key]['num_warnings'] = 0
                return

def initGUI():
    gui = MainWindow()
    return gui

if __name__ == "__main__":
    # test code
    import datetime as dt
    gui = initGUI()
    labels = {
        "sfdfd": "sfdfdf", "sfdfd2": "sfdfdf2", "sfdfd3": "sfdfdf3", "sfdfd4":
        "sfdfdf4", "sfdfd5": "sfdfdf5", "sfdfd6": "sfdfdf6", "sfdfd7": "sfdfdf7", "sfdfd8": "sfdfdf8",
        "sfdfd9": "sfdfdf9", "sfdfd10": "sfdfdf10",
    }
    labels_gradients = {
        "sfdfd": "20", "sfdfd2": "30", "sfdfd3": "50", "sfdfd4":
        "60", "sfdfd5": "70", "sfdfd6": "80", "sfdfd7": "90", "sfdfd8": "100",
        "sfdfd9": "110", "sfdfd10": "100",
    }
    colors = {key: '#c0c0c0' for key in labels}
    sizes = {key: 2 for key in labels}
    sizes['sfdfd'] = 1
    sizes['sfdfd2'] = 1
    sizes['sfdfd9'] = 1
    sizes['sfdfd10'] = 1
    gui.init_labels(labels, colors, sizes)
    gui.update_values(labels, dt.datetime.now().strftime("%b %d, %H:%M:%S"))
    gui.update_values_gradients(labels_gradients)
    if CFG.HELIUM != None:
        gui.update_helium('LHe 007 mm (Aug 1st, 19:17)')

    gui.root.mainloop()
<file_sep>/status-read.py
# Status display for Raspberry Pi
#
# required software: python3.5 or newer, https://pypi.org/project/adafruit-circuitpython-ads1x15/0.5.3/
#
# enalbe GPIO and SPI on your RPi

import collections
import copy
import datetime as dt
from functools import wraps
import numpy as np
import pandas as pd
import pickle as pk
import os
import threading
import time
import serial
import board
import busio
import Adafruit_GPIO
from Adafruit_MAX31856 import MAX31856 as MAX31856
import adafruit_ads1x15.single_ended

import config as CFG    # config file - individual for every machine
import GUI              # GUI for visualization and interaction on screen

class measure:
    # decoding
    conv_to_decode = CFG.conv_to_decode
    decoding_dict = CFG.decoding_dict       # converts values to labels like 'Overrange', 'Off', ...

    # temperature calibrations
    # voltage to temperature calibration for diode measuerement
    temp_calib_diode = CFG.temp_calib_diode
    # voltage to temperature calibration for type-k thermocouple
    temp_calib_type_K = CFG.temp_calib_type_K
    # resistance to temperature calibration for pt100 sensor
    temp_calib_resistor = CFG.temp_calib_resistor

    def __init__(self):
        self.day_now = dt.datetime.now().strftime(CFG.date_fmt_day)     # get date of current day

        self.init_data_dict()       # get dictionary of values, which should be measured

        # for displaying gradients
        self.gradient_data_current = 0
        self.gradient_data_num = int(np.max(CFG.GRADIENT / CFG.GRADIENT_RUNEVERY))
        self.gradient_data = [copy.deepcopy(self.data) for _ in range(self.gradient_data_num)]  # we dont want references
        now = dt.datetime.now()
        self.gradient_data_timestamp = [now] * self.gradient_data_num

        self.sensor_types=list(set([self.data[key]['sensor_type'] for key in self.data]))   # get sensor types

        # initialize sensors
        for key in self.data:
            if self.data[key]['sensor_type'] in ['ADC_diods','ADC_resistor']:
                self.init_adc(key)
            if self.data[key]['sensor_type'] == 'maxigauges':
                self.init_serial_maxigauge(key)
            if self.data[key]['sensor_type'] == 'mvc_prep':
                self.init_serial_mvc_prep(key)
            if self.data[key]['sensor_type'] == 'mvc_stm':
                self.init_serial_mvc_stm(key)
            if self.data[key]['sensor_type'] == 'ser_ion_prep':
                self.init_serial_ion_prep(key)
            if self.data[key]['sensor_type'] == 'ser_ion_cryo':
                self.init_serial_ion_cryo(key)
            if self.data[key]['sensor_type'] == 'ser_ion_stm':
                self.init_serial_ion_stm(key)
            if self.data[key]['sensor_type'] == 'SPI0':
                if self.data[key]['sensor'] == 'CS0':
                    self.init_SPI0_CS0(key)
                if self.data[key]['sensor'] == 'CS1':
                    self.init_SPI0_CS1(key)
            if self.data[key]['sensor_type'] == 'SPI1':
                if self.data[key]['sensor'] == 'CS0':
                    self.init_SPI1_CS0(key)
                if self.data[key]['sensor'] == 'CS1':
                    self.init_SPI1_CS1(key)

        self.first_run = True

        # check if helium measurement is enabled
        if CFG.HELIUM != None:
            self.helium_status = {}
            self.helium_check = False
            self.helium_save = False
            self.helium_turn_sensor_off = False
            self.helium_turn_sensor_off_retries = 0

        # threads
        self.threads_running = {}
        self.lock = threading.Lock()

        self.log_writing_header = False

        # loop properties
        self.main_loop_time = 0.08
        self.main_loop_time_normal = 0.08
        self.main_loop_time_slow = 0.5

        self.time_loop = time.time()
        self.fps = ''

        self.gui = None

    def init_data_dict(self):
        # initialize data dictionary with for values which should be measured
        self.data=CFG.data

    def init_adc(self,key):
        # initialize analog-to-digital converter chip (adafruit_ads1x15)
        try:
            i2c = busio.I2C(board.SCL, board.SDA)
            self.adc = adafruit_ads1x15.single_ended.ADS1115(i2c, address=CFG.ADC_ADDR_VARIOUS)
            if self.data[key]['sensor_type'] not in ['ADC_diods','ADC_resistor']:
                self.data[key]['used_sensor']=self.adc
        except:
            self.adc = None
        try:
            self.adc2 = adafruit_ads1x15.single_ended.ADS1115(i2c, address=CFG.ADC_ADDR_DIODS)
            if self.data[key]['sensor_type'] in ['ADC_diods','ADC_resistor']:
                self.data[key]['used_sensor']=self.adc2
        except:
            self.adc2 = None

    def init_serial_maxigauge(self,key):
        # initialize maxigauges (pfeiffer)
        self.ser_maxi = None
        try:
            self.ser_maxi = serial.Serial(timeout=0.5,
                                     baudrate=9600,
                                     stopbits=serial.STOPBITS_ONE,
                                     bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_NONE
                                     )
            self.ser_maxi.port = CFG.COM_PORT_MAXIGAUGE
            self.ser_maxi.open()
            self.ser_maxi.reset_input_buffer()
            self.ser_maxi.reset_output_buffer()
            self.data[key]['used_sensor']=self.ser_maxi
        except:
            pass

    def init_serial_mvc_prep(self,key):
        # initialize pressure gauge (vacom) in prep chamber
        self.ser_mvc_prep = None
        try:
            self.ser_mvc_prep = serial.Serial(timeout=0.5,
                                     baudrate=19200,
                                     stopbits=serial.STOPBITS_ONE,
                                     bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_NONE
                                     )
            self.ser_mvc_prep.port = CFG.COM_PORT_MVC_GAUGE_PREP
            self.ser_mvc_prep.open()
            self.ser_mvc_prep.reset_input_buffer()
            self.ser_mvc_prep.reset_output_buffer()
            self.data[key]['used_sensor']=self.ser_mvc_prep
        except:
            pass

    def init_serial_mvc_stm(self,key):
        # initialize mvc pressure gauge (vacom) in stm/afm chamber
        self.ser_mvc_stm = None
        try:
            self.ser_mvc_stm = serial.Serial(timeout=0.5,
                                     baudrate=19200,
                                     stopbits=serial.STOPBITS_ONE,
                                     bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_NONE
                                     )
            self.ser_mvc_stm.port = CFG.COM_PORT_MVC_GAUGE_STM
            self.ser_mvc_stm.open()
            self.ser_mvc_stm.reset_input_buffer()
            self.ser_mvc_stm.reset_output_buffer()
            self.data[key]['used_sensor']=self.ser_mvc_stm
        except:
            pass

    def init_serial_ion_prep(self,key):
        # initialize ion pump (gamma vacuum) in prep chamber (AFM/XPS)
        self.ser_ion_prep = None
        try:
            self.ser_ion_prep = serial.Serial(timeout=0.5,
                                     baudrate=9600,
                                     stopbits=serial.STOPBITS_ONE,
                                     bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_NONE
                                     )
            self.ser_ion_prep.port = CFG.COM_PORT_ION_PREP
            self.ser_ion_prep.open()
            self.ser_ion_prep.reset_input_buffer()
            self.ser_ion_prep.reset_output_buffer()
            self.data[key]['used_sensor']=self.ser_ion_prep
        except:
            pass

    def init_serial_ion_cryo(self,key):
        # initialize ion pump (gamma vacuum) in cryo chamber (AFM)
        self.ser_ion_cryo = None
        try:
            self.ser_ion_cryo = serial.Serial(timeout=0.5,
                                     baudrate=9600,
                                     stopbits=serial.STOPBITS_ONE,
                                     bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_NONE
                                     )
            self.ser_ion_cryo.port = CFG.COM_PORT_ION_CRYO
            self.ser_ion_cryo.open()
            self.ser_ion_cryo.reset_input_buffer()
            self.ser_ion_cryo.reset_output_buffer()
            self.data[key]['used_sensor']=self.ser_ion_cryo
        except:
            pass

    def init_serial_ion_stm(self,key):
        # initialize ion pump (gamma vacuum) in stm chamber (XPS)
        self.ser_ion_stm = None
        try:
            self.ser_ion_stm = serial.Serial(timeout=0.5,
                                     baudrate=9600,
                                     stopbits=serial.STOPBITS_ONE,
                                     bytesize=serial.EIGHTBITS,
                                     parity=serial.PARITY_NONE
                                     )
            self.ser_ion_stm.port = CFG.COM_PORT_ION_STM
            self.ser_ion_stm.open()
            self.ser_ion_stm.reset_input_buffer()
            self.ser_ion_stm.reset_output_buffer()
            self.data[key]['used_sensor']=self.ser_ion_stm
        except:
            pass

    def init_SPI0_CS0(self,key):
        # initialize temperature measuremeant via MAX31856 chip
        self.sensor_temp_sample = None
        try:
            self.sensor_temp_sample = MAX31856(hardware_spi=Adafruit_GPIO.SPI.SpiDev(
                CFG.SPI0_DEV, CFG.SPI0_CS0), tc_type=CFG.SPI0_CS0_temp_type, avgsel=0xF)  # 0x8 for 8 samples average, 0xF for 16
            self.data[key]['used_sensor']=self.sensor_temp_sample
        except:
            pass

    def init_SPI0_CS1(self,key):
        # initialize temperature measuremeant via MAX31856 chip
        self.sensor_temp_OMBE1 = None
        try:
            self.sensor_temp_OMBE1 = MAX31856(hardware_spi=Adafruit_GPIO.SPI.SpiDev(
                CFG.SPI0_DEV, CFG.SPI0_CS1), tc_type=CFG.SPI0_CS1_temp_type, avgsel=0xF)  # 0x8 for 8 samples average, 0xF for 16
            self.data[key]['used_sensor']=self.sensor_temp_OMBE1
        except:
            pass

    def init_SPI1_CS0(self,key):
        # initialize temperature measuremeant via MAX31856 chip
        self.sensor_temp_OMBE2 = None
        try:
            self.sensor_temp_OMBE2 = MAX31856(hardware_spi=Adafruit_GPIO.SPI.SpiDev(
                CFG.SPI1_DEV, CFG.SPI1_CS0), tc_type=CFG.SPI1_CS0_temp_type, avgsel=0xF)  # 0x8 for 8 samples average, 0xF for 16
            self.data[key]['used_sensor']=self.sensor_temp_OMBE2
        except:
            pass

    def init_SPI1_CS1(self,key):
        # initialize temperature measuremeant via MAX31856 chip
        self.sensor_temp_OMBE3 = None
        try:
            self.sensor_temp_OMBE3 = MAX31856(hardware_spi=Adafruit_GPIO.SPI.SpiDev(
                CFG.SPI1_DEV, CFG.SPI1_CS1), tc_type=CFG.SPI1_CS1_temp_type, avgsel=0xF)  # 0x8 for 8 samples average, 0xF for 16
            self.data[key]['used_sensor']=self.sensor_temp_OMBE3
        except:
            pass

    def _start_async(interval, check_lastrun=False):
        # decorator to start function as thread
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                self = args[0]
                if func.__name__ not in self.threads_running:
                    self.threads_running[func.__name__] = {}
                    self.threads_running[func.__name__]['lastrun'] = dt.datetime.now() - dt.timedelta(days=365 * 10)
                run_thread = False
                if 'thread' not in self.threads_running[func.__name__]:
                    run_thread = True
                elif not self.threads_running[func.__name__]['thread'].is_alive():
                    if check_lastrun:
                        if dt.datetime.now() - self.threads_running[func.__name__]['lastrun'] > dt.timedelta(seconds=interval / 2):
                            run_thread = True
                    else:
                        run_thread = True

                if run_thread:
                    # we always need to create a new thread-object because threads cant be re-run
                    self.threads_running[func.__name__]['lastrun'] = dt.datetime.now()
                    self.threads_running[func.__name__]['thread'] = threading.Timer(interval, func, args=args, kwargs=kwargs)
                    self.threads_running[func.__name__]['thread'].start()
            return wrapper
        return decorator

    def cancel_all_threads(self):
        for name, tdict in self.threads_running:
            if 'thread' in tdict:
                if tdict['thread'].is_alive():
                    tdict['thread'].cancel()

    def read_maxigauge(self, key):
        # controller returns something like 'x,x.xxxEsx <CR><LF>'
        # first digit is the error code, then comma, then pressure followed by Carrige return <CR>, Line feed <LF>
        # x,x.xxxEsx <CR><LF> x[Status],[x.xxxEsx] Measurement value (always engeneers' format)
        # 0 Measurement data okay, 1 Underrange, 2 Overrange
        # 3 Sensor error, 4 Sensor off, 5 No sensor, 6 Identification error
        if self.data[key]['used_sensor'] != None:
            self.data[key]['used_sensor'].flushInput()
            self.send_command(self.data[key]['used_sensor'],'PR%i\r\n' % self.data[key]['sensor'])  # request channel
            self.send_command(self.data[key]['used_sensor'],'\x05')  # enquire data
            #~ self.send_command('PR%i\r\n\x05' % (channel))  # request channel
            start_time = time.time()
            string_out = ''
            while string_out == '':
                if time.time()-start_time > 1:
                    return -2000, -2000
                try:
                    string_out = self.read_port(self.data[key]['used_sensor'])
                    if string_out == '':
                        time.sleep(0.01)
                        continue
                    string_split = string_out.split(',')          # splits read string into string[-1],string[0]
                    string_pres = str(string_split[1])            # pressure value converted to string
                    string_sta = int(string_split[0][-1])         # status value converted to int
                    pressure = float(string_pres)                 # float of pressure
                    status = int(string_sta)                      # status as integer value
                except ValueError:
                    pass
                except IndexError:
                    pass
                time.sleep(0.01)
            return self.conv_to_decode[status], pressure
        else:
            return -4000,-4000

    def read_mvcgauge(self, key):
        # communication described in MVC - manual
        if self.data[key]['used_sensor'] != None:
            input_command = 'rpv{}\r'.format(self.data[key]['sensor']).encode('utf-8')             # encode as utf-8
            convinput = self.to_bytes(input_command)                                               # convert to byte sequence
            self.data[key]['used_sensor'].write(convinput)                                         # send to wire
            time.sleep(0.05)
            out = ''                                                                               # string to hold the received message, empty one for new reading
            out += self.data[key]['used_sensor'].readline().decode('utf-8',errors='ignore')
            try:
                current=float(out.split(',')[1])
                if out.split(',')[0] != '0':
                    return -2000, -2000
                return 0,current
            except:
                return -3000,-3000
        else:
            return -4000,-4000

    def read_ionpump(self, key):
        # communication described in Gamma Vacuum - manual
        # important here: use crossed-rs232 cabel
        if self.data[key]['used_sensor']!= None:
            input_command = '~ 05 0A 01 00\r'.encode('utf-8')             # encode as utf-8
            convinput = self.to_bytes(input_command)                   # convert to byte sequence
            self.data[key]['used_sensor'].write(convinput)                                # send to wire
            time.sleep(0.05)
            out = ''                                            # string to hold the received message, empty one for new reading
            out += self.data[key]['used_sensor'].readline().decode('utf-8',errors='ignore')
            try:
                current=float(out.split(' ')[3])
                if out.split(' ')[1] != 'OK':
                    return -2000, -2000
                return 0,current
            except:
                return -3000,-3000
        else:
            return -4000,-4000

    def send_command(self, sensor, command):
        # send command to maxigauges
        # Takes ascii string 'command' and converts it to bytes to send it over serial connection
        input_command = command.encode('utf-8')             # encode as utf-8
        convinput = self.to_bytes(input_command)                   # convert to byte sequence
        sensor.write(convinput)                                # send to wire
        time.sleep(0.05)

    def read_port(self, sensor):
        # read port from maxigauges
        # Reads serial port, gets bytes over wire, decodes them with utf-8 and returns string with received message
        out = ''                                            # string to hold the received message, empty one for new reading
        out += sensor.readline().decode('utf-8')
        out += sensor.readline().decode('utf-8')               # we have to read twice
        return out

    def to_bytes(self, seq):
        # convert a sequence of int/str to a byte sequence and returns it
        if isinstance(seq, bytes):
            return seq
        elif isinstance(seq, bytearray):
            return bytes(seq)
        elif isinstance(seq, memoryview):
            return seq.tobytes()
        else:
            b = bytearray()
            for item in seq:
                # this one handles int and str for our emulation and ints for Python 3.x
                b.append(item)
            return bytes(b)

    def sleep_precise(self, duration, prec=0.0001):
        # better than time.sleep()
        start = time.time()
        while time.time() - start < duration - prec:
            time.sleep(prec)
        # print('sleep deviation: ', time.time()-start-duration)

    def transmit_outlet_code(self, turn_on=True):
        # transmit code to start helium measurement
        # Transmit a chosen code string using the GPIO transmitter
        NUM_ATTEMPTS = 20
        gpio = Adafruit_GPIO.get_platform_gpio()
        gpio.rpi_gpio.setmode(gpio.rpi_gpio.BCM)
        gpio.rpi_gpio.setup(CFG.TRANSMIT_PIN, Adafruit_GPIO.OUT)
        if turn_on:
            code = CFG.A_ON
        else:
            code = CFG.A_OFF
        for t in range(NUM_ATTEMPTS):
            self.lock.acquire()  # timing is critical
            for i in code:
                if i == '1':
                    gpio.output(CFG.TRANSMIT_PIN, 1)
                    self.sleep_precise(CFG.short_delay)
                    gpio.output(CFG.TRANSMIT_PIN, 0)
                    self.sleep_precise(CFG.long_delay)
                elif i == '0':
                    gpio.output(CFG.TRANSMIT_PIN, 1)
                    self.sleep_precise(CFG.long_delay)
                    gpio.output(CFG.TRANSMIT_PIN, 0)
                    self.sleep_precise(CFG.short_delay)
                else:
                    continue
            gpio.output(CFG.TRANSMIT_PIN, 0)
            self.lock.release()
            self.sleep_precise(CFG.extended_delay)
        gpio.cleanup()

    @_start_async(CFG.GRADIENT_RUNEVERY)
    def measure_gradient(self):
        for key in self.data.keys():
            self.gradient_data[self.gradient_data_current][key]['value'] = self.data[key]['value']
        self.gradient_data_timestamp[self.gradient_data_current] = dt.datetime.now()
        self.gradient_data_current += 1
        if self.gradient_data_current >= self.gradient_data_num:
            self.gradient_data_current = 0
        self.update_values_gradient()

    @_start_async(0.001)
    def measure_values_maxigauge(self):
        self.data_unreliable = collections.OrderedDict()
        for key in self.data:
            if self.data[key]['sensor_type'] == 'maxigauges':
                self.data[key]['status'], self.data[key]['value'] = self.read_maxigauge(key)

    @_start_async(0.001)
    def measure_values_ionpumps(self):
        self.data_unreliable = collections.OrderedDict()
        for key in self.data:
            if self.data[key]['sensor_type'] in ['ser_ion_cryo','ser_ion_prep','ser_ion_stm']:
                self.data[key]['status'], self.data[key]['value'] = self.read_ionpump(key)

    @_start_async(0.001)
    def measure_values_mvc_gauge_prep(self):
        if not self.ser_mvc_prep.is_open:
            self.init_serial_mvc_prep()

        self.data_unreliable = collections.OrderedDict()
        for key in self.data:
            if self.data[key]['sensor_type'] == 'mvc_prep':
                self.data[key]['status'], self.data[key]['value'] = self.read_mvcgauge(key)

    @_start_async(0.001)
    def measure_values_mvc_gauge_stm(self):
        if not self.ser_mvc_stm.is_open:
            self.init_serial_mvc_stm()

        self.data_unreliable = collections.OrderedDict()
        for key in self.data:
            if self.data[key]['sensor_type'] == 'mvc_stm':
                self.data[key]['status'], self.data[key]['value'] = self.read_mvcgauge(key)

    @_start_async(0.001)
    def measure_values_analog(self):
        self.data_unreliable = collections.OrderedDict()
        for key in self.data:
            if self.data[key]['sensor_type'] in ['ADC_diods', 'SPI0', 'SPI1', 'ADC_resistor']:
                self.data[key]['value'], self.data[key]['status'], self.data_unreliable[key] = self.read_analog(key)

    def read_analog(self, key):
        # measure values from adc chips and temperature chips
        measurement_points = 15
        extra_feature_multi = np.zeros(measurement_points)
        val_unreliable = False
        for cycle in range(measurement_points):
            # average over several values (measurement_points)
            if self.data[key]['sensor_type'] in ['ADC_diods','ADC_resistor']:
                # ADC chip
                if self.data[key]['used_sensor'] is None:
                    extra_feature_multi[cycle] = -4000
                else:
                    try:
                        extra_feature_multi[cycle] = self.data[key]['used_sensor'].read_volts(channel=self.data[key]['sensor'], gain=2)
                    except:
                        extra_feature_multi[cycle] = -4000
            elif self.data[key]['sensor_type'] in ['SPI0', 'SPI1']:
                # temperatuer chip
                if key == 'TLAB':
                    try:
                        extra_feature_multi[cycle] = self.data[key]['used_sensor'].read_internal_temp_c()
                    except:
                        extra_feature_multi[cycle] = -4000
                else:
                    try:
                        extra_feature_multi[cycle] = self.data[key]['used_sensor'].read_temp_c()
                    except:
                        extra_feature_multi[cycle] = -4000
        if min(extra_feature_multi) <= -1000:
            # remove non-physical values and declare them as unreliable
            val = min(extra_feature_multi)
            status = min(extra_feature_multi)
            val_unreliable = np.mean(extra_feature_multi)
        else:
            # if values make sense, average them and convert with respective calibration
            val = np.mean(extra_feature_multi)
            status = 0
            try:
                if self.data[key]['sensor_type'] not in ['TSAM', 'TLAB', 'TOM1', 'TOM2', 'TOM3']:
                    status = 0
                    if self.data[key]['sensor_type'] == 'ADC_diods':
                        val = self.temp_calib_diode(val)
                    elif self.data[key]['sensor_type'] == 'ADC_resistor':
                        val = self.temp_calib_resistor(val * 1000)
                    if key == 'TMAN':
                        val += CFG.kel_cel
            except ValueError:
                val = -3000
                status = -3000
            if np.std(extra_feature_multi) > 3:
                # if deviations are to large -> unreliable
                val = -2000
                status = -2000
                val_unreliable = np.mean(extra_feature_multi)
        if key in ['TSAM', 'TLAB', 'TOM1', 'TOM2', 'TOM3']:
            # range of temp-chip is between -200C and 1500C
            if val <= -198.0 or val >1500 and val not in CFG.conv_to_decode:
                val = -4000
                status = -4000
        if key in ['TAFM', 'TCRY', 'TMAN', 'TSTM']:
            if val >= 390.0:
                val = -4000
                status = -4000
        return val, status, val_unreliable

    def init_labels(self, gui):
        # initialize labels for GUI
        self.gui = gui
        colors = {key: self.data[key]['color'] for key in self.data}
        sizes = {key: self.data[key]['gui_size'] for key in self.data}
        labels_strs = [' {0: >4} {1: <6} = '.format(key, '[{}]'.format(self.data[key]['unit'])) for key in self.data]
        labels = collections.OrderedDict()
        gui_orders=[self.data[key]['gui_order'] for key in self.data]
        gui_orders_indices = np.argsort(gui_orders)
        for i in gui_orders_indices:
            labels[list(self.data.keys())[i]] = labels_strs[i]
        gui.init_labels(labels, colors, sizes)

    def update_values(self):
        # update label values in GUI
        values = {}
        for key in self.data:
            if int(self.data[key]['value']) in self.decoding_dict.keys():
                values[key] = self.decoding_dict[self.data[key]['value']]
            else:
                values[key] = '{0: {1}}'.format(float(self.data[key]['value']), self.data[key]['format'])
                if key in self.data_unreliable and self.data_unreliable[key] is not False:
                    values[key] += '*'
        for key in values:
            values[key] = '{0: >10}'.format(values[key])
        timestr = dt.datetime.now().strftime(CFG.date_fmt_display) + ' ' + self.fps
        self.gui.update_values(values, timestr)

    def update_values_gradient(self):
        # update gradient values in GUI
        values = {}
        for key in self.data:
            if self.data[key]['gui_size'] < 2:
                continue
            values[key] = '-'
            now = dt.datetime.now()
            if int(self.data[key]['value']) in self.decoding_dict.keys():
                continue
            yvalues = [gdata[key]['value'] if int(gdata[key]['value']) not in self.decoding_dict.keys() else np.nan for gdata in self.gradient_data]
            xvalues = [(gtime-now).seconds for gtime in self.gradient_data_timestamp]
            if np.nan not in yvalues:
                k = np.polyfit(xvalues, yvalues, 1)[0] * CFG.GRADIENT_SHOW
                if np.abs(k/self.data[key]['value']) > 1e-3:
                    values[key] = '{0: {1}}'.format(k, self.data[key]['format_gradient'])
        for key in values:
            values[key] = '{0:^6}'.format(values[key])
        self.gui.update_values_gradient(values)

    @_start_async(0.1, check_lastrun=True)
    def measure_helium(self):
        # measure helium
        helium_log = os.getcwd() + dt.datetime.now().strftime(CFG.HELIUM_LOG)
        helium_check_file = os.getcwd() + CFG.HELIUM_CHECK
        if not self.helium_check:
            if os.path.isfile(helium_check_file):
                # check if 'HELIUM_CHECK_file' exits -> if yes -> measure helium
                self.helium_check = True
                self.helium_save = False
                self.heliums = []
                self.main_loop_time = self.main_loop_time_slow
                time.sleep(0.8)
                self.transmit_outlet_code(turn_on=True)
                self.main_loop_time = self.main_loop_time_normal
                self.helium_turn_sensor_off = False  # we could also check whether it is on
                self.helium_turn_sensor_off_retries = 0
                self.t_hchk1 = time.time()
                print('Helium check.')
            if not os.path.isfile(helium_log):
                # write header to helium log file
                with open(helium_log, "w") as logfile_helium:
                    logfile_helium.write("Time\tLHE[mm]\n")
        if self.helium_check:
            self.t_hchk2 = time.time()
            # get values from ADC chip and average for a certain time
            if self.t_hchk2 - self.t_hchk1 > 10 and len(self.heliums) > 15 and np.mean(self.heliums) > 0:
                self.heliums = np.array(self.heliums, float)
                time.sleep(5)   # time to let helium level hardware start up
                helium = np.mean(self.heliums[abs(self.heliums - np.mean(self.heliums)) < 1 * np.std(self.heliums)])
                self.helium_save = True
                print('Helium check: successful.')
            elif self.t_hchk2 - self.t_hchk1 > 30:  # timeout
                helium = -3000
                self.helium_save = True
                print('Helium check: timeout.')
            elif self.t_hchk2 - self.t_hchk1 > 5:
                for i in range(5):
                    # print(self.adc[0].volts * 1000.0 / 2.0)
                    try:
                        helium = self.adc[0].volts * 1000.0 / 2.0
                    except Exception:
                        pass
                    finally:
                        if helium > 0:
                            self.heliums.append(helium)
        if self.helium_save:
            # save helium level to log
            now = dt.datetime.now()
            value = int(helium)
            self.helium_status['date_last_measured'] = now
            self.helium_status['value'] = value
            with open(helium_log, "a") as logfile_helium:
                logfile_helium.write("%s\t%i\n" % (now.strftime(CFG.date_fmt), value))
            if os.path.isfile(helium_check_file):
                os.remove(helium_check_file)
            self.helium_save = False
            self.helium_check = False
            self.helium_turn_sensor_off = True
            self.main_loop_time = self.main_loop_time_slow
            time.sleep(0.8)
            self.transmit_outlet_code(turn_on=False)
            self.main_loop_time = self.main_loop_time_normal
            print('Helium check: saved: {} mm.'.format(value))
            self.display_helium_now()
        if self.helium_turn_sensor_off:
            time.sleep(2 ** self.helium_turn_sensor_off_retries)
            if self.adc[0].volts < 0.02:  # baseline should be below this
                self.helium_turn_sensor_off = False
                print('Helium check: finished and sensor off.')
            else:
                self.main_loop_time = self.main_loop_time_slow
                time.sleep(2)
                self.transmit_outlet_code(turn_on=False)
                self.helium_turn_sensor_off_retries += 1
                print('Helium check: turning sensor off again, retry {}.'.format(self.helium_turn_sensor_off_retries))
                self.main_loop_time = self.main_loop_time_normal

    @_start_async(0.001)  # but it is not in the main loop
    def read_helium_from_log(self):
        data = self.measure_getlast()
        if data is not None and data.shape[0] > 0:
            date_last_measured = data.tail(1).index.to_pydatetime()[0]
            column_name = 'LHE[mm]'
            value = float(data.tail(1)[column_name])
        self.helium_status['date_last_measured'] = date_last_measured
        self.helium_status['value'] = value
        self.display_helium_now()

    @_start_async(1800)
    def display_helium(self):
        self.display_helium_now()

    def display_helium_now(self):
        value = self.helium_status['value']
        date_last_measured = self.helium_status['date_last_measured']
        if int(value) in self.decoding_dict:
            value = self.decoding_dict[value]
        if isinstance(value, str):
            str_value = '{}'.format(value)
        else:
            str_value = '{:.0f}'.format(value)
        str_out = "LHe: {} mm ({})".format(str_value, self.date_format_bot(date_last_measured))
        gui.update_helium(str_out) # forward helium lavel value to gui

    def date_format_bot(self, this_date):
        # returns a formatted date according to the config
        if dt.datetime.now() - this_date < dt.timedelta(hours=12):
            return this_date.strftime(CFG.date_fmt_display_he_short)
        else:
            return this_date.strftime(CFG.date_fmt_display_he)

    def str2date(self, x):
        # convert string from log to datetime object
        return dt.datetime(int(x[0:4]), int(x[5:7]), int(x[8:10]), int(x[11:13]), int(x[14:16]), int(x[17:19]))

    def measure_getlast(self):
        # reads last log entries for a specific measure entity
        helium_log = os.getcwd() + dt.datetime.now().strftime(CFG.HELIUM_LOG)
        df = pd.DataFrame()
        try:
            df = pd.read_csv(helium_log,
                             sep='\t',
                             comment='#',
                             parse_dates=True,
                             date_parser=self.str2date,
                             index_col=0,
                             dtype=np.float,
                             error_bad_lines=False)
        except pd.errors.EmptyDataError:
            return None
        return df[-1:]

    def check_day(self):
        self.pressurelogfile_name = dt.datetime.now().strftime(os.getcwd() + CFG.PRESSURE_LOGS)
        return(os.path.isfile(self.pressurelogfile_name))

    def save_header_to_log(self):
        if not self.check_day():  # write header if logfile doesnt exist
            self.log_writing_header = True
            header = 'Time\t'
            for key in self.data:
                if self.data[key]['log_to_file']:
                    header += '{0}[{1}]\t'.format(key, self.data[key]['unit'])
            header = header[:-1] + '\n'

            os.makedirs(os.path.dirname(self.pressurelogfile_name), exist_ok=True)
            with open(self.pressurelogfile_name, "a") as logfile:
                logfile.write(header)
            self.log_writing_header = False

    @_start_async(2, check_lastrun=True)
    def save_to_log(self):
        self.save_header_to_log()
        formattedData = [''] * len(self.data)
        for n, key in enumerate(self.data):
            if not self.data[key]['log_to_file']:
                continue
            if isinstance(self.data[key]['value'], str):
                formattedData[n] = '%s' % self.data[key]['value']
            else:
                formattedData[n] = '{0: {1}}'.format(float(self.data[key]['value']), self.data[key]['format'])
        with open(self.pressurelogfile_name, "a") as logfile:
            logfile.write("%s\t" % dt.datetime.now().strftime(CFG.date_fmt) + "\t".join(formattedData) + "\n")
        self.thread_save_to_log_running = False

    @_start_async(1, check_lastrun=True)
    def sanity_checks(self):
        # checks for values exceeding limits and initialize warning
        for key, ddict in self.data.items():
            if 'limit_max' in ddict:
                if ddict['value'] > ddict['limit_max']:
                    self.gui.warning(key, 'Warning: {}'.format(ddict['limit_max_warning']), key)
                else:
                    self.gui.dewarning(key)

    def thread_main_loop_sensors_start(self):
        self.thread_main_loop_sensors = threading.Timer(self.main_loop_time, self.main_loop_sensors)
        self.thread_main_loop_sensors.start()

    def main_loop_sensors(self):
        # measurement

        if ('ADC_diods' in self.sensor_types) or ('ADC_resistor' in self.sensor_types) or ('temp_chip' in self.sensor_types):
            self.measure_values_analog()
        if 'maxigauges' in self.sensor_types:
            self.measure_values_maxigauge()
        if 'mvc_prep' in self.sensor_types:
            self.measure_values_mvc_gauge_prep()
        if 'mvc_stm' in self.sensor_types:
            self.measure_values_mvc_gauge_stm()
        if ('ser_ion_prep' in self.sensor_types) or ('ser_ion_cryo' in self.sensor_types) or ('ser_ion_stm' in self.sensor_types):
            self.measure_values_ionpumps()
        if CFG.HELIUM != None:
            self.measure_helium()

        if CFG.GRADIENT_RUNEVERY > 0:
            self.measure_gradient()

        # display
        self.update_values()
        if CFG.HELIUM != None:
            self.display_helium()

        # log
        self.save_to_log()

        # sanity checks and warnings
        self.sanity_checks()

        if CFG.FPS_SHOW:
            self.fps = ' fps:{:4.1f}'.format(1 / (time.time() - self.time_loop))
        self.time_loop = time.time()

        if APP_RUNNING:
            self.thread_main_loop_sensors_start()
        else:
            self.cancel_all_threads()

    def main_loop_init(self):
        # if loop runs for the first time
        print('Reading initial sensor data.')
        if ('ADC_diods' in self.sensor_types) or ('ADC_resistor' in self.sensor_types) or ('temp_chip' in self.sensor_types):
            self.measure_values_analog()
        if 'maxigauges' in self.sensor_types:
            self.measure_values_maxigauge()
        if 'mvc_prep' in self.sensor_types:
            self.measure_values_mvc_gauge_prep()
        if 'mvc_stm' in self.sensor_types:
            self.measure_values_mvc_gauge_stm()
        if ('ser_ion_prep' in self.sensor_types) or ('ser_ion_cryo' in self.sensor_types) or ('ser_ion_stm' in self.sensor_types):
            self.measure_values_ionpumps()
        if CFG.HELIUM != None:
            self.read_helium_from_log()

        # wait for the first measurements
        if ('ADC_diods' in self.sensor_types) or ('ADC_resistor' in self.sensor_types) or ('temp_chip' in self.sensor_types):
            self.threads_running['measure_values_analog']['thread'].join()
        if 'maxigauges' in self.sensor_types:
            self.threads_running['measure_values_maxigauge']['thread'].join()
        if 'mvc_prep' in self.sensor_types:
            self.threads_running['measure_values_mvc_gauge_prep']['thread'].join()
        if 'mvc_stm' in self.sensor_types:
            self.threads_running['measure_values_mvc_gauge_stm']['thread'].join()
        if ('ser_ion_prep' in self.sensor_types) or ('ser_ion_cryo' in self.sensor_types) or ('ser_ion_stm' in self.sensor_types):
            self.threads_running['measure_values_ionpumps']['thread'].join()

        # self.threads_running['read_helium_from_log']['thread'].join()

        print('Starting main loop.')
        global APP_RUNNING
        APP_RUNNING = True
        self.thread_main_loop_sensors_start()

if __name__ == '__main__':
    print('Initializing measurement system.')
    msr = measure()
    print('Initializing GUI.')
    gui = GUI.initGUI()
    msr.init_labels(gui)
    gui.root.after(10, msr.main_loop_init)
    gui.startApp()

repo_name: MathiasPoe/RPi-read-lab-status


text:
<file_sep># mush-test
mush-testmush-testmush-testmush-testmush-test

repo_name: mberenh/mush-test


text:
<repo_name>rawuor/CODE2040<file_sep>/README.md
# CODE2040
Challenges
<file_sep>/stage1.py
__author__ = 'IvyAwuor'

import requests
import json
import datetime

#-----------------Register -------------------------#
token="<KEY>"
login = {'token':'<KEY>', 'github':'https://github.com/rawuor/CODE2040'}
#r = requests.post("http://challenge.code2040.org/api/register", data={'token': token, 'github': 'https://github.com/rawuor/CODE2040'})
r = requests.post("http://challenge.code2040.org/api/register", data=login)
print(r.status_code, r.reason)

#--------------Step II: Reverse a string -------------------#
r = requests.post("http://challenge.code2040.org/api/reverse", data={'token':token})
print(r.status_code, r.reason)
string = r.content;
#print(string)
print("The input string is " + string)
outputstring = string[::-1]
print("Reversed string is " + outputstring)

#Send back reversed string
tokstring = {'token':token, 'string':outputstring}
r = requests.post("http://challenge.code2040.org/api/reverse/validate",data=tokstring)
print(r.status_code, r.reason)

#--------------Step III: Needle in a haystack --------------#
#grab the dictionary
r = requests.post("http://challenge.code2040.org/api/haystack", data={'token':token})
#rec = loads(urlopen(r).read())['result']
print (r.status_code, r.reason)
result = r.content
#needle =  result["needle"]
result = json.loads(r.content)
needle = result["needle"]
haystack = result["haystack"]
position = haystack.index(needle)
print(position)
resultdict = {"token":token, "needle":position}
r = requests.post("http://challenge.code2040.org/api/haystack/validate", data=resultdict)
print(r.status_code, r.reason)

#------------------Step IV: Prefix --------------------------#
#grab dictionary
r = requests.post("http://challenge.code2040.org/api/prefix", data = {"token":token})
print(r.status_code, r.reason)
result = r.json()
prefix = result["prefix"]
array = result ["array"]
temp = []

for j in range(0, len(array)):
    if array[j].startswith(prefix) == False:
        temp.append(str(array[j]))
length = len(temp)
print (length)

jsonstring = {"token":token,"array":temp}
jsonstring = json.dumps(jsonstring)
headers = {'Content-type': 'application/json'}
r = requests.post("http://challenge.code2040.org/api/prefix/validate", data=jsonstring, headers=headers)
print(r.status_code, r.reason)

#----------------Step V: Dating game -------------------------------------------#
r = requests.post("http://challenge.code2040.org/api/dating", data={"token":token})
print(r.status_code, r.reason)
result = r.json();
date = result["datestamp"]
interval = result["interval"]

print("Datestamp is " + date)
datestamp = datetime.datetime.strptime(date, "%Y-%m-%dT%H:%M:%SZ")
datestampplus = datestamp + datetime.timedelta(0,interval)