Changeset: d8f1402029e1 for MonetDB URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=d8f1402029e1 Removed Files: clients/iotclient/src/Utilities/numberutilities.py Modified Files: clients/iotclient/README clients/iotclient/src/Flask/restresources.py clients/iotclient/src/Settings/filesystem.py clients/iotclient/src/Settings/iotlogger.py clients/iotclient/src/Settings/mapiconnection.py clients/iotclient/src/Streams/datatypes.py clients/iotclient/src/Streams/jsonschemas.py clients/iotclient/src/Streams/semanticvalidation.py clients/iotclient/src/Streams/streams.py clients/iotclient/src/Streams/streamscontext.py clients/iotclient/src/main.py Branch: iot Log Message:
Corrected several possible bugs, added decimal/numeric serialization. diffs (truncated from 439 to 300 lines): diff --git a/clients/iotclient/README b/clients/iotclient/README --- a/clients/iotclient/README +++ b/clients/iotclient/README @@ -23,8 +23,7 @@ REST resources available on iot webserve Currently supports most of MonetDB datatypes. An implicit Timestamp column is added with the tuples arrival date. The flushing of the baskets can be either Tuple or Time based, as specied on the streams' creation. -More details coming soon. +I will migrate this documentation into pydocs soon :) Maintainer: Pedro Ferreira at CWI, email: [email protected] - diff --git a/clients/iotclient/src/Flask/restresources.py b/clients/iotclient/src/Flask/restresources.py --- a/clients/iotclient/src/Flask/restresources.py +++ b/clients/iotclient/src/Flask/restresources.py @@ -6,11 +6,11 @@ from flask import request from flask_restful import Resource from jsonschema import Draft4Validator, FormatChecker from tzlocal import get_localzone -from src.Streams.jsonschemas import CREATE_STREAMS_SCHEMA, DELETE_STREAMS_SCHEMA -from src.Streams.streamscontext import IOTStreamsException, IOTStreams +from Streams.jsonschemas import CREATE_STREAMS_SCHEMA, DELETE_STREAMS_SCHEMA +from Streams.streamscontext import IOTStreamsException, IOTStreams Stream_context = IOTStreams() -local_tz = get_localzone() # for the correction of dates we must add the timezone +local_tz = get_localzone() # for the correction of dates we must add the system's timezone class StreamInput(Resource): diff --git a/clients/iotclient/src/Settings/filesystem.py b/clients/iotclient/src/Settings/filesystem.py --- a/clients/iotclient/src/Settings/filesystem.py +++ b/clients/iotclient/src/Settings/filesystem.py @@ -2,6 +2,8 @@ import os import sys baskets_base_location = None +BASKETS_BASE_DIRECTORY = "baskets" + if sys.platform in ("linux", "linux2", "darwin"): filesystem_location = '/etc/iotcollector' @@ -12,16 +14,28 @@ elif sys.platform == "win32": def set_filesystem_location(new_location): global filesystem_location - if os.path.isdir(new_location): - filesystem_location = new_location - else: - print >> sys.stderr, "The provided filesystem doesn't exist!" + try: + if os.path.isdir(new_location): + filesystem_location = new_location + else: + print >> sys.stderr, "The provided filesystem doesn't exist!" + sys.exit(1) + except (Exception, OSError) as ex: + print >> sys.stderr, ex sys.exit(1) def init_file_system(): global baskets_base_location - baskets_base_location = os.path.join(filesystem_location, "baskets") - if not os.path.exists(baskets_base_location): - os.makedirs(baskets_base_location) + try: + baskets_base_location = os.path.join(filesystem_location, BASKETS_BASE_DIRECTORY) + if not os.path.exists(baskets_base_location): + os.makedirs(baskets_base_location) + except (Exception, OSError) as ex: + print >> sys.stderr, ex + sys.exit(1) + + +def get_baskets_base_location(): + return baskets_base_location diff --git a/clients/iotclient/src/Settings/iotlogger.py b/clients/iotclient/src/Settings/iotlogger.py --- a/clients/iotclient/src/Settings/iotlogger.py +++ b/clients/iotclient/src/Settings/iotlogger.py @@ -29,6 +29,6 @@ def init_logging(): log_handler = logging.FileHandler(logging_location, mode='a+') log_handler.setFormatter(formatter) logger.addHandler(log_handler) - except Exception as ex: - print >> sys.stderr, ex.message + except (Exception, OSError) as ex: + print >> sys.stderr, ex sys.exit(1) diff --git a/clients/iotclient/src/Settings/mapiconnection.py b/clients/iotclient/src/Settings/mapiconnection.py --- a/clients/iotclient/src/Settings/mapiconnection.py +++ b/clients/iotclient/src/Settings/mapiconnection.py @@ -1,5 +1,6 @@ import sys import pymonetdb +import getpass Connection = None @@ -7,11 +8,12 @@ Connection = None def init_monetdb_connection(hostname, port, user_name, database): global Connection - user_password = 'monetdb' # raw_input("Enter Password: ") + user_password = getpass.getpass(prompt='User password:') try: Connection = pymonetdb.connect(hostname=hostname, port=port, username=user_name, password=user_password, database=database) + print >> sys.stdout, 'User %s connection successful to the database %s' % (user_name, database) except Exception as ex: print >> sys.stderr, ex.message sys.exit(1) @@ -26,7 +28,8 @@ def mapi_create_stream(schema, stream, c Connection.execute("CREATE SCHEMA " + schema + ";") except: pass - Connection.execute(''.join(["CREATE STREAM TABLE ", schema, ".", stream, " (", columns, ");"])) + sql_str = ''.join(["CREATE STREAM TABLE ", schema, ".", stream, " (", columns, ");"]) + Connection.execute(sql_str) def mapi_flush_baskets(schema, stream, baskets): diff --git a/clients/iotclient/src/Streams/datatypes.py b/clients/iotclient/src/Streams/datatypes.py --- a/clients/iotclient/src/Streams/datatypes.py +++ b/clients/iotclient/src/Streams/datatypes.py @@ -3,14 +3,15 @@ import dateutil import itertools import struct import copy +import math from abc import ABCMeta, abstractmethod from dateutil import parser -from src.Streams.jsonschemas import UUID_REG +from jsonschemas import UUID_REG -# TODO later check the byte order https://docs.python.org/2/library/struct.html#byte-order-size-and-alignment -# TODO Also check the consequences of aligment on packing HUGEINTs! -# TODO The null constants might change from system to system due to different CPU's +# Later check the byte order https://docs.python.org/2/library/struct.html#byte-order-size-and-alignment +# Also check the consequences of aligment on packing HUGEINTs! +# The null constants might change from system to system due to different CPU's ALIGNMENT = '<' # for now is little-endian for Intel CPU's NIL_STRING = "\200" @@ -19,6 +20,7 @@ INT8_MIN = 0x80 INT16_MIN = 0x8000 INT32_MIN = 0x80000000 INT64_MIN = 0x8000000000000000 +INT64_MAX = 0xFFFFFFFFFFFFFFFF INT128_MIN = 0x80000000000000000000000000000000 FLOAT_NAN = struct.unpack('f', '\xff\xff\x7f\xff')[0] @@ -94,10 +96,10 @@ class StreamDataType(object): return json_data @abstractmethod - def get_sql_params(self): + def get_sql_params(self): # get other possible parameters such as if nullable, default value, maximum and minimum return [] - def create_stream_sql(self): + def create_stream_sql(self): # get column creation statement on SQL array = [self._column_name, " "] array.extend(self.get_sql_params()) return ''.join(array) @@ -112,6 +114,12 @@ class BaseTextType(StreamDataType): def get_nullable_constant(self): return NIL_STRING + def set_default_value(self, default_value): + self._default_value = None + + def add_json_schema_entry(self, schema): + pass + def prepare_parameters(self): return {'lengths_sum': 0} @@ -124,6 +132,9 @@ class BaseTextType(StreamDataType): string_pack = "".join(extracted_values) return struct.pack(ALIGNMENT + str(parameters['lengths_sum']) + 's', string_pack) + def get_sql_params(self): + return [] + class TextType(BaseTextType): """Covers: CHAR, VARCHAR, CHARACTER VARYING, TEXT, STRING, CLOB and CHARACTER LARGE OBJECT @@ -335,8 +346,6 @@ class SmallIntegerType(NumberBaseType): def pack_parsed_values(self, extracted_values, counter, parameters): return struct.pack(ALIGNMENT + str(counter) + self._pack_sym, extracted_values) -max_int64 = 0xFFFFFFFFFFFFFFFF - class HugeIntegerType(NumberBaseType): """Covers: HUGEINT""" @@ -355,11 +364,11 @@ class HugeIntegerType(NumberBaseType): return int(value) def process_next_value(self, entry, counter, parameters, errors): - return [entry & max_int64, (entry >> 64) & max_int64] + return [entry & INT64_MAX, (entry >> 64) & INT64_MAX] def pack_parsed_values(self, extracted_values, counter, parameters): - concat_array = list(itertools.chain(*extracted_values)) - return struct.pack(ALIGNMENT + str(counter << 1) + 'Q', *concat_array) + extracted_values = list(itertools.chain(*extracted_values)) + return struct.pack(ALIGNMENT + str(counter << 1) + 'Q', *extracted_values) class FloatType(NumberBaseType): @@ -388,7 +397,7 @@ class FloatType(NumberBaseType): return struct.pack(ALIGNMENT + str(counter) + self._pack_sym, *extracted_values) -class DecimalType(NumberBaseType): # TODO finish this class, how to serialize these values +class DecimalType(NumberBaseType): """Covers: DECIMAL and NUMERIC""" def __init__(self, **kwargs): @@ -402,22 +411,51 @@ class DecimalType(NumberBaseType): # TO else: self._scale = 0 + if self._scale > self._precision: + raise Exception('The scale must be between 0 and the precision!') + + if self._precision <= 2: # calculate the number of bytes to use according to the precision + self._pack_sym = 'b' + elif 2 < self._precision <= 4: + self._pack_sym = 'h' + elif 4 < self._precision <= 8: + self._pack_sym = 'i' + elif 8 < self._precision <= 18: + self._pack_sym = 'q' + elif 18 < self._precision <= 38: + self._pack_sym = 'Q' + + self._nullable_constant = {'b': INT8_MIN, 'h': INT16_MIN, 'i': INT32_MIN, 'q': INT64_MIN, 'Q': INT128_MIN} \ + .get(self._pack_sym) + def add_json_schema_entry(self, schema): super(DecimalType, self).add_json_schema_entry(schema) schema[self._column_name]['type'] = 'number' def get_nullable_constant(self): - return 0 + return self._nullable_constant def process_default_value(self, value): - return float(value) + number_digits = int(math.ceil(math.log10(abs(value)))) + if number_digits > self._precision: + raise Exception('Too many digits on default value: %s > %s' % (number_digits, self._precision)) + return int(value) def process_next_value(self, entry, counter, parameters, errors): - # precision, scale = precision_and_scale(entry) - return float(entry) + number_digits = int(math.ceil(math.log10(abs(entry)))) + if number_digits > self._precision: + errors[counter] = 'Too many digits: %s > %s' % (number_digits, self._precision) + parsed_value = int(entry) + if self._pack_sym != 'Q': + return parsed_value + else: + return [parsed_value & INT64_MAX, (parsed_value >> 64) & INT64_MAX] def pack_parsed_values(self, extracted_values, counter, parameters): - return struct.pack(ALIGNMENT + str(counter) + 'd', *extracted_values) + if self._pack_sym == 'Q': + extracted_values = list(itertools.chain(*extracted_values)) + counter <<= 1 + return struct.pack(ALIGNMENT + str(counter) + self._pack_sym, *extracted_values) def to_json_representation(self): json_value = super(DecimalType, self).to_json_representation() @@ -425,6 +463,11 @@ class DecimalType(NumberBaseType): # TO json_value['scale'] = self._scale return json_value + def get_sql_params(self): # override the column type to include the precision and scale + array = super(DecimalType, self).get_sql_params() + array[0] = ''.join([self._data_type, " (", str(self._precision), ",", str(self._scale), ")"]) + return array + class BaseDateTimeType(StreamDataType): # The validation of time variables can't be done on the schema __metaclass__ = ABCMeta @@ -439,6 +482,7 @@ class BaseDateTimeType(StreamDataType): self._maximum = self.parse_entry(kwargs['maximum']) if hasattr(self, '_minimum') and hasattr(self, '_maximum') and self._minimum > self._maximum: raise Exception('The minimum value is higher than the maximum!') + self._default_value_text = None # needed later for the SQL creation statement _______________________________________________ checkin-list mailing list [email protected] https://www.monetdb.org/mailman/listinfo/checkin-list
