FanchenBao

import logging
import logging.handlers
import multiprocessing
from time import sleep
from random import random, randint


# Almost the same as the demo code, but added `console_handler` to directly
# read logging info from the console
def listener_configurer():

FanchenBao

def main():
    queue = multiprocessing.Queue(-1)
    # set up main logger following example from work_process
    worker_configurer(queue)
    main_logger = logging.getLogger('main')

    listener = multiprocessing.Process(
        target=listener_process, args=(queue,))
    listener.start()

FanchenBao

def main():
    queue = multiprocessing.Queue(-1)
    listener = multiprocessing.Process(
        target=listener_process, args=(queue,))
    listener.start()

    # set up main logger after listener
    worker_configurer(queue)
    main_logger = logging.getLogger('main')
    main_logger.info('Logging from main')

FanchenBao

# comment out logger configuration for worker process
def worker_process(queue):
#    worker_configurer(queue)
    for i in range(3):
        sleep(random())
        innerlogger = logging.getLogger('worker')
        innerlogger.info(f'Logging a random number {randint(0, 10)}')

FanchenBao

# logger configuration uncommented in worker_process
def worker_process(queue):
    worker_configurer(queue)
    for i in range(3):
        sleep(random())
        innerlogger = logging.getLogger('worker')
        innerlogger.info(f'Logging a random number {randint(0, 10)}')


# logger configuration commented out in main

FanchenBao

import logging
from logging import handlers
from time import sleep


def listener_configurer():
    root = logging.getLogger()
    file_handler = handlers.RotatingFileHandler('mptest.log', 'a', 300, 10)
    console_handler = logging.StreamHandler()
    formatter = logging.Formatter('%(asctime)s %(processName)-10s %(name)s %(levelname)-8s %(message)s')

FanchenBao

import re
from typing import List


"""
This scripts accumulates the 'total' column of vnstat output in this format:

wlan0  /  hourly

         hour        rx      |     tx      |    total    |   avg. rate

FanchenBao

from multiprocessing.queues import Queue
import multiprocessing

# The following implementation of custom MyQueue to avoid NotImplementedError
# when calling queue.qsize() in MacOS X comes almost entirely from this github
# discussion: https://github.com/keras-team/autokeras/issues/368
# Necessary modification is made to make the code compatible with Python3.


class SharedCounter(object):

FanchenBao

import requests


credential_provider_endpoint = 'https://<your_credentials_provider_endpoint>/role-aliases/iot-s3-access-role-alias/credentials'
device_cert_path = '<path_to_device_cert>'
device_private_key_path = '<path_to_device_private_key>'
resp = requests.get(
    credential_provider_endpoint, 
    headers={'x-amzn-iot-thingname': 'TestThing'}, 
    cert=(device_cert_path, device_private_key_path),

FanchenBao

import boto3
import requests



def obtain_temporary_credentials() -> Tuple[str, str, str]:
    """Obtain temporary credentials from AWS IoT device certificate and private key."""
    credential_provider_endpoint = 'https://<your_credentials_provider_endpoint>/role-aliases/iot-s3-access-role-alias/credentials'
    device_cert_path = '<path_to_device_cert>'
    device_private_key_path = '<path_to_device_private_key>'