Skip to content
GitLab
Commit 0d2a7a78 authored by Wei Shoulin's avatar Wei Shoulin
Browse files

update gaia3

parent eaa4815c
Hide whitespace changes
Inline Side-by-side
README.md
View file @ 0d2a7a78
...@@ -9,10 +9,13 @@ This package provides APIs to access csst's files and databases. ...@@ -9,10 +9,13 @@ This package provides APIs to access csst's files and databases.
This library can be installed with the following command: This library can be installed with the following command:
```bash ```bash
pip install git+https://github.com/astronomical-data-processing/csst-dfs-api-cluster.git git clone https://github.com/astronomical-data-processing/csst-dfs-api-cluster.git
cd csst-dfs-api-cluster
pip install -r requirements.txt
python setup.py install
``` ```
## Configuration ## Configuration
set enviroment variable set enviroment variable
CSST_DFS_CONFIG_SERVER = ip:port CSST_DFS_GATEWAY = ip:port
csst_dfs_api_cluster/common/catalog.py
View file @ 0d2a7a78
from ..common.client_config import ClientConfigurator import grpc
from ..common.service import ServiceProxy from csst_dfs_commons.models import Result
from csst_dfs_proto.common.ephem import ephem_pb2, ephem_pb2_grpc from csst_dfs_proto.common.ephem import ephem_pb2, ephem_pb2_grpc
from ..common.service import ServiceProxy
from ..common.constants import * from ..common.constants import *
import grpc
class CatalogApi(object): class CatalogApi(object):
def __init__(self): def __init__(self):
self.proxy = ServiceProxy(ClientConfigurator().gatewayCfg) self.proxy = ServiceProxy()
self.stub = self.proxy.insecure(ephem_pb2_grpc.EphemSearchSrvStub) self.stub = self.proxy.insecure(ephem_pb2_grpc.EphemSearchSrvStub)
def gaia3_query(self, ra: float, dec: float, radius: float, min_mag: float, max_mag: float, obstime: int, limit: int): def gaia3_query(self, ra: float, dec: float, radius: float, min_mag: float, max_mag: float, obstime: int, limit: int):
...@@ -32,8 +31,12 @@ class CatalogApi(object): ...@@ -32,8 +31,12 @@ class CatalogApi(object):
obstime = obstime, obstime = obstime,
limit = limit limit = limit
)) ))
return resp
except grpc.RpcError as identifier: if resp.success:
raise Exception("Rpc Error") return Result.ok_data(data=resp.records).append("totalCount", resp.totalCount)
else:
return Result.error(message = resp.message)
except grpc.RpcError as e:
return Result.error(message="%s:%s" % (e.code().value, e.details))
csst_dfs_api_cluster/common/client_config.py deleted 100644 → 0
View file @ eaa4815c
import os
from .config.grpc_loader import GrpcLoader
from .config.configurator import Configurator
from .utils import singleton
def configOptions(opts):
opts.address = os.getenv('CSST_DFS_CONFIG_SERVER', "localhost:9600")
@singleton
class ClientConfigurator(object):
def __init__(self) -> None:
self.configurator = Configurator(GrpcLoader(configOptions))
self.globalCfg = self.configurator.Global()
self.gatewayCfg = self.configurator.Gateway()
globalConfig = ClientConfigurator()
csst_dfs_api_cluster/common/config/changeset.py deleted 100644 → 0
View file @ eaa4815c
import time
class ChangeSet(object):
def __init__(self):
self._data = bytes([])
self._checksum = ""
self._format = ""
self._source = ""
self._timestamp = time.time()
@property
def data(self):
return self._data
@data.setter
def data(self, value):
self._data = value
@property
def checksum(self):
return self._checksum
@checksum.setter
def checksum(self, value):
self._checksum = value
@property
def format(self):
return self._format
@format.setter
def format(self, value):
self._format = value
@property
def source(self):
return self._source
@source.setter
def source(self, value):
self._source = value
@property
def timestamp(self):
return self._timestamp
@timestamp.setter
def timestamp(self, value):
self._timestamp = value
csst_dfs_api_cluster/common/config/configurator.py deleted 100644 → 0
View file @ eaa4815c
import orjson
from .model import Server as ServerCfg, Global as GlobalCfg, Gateway as GatewayCfg
class Configurator(object):
def __init__(self, loader):
self.loader = loader
change_set = self.loader.Read()
self.config = orjson.loads(change_set.data)
def Server(self, srv_name):
return self.Any(srv_name, ServerCfg)
def Gateway(self, path = "gateway"):
return self.Any(path, GatewayCfg)
def Global(self, path = "global"):
return self.Any(path, GlobalCfg)
def Any(self, path, ref_object):
paths = path.split(".")
cfg = self.config
for p in paths:
cfg = cfg[p]
if isinstance(cfg, dict):
return ref_object(**cfg)
else:
return ref_object(cfg)
\ No newline at end of file
csst_dfs_api_cluster/common/config/grpc_loader.py deleted 100644 → 0
View file @ eaa4815c
import grpc
from .grpc_options import GrpcOptions
from pymicro.config.grpc.proto import grpc_pb2, grpc_pb2_grpc
from .changeset import ChangeSet
class GrpcLoader(object):
def __init__(self, *args, **kwargs):
self.options = GrpcOptions()
for o in args:
o(self.options)
self._connect()
def Read(self):
request = grpc_pb2.ReadRequest(path = self.options.path)
response = self.stub.Read(request)
return self._to_changset(response.change_set)
def _connect(self):
self.channel = grpc.insecure_channel(self.options.address)
self.stub = grpc_pb2_grpc.SourceStub(self.channel)
def _to_changset(self, cs):
changeset = ChangeSet()
changeset.data = cs.data
changeset.checksum = cs.checksum
changeset.source = cs.source
changeset.format = cs.format
changeset.timestamp = cs.timestamp
return changeset
csst_dfs_api_cluster/common/config/grpc_options.py deleted 100644 → 0
View file @ eaa4815c
from pymicro.config.options import Options
class GrpcOptions(Options):
def __init__(self):
self.address = "localhost:9600"
self.path = "micro"
\ No newline at end of file
csst_dfs_api_cluster/common/config/model.py deleted 100644 → 0
View file @ eaa4815c
import dataclasses
@dataclasses.dataclass
class Server:
name: str=""
version: str=""
address: str=""
port: int=-1
@dataclasses.dataclass
class Global:
fitsFileRootDir: str=""
fileExternalPrefix: str=""
@dataclasses.dataclass
class Gateway:
enabled: bool=True
url: str=""
csst_dfs_api_cluster/common/config/options.py deleted 100644 → 0
View file @ eaa4815c
class Options(object):
def __init__(self):
pass
\ No newline at end of file
csst_dfs_api_cluster/common/config/proto/grpc.proto deleted 100644 → 0
View file @ eaa4815c
syntax = "proto3";
service Source {
rpc Read(ReadRequest) returns (ReadResponse) {};
rpc Watch(WatchRequest) returns (stream WatchResponse) {};
}
message ChangeSet {
bytes data = 1;
string checksum = 2;
string format = 3;
string source = 4;
int64 timestamp = 5;
}
message ReadRequest {
string path = 1;
}
message ReadResponse {
ChangeSet change_set = 1;
}
message WatchRequest {
string path = 1;
}
message WatchResponse {
ChangeSet change_set = 1;
}
csst_dfs_api_cluster/common/config/proto/grpc_pb2.py deleted 100644 → 0
View file @ eaa4815c
# -*- coding: utf-8 -*-
# Generated by the protocol buffer compiler. DO NOT EDIT!
# source: grpc.proto
import sys
_b=sys.version_info[0]<3 and (lambda x:x) or (lambda x:x.encode('latin1'))
from google.protobuf import descriptor as _descriptor
from google.protobuf import message as _message
from google.protobuf import reflection as _reflection
from google.protobuf import symbol_database as _symbol_database
# @@protoc_insertion_point(imports)
_sym_db = _symbol_database.Default()
DESCRIPTOR = _descriptor.FileDescriptor(
name='grpc.proto',
package='',
syntax='proto3',
serialized_options=None,
serialized_pb=_b('\n\ngrpc.proto\"^\n\tChangeSet\x12\x0c\n\x04\x64\x61ta\x18\x01 \x01(\x0c\x12\x10\n\x08\x63hecksum\x18\x02 \x01(\t\x12\x0e\n\x06\x66ormat\x18\x03 \x01(\t\x12\x0e\n\x06source\x18\x04 \x01(\t\x12\x11\n\ttimestamp\x18\x05 \x01(\x03\"\x1b\n\x0bReadRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\".\n\x0cReadResponse\x12\x1e\n\nchange_set\x18\x01 \x01(\x0b\x32\n.ChangeSet\"\x1c\n\x0cWatchRequest\x12\x0c\n\x04path\x18\x01 \x01(\t\"/\n\rWatchResponse\x12\x1e\n\nchange_set\x18\x01 \x01(\x0b\x32\n.ChangeSet2[\n\x06Source\x12%\n\x04Read\x12\x0c.ReadRequest\x1a\r.ReadResponse\"\x00\x12*\n\x05Watch\x12\r.WatchRequest\x1a\x0e.WatchResponse\"\x00\x30\x01\x62\x06proto3')
)
_CHANGESET = _descriptor.Descriptor(
name='ChangeSet',
full_name='ChangeSet',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
name='data', full_name='ChangeSet.data', index=0,
number=1, type=12, cpp_type=9, label=1,
has_default_value=False, default_value=_b(""),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
name='checksum', full_name='ChangeSet.checksum', index=1,
number=2, type=9, cpp_type=9, label=1,
has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
name='format', full_name='ChangeSet.format', index=2,
number=3, type=9, cpp_type=9, label=1,
has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
name='source', full_name='ChangeSet.source', index=3,
number=4, type=9, cpp_type=9, label=1,
has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
_descriptor.FieldDescriptor(
name='timestamp', full_name='ChangeSet.timestamp', index=4,
number=5, type=3, cpp_type=2, label=1,
has_default_value=False, default_value=0,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
],
extensions=[
],
nested_types=[],
enum_types=[
],
serialized_options=None,
is_extendable=False,
syntax='proto3',
extension_ranges=[],
oneofs=[
],
serialized_start=14,
serialized_end=108,
)
_READREQUEST = _descriptor.Descriptor(
name='ReadRequest',
full_name='ReadRequest',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
name='path', full_name='ReadRequest.path', index=0,
number=1, type=9, cpp_type=9, label=1,
has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
],
extensions=[
],
nested_types=[],
enum_types=[
],
serialized_options=None,
is_extendable=False,
syntax='proto3',
extension_ranges=[],
oneofs=[
],
serialized_start=110,
serialized_end=137,
)
_READRESPONSE = _descriptor.Descriptor(
name='ReadResponse',
full_name='ReadResponse',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
name='change_set', full_name='ReadResponse.change_set', index=0,
number=1, type=11, cpp_type=10, label=1,
has_default_value=False, default_value=None,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
],
extensions=[
],
nested_types=[],
enum_types=[
],
serialized_options=None,
is_extendable=False,
syntax='proto3',
extension_ranges=[],
oneofs=[
],
serialized_start=139,
serialized_end=185,
)
_WATCHREQUEST = _descriptor.Descriptor(
name='WatchRequest',
full_name='WatchRequest',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
name='path', full_name='WatchRequest.path', index=0,
number=1, type=9, cpp_type=9, label=1,
has_default_value=False, default_value=_b("").decode('utf-8'),
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
],
extensions=[
],
nested_types=[],
enum_types=[
],
serialized_options=None,
is_extendable=False,
syntax='proto3',
extension_ranges=[],
oneofs=[
],
serialized_start=187,
serialized_end=215,
)
_WATCHRESPONSE = _descriptor.Descriptor(
name='WatchResponse',
full_name='WatchResponse',
filename=None,
file=DESCRIPTOR,
containing_type=None,
fields=[
_descriptor.FieldDescriptor(
name='change_set', full_name='WatchResponse.change_set', index=0,
number=1, type=11, cpp_type=10, label=1,
has_default_value=False, default_value=None,
message_type=None, enum_type=None, containing_type=None,
is_extension=False, extension_scope=None,
serialized_options=None, file=DESCRIPTOR),
],
extensions=[
],
nested_types=[],
enum_types=[
],
serialized_options=None,
is_extendable=False,
syntax='proto3',
extension_ranges=[],
oneofs=[
],
serialized_start=217,
serialized_end=264,
)
_READRESPONSE.fields_by_name['change_set'].message_type = _CHANGESET
_WATCHRESPONSE.fields_by_name['change_set'].message_type = _CHANGESET
DESCRIPTOR.message_types_by_name['ChangeSet'] = _CHANGESET
DESCRIPTOR.message_types_by_name['ReadRequest'] = _READREQUEST
DESCRIPTOR.message_types_by_name['ReadResponse'] = _READRESPONSE
DESCRIPTOR.message_types_by_name['WatchRequest'] = _WATCHREQUEST
DESCRIPTOR.message_types_by_name['WatchResponse'] = _WATCHRESPONSE
_sym_db.RegisterFileDescriptor(DESCRIPTOR)
ChangeSet = _reflection.GeneratedProtocolMessageType('ChangeSet', (_message.Message,), {
'DESCRIPTOR' : _CHANGESET,
'__module__' : 'grpc_pb2'
# @@protoc_insertion_point(class_scope:ChangeSet)
})
_sym_db.RegisterMessage(ChangeSet)
ReadRequest = _reflection.GeneratedProtocolMessageType('ReadRequest', (_message.Message,), {
'DESCRIPTOR' : _READREQUEST,
'__module__' : 'grpc_pb2'
# @@protoc_insertion_point(class_scope:ReadRequest)
})
_sym_db.RegisterMessage(ReadRequest)
ReadResponse = _reflection.GeneratedProtocolMessageType('ReadResponse', (_message.Message,), {
'DESCRIPTOR' : _READRESPONSE,
'__module__' : 'grpc_pb2'
# @@protoc_insertion_point(class_scope:ReadResponse)
})
_sym_db.RegisterMessage(ReadResponse)
WatchRequest = _reflection.GeneratedProtocolMessageType('WatchRequest', (_message.Message,), {
'DESCRIPTOR' : _WATCHREQUEST,
'__module__' : 'grpc_pb2'
# @@protoc_insertion_point(class_scope:WatchRequest)
})
_sym_db.RegisterMessage(WatchRequest)
WatchResponse = _reflection.GeneratedProtocolMessageType('WatchResponse', (_message.Message,), {
'DESCRIPTOR' : _WATCHRESPONSE,
'__module__' : 'grpc_pb2'
# @@protoc_insertion_point(class_scope:WatchResponse)
})
_sym_db.RegisterMessage(WatchResponse)
_SOURCE = _descriptor.ServiceDescriptor(
name='Source',
full_name='Source',
file=DESCRIPTOR,
index=0,
serialized_options=None,
serialized_start=266,
serialized_end=357,
methods=[
_descriptor.MethodDescriptor(
name='Read',
full_name='Source.Read',
index=0,
containing_service=None,
input_type=_READREQUEST,
output_type=_READRESPONSE,
serialized_options=None,
),
_descriptor.MethodDescriptor(
name='Watch',
full_name='Source.Watch',
index=1,
containing_service=None,
input_type=_WATCHREQUEST,
output_type=_WATCHRESPONSE,
serialized_options=None,
),
])
_sym_db.RegisterServiceDescriptor(_SOURCE)
DESCRIPTOR.services_by_name['Source'] = _SOURCE
# @@protoc_insertion_point(module_scope)
csst_dfs_api_cluster/common/config/proto/grpc_pb2_grpc.py deleted 100644 → 0
View file @ eaa4815c
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT!
import grpc
from . import grpc_pb2 as grpc__pb2
class SourceStub(object):
# missing associated documentation comment in .proto file
pass
def __init__(self, channel):
"""Constructor.
Args:
channel: A grpc.Channel.
"""
self.Read = channel.unary_unary(
'/Source/Read',
request_serializer=grpc__pb2.ReadRequest.SerializeToString,
response_deserializer=grpc__pb2.ReadResponse.FromString,
)
self.Watch = channel.unary_stream(
'/Source/Watch',
request_serializer=grpc__pb2.WatchRequest.SerializeToString,
response_deserializer=grpc__pb2.WatchResponse.FromString,
)
class SourceServicer(object):
# missing associated documentation comment in .proto file
pass
def Read(self, request, context):
# missing associated documentation comment in .proto file
pass
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def Watch(self, request, context):
# missing associated documentation comment in .proto file
pass
context.set_code(grpc.StatusCode.UNIMPLEMENTED)
context.set_details('Method not implemented!')
raise NotImplementedError('Method not implemented!')
def add_SourceServicer_to_server(servicer, server):
rpc_method_handlers = {
'Read': grpc.unary_unary_rpc_method_handler(
servicer.Read,
request_deserializer=grpc__pb2.ReadRequest.FromString,
response_serializer=grpc__pb2.ReadResponse.SerializeToString,
),
'Watch': grpc.unary_stream_rpc_method_handler(
servicer.Watch,
request_deserializer=grpc__pb2.WatchRequest.FromString,
response_serializer=grpc__pb2.WatchResponse.SerializeToString,
),
}
generic_handler = grpc.method_handlers_generic_handler(
'Source', rpc_method_handlers)
server.add_generic_rpc_handlers((generic_handler,))
csst_dfs_api_cluster/common/service.py
View file @ 0d2a7a78
import os
import grpc import grpc
import random
class _Service:
def __init__(self, data):
self._data = data
@property
def name(self):
raise NotImplementedError()
@property
def version(self):
raise NotImplementedError()
@property
def metadata(self):
raise NotImplementedError()
@property
def id(self):
raise NotImplementedError()
@property
def address(self):
raise NotImplementedError()
@property
def port(self):
raise NotImplementedError()
@property
def node_metadata(self):
raise NotImplementedError()
def __str__(self):
return '{}:{}'.format(self.name, self.address)
class Service(_Service):
def __init__(self, data):
self._data = data
@property
def name(self):
return self._data['name']
@property
def version(self):
return self._data['version']
@property
def metadata(self):
return self._data['metadata']
@property
def _node(self):
return random.choice(self._data['nodes'])
@property
def id(self):
return self._node['id']
@property
def address(self):
return self._node['address']
@property
def node_metadata(self):
return self._node['metadata']
class ServiceProxy: class ServiceProxy:
def __init__(self, gatewayCfg = None): def __init__(self):
self.gatewayCfg = gatewayCfg self.gateway = os.getenv("CSST_DFS_GATEWAY",'172.31.248.218:30880')
def insecure(self, stubCls): def insecure(self, stubCls):
channel = grpc.insecure_channel(self.gatewayCfg.url) options = [('grpc.max_send_message_length', 1000 * 1024 * 1024),
('grpc.max_receive_message_length', 1000 * 1024 * 1024),
('grpc.enable_retries', 1),
('grpc.service_config', '{ "retryPolicy":{ "maxAttempts": 4, "initialBackoff": "0.1s", "maxBackoff": "1s", "backoffMutiplier": 2, "retryableStatusCodes": [ "UNAVAILABLE" ] } }')]
channel = grpc.insecure_channel(self.gateway, options=options)
return stubCls(channel) return stubCls(channel)
\ No newline at end of file
csst_dfs_api_cluster/ifs/fits.py
View file @ 0d2a7a78
...@@ -3,7 +3,6 @@ import grpc ...@@ -3,7 +3,6 @@ import grpc
from csst_dfs_proto.ifs.fits import fits_pb2, fits_pb2_grpc from csst_dfs_proto.ifs.fits import fits_pb2, fits_pb2_grpc
from ..common.client_config import ClientConfigurator
from ..common.service import ServiceProxy from ..common.service import ServiceProxy
from ..common.utils import * from ..common.utils import *
from ..common.constants import UPLOAD_CHUNK_SIZE from ..common.constants import UPLOAD_CHUNK_SIZE
...@@ -11,7 +10,7 @@ from ..common.constants import UPLOAD_CHUNK_SIZE ...@@ -11,7 +10,7 @@ from ..common.constants import UPLOAD_CHUNK_SIZE
class FitsApi(object): class FitsApi(object):
def __init__(self, sub_system = "ifs"): def __init__(self, sub_system = "ifs"):
self.sub_system = sub_system self.sub_system = sub_system
self.proxy = ServiceProxy(ClientConfigurator().gatewayCfg) self.proxy = ServiceProxy()
self.stub = self.proxy.insecure(fits_pb2_grpc.FitsSrvStub) self.stub = self.proxy.insecure(fits_pb2_grpc.FitsSrvStub)
def find(self, **kwargs): def find(self, **kwargs):
......
csst_dfs_api_cluster/ifs/reffits.py
View file @ 0d2a7a78
...@@ -2,7 +2,6 @@ ...@@ -2,7 +2,6 @@
import os import os
import grpc import grpc
from ..common.client_config import ClientConfigurator
from ..common.service import ServiceProxy from ..common.service import ServiceProxy
from ..common.utils import * from ..common.utils import *
from ..common.constants import * from ..common.constants import *
......
csst_dfs_api_cluster/ifs/result0.py
View file @ 0d2a7a78
...@@ -2,7 +2,6 @@ ...@@ -2,7 +2,6 @@
import os import os
import grpc import grpc
from ..common.client_config import ClientConfigurator
from ..common.service import ServiceProxy from ..common.service import ServiceProxy
from ..common.utils import * from ..common.utils import *
from ..common.constants import * from ..common.constants import *
......
csst_dfs_api_cluster/ifs/result1.py
View file @ 0d2a7a78
...@@ -2,7 +2,6 @@ ...@@ -2,7 +2,6 @@
import os import os
import grpc import grpc
from ..common.client_config import ClientConfigurator
from ..common.service import ServiceProxy from ..common.service import ServiceProxy
from ..common.utils import * from ..common.utils import *
from ..common.constants import * from ..common.constants import *
......
requirements.txt
View file @ 0d2a7a78
astropy astropy>=4.0
grpc grpcio>=1.28.1
orjson orjson>=3.0.2
git+https://github.com/astronomical-data-processing/csst-dfs-proto-py.git git+https://github.com/astronomical-data-processing/csst-dfs-proto-py.git
\ No newline at end of file git+https://github.com/astronomical-data-processing/csst-dfs-commons.git
\ No newline at end of file
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment