diff --git a/.env b/.env index 513355e..fdd03ba 100644 --- a/.env +++ b/.env @@ -1,4 +1,5 @@ #General settings +TZ=America/Lima LC_ALL=C.UTF-8 SIRM_SITE= PROC_SITE= @@ -25,14 +26,14 @@ PGDATA=/var/lib/postgresql/data #Volumes - path EXPOSE_SIRM=./volumes/sirm EXPOSE_PROC=./volumes/proc -EXPOSE_SCHAIN=./volumes/schain EXPOSE_CAM=/path/to/cam +EXPOSE_SCHAIN=./volumes/schain EXPOSE_NAS=/path/to/nas_data EXPOSE_PGDATA=/path/to/pg_data EXPOSE_CERTS=/path/to/certs EXPOSE_DHPARAM=/path/to/dhparam #Superuser settings -SIRM_USER=admin -SIRM_PASSWORD=soporte -SIRM_EMAIL=admin@igp.gob.pe \ No newline at end of file +SIRM_USER=***** +SIRM_PASSWORD=******* +SIRM_EMAIL=*****@igp.gob.pe \ No newline at end of file diff --git a/docker-compose.yml b/docker-compose.yml index e358395..69d4ae7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -98,6 +98,7 @@ services: - POSTGRES_PORT_5432_TCP_PORT=${POSTGRES_PORT_5432_TCP_PORT} - EXPOSE_NAS=${EXPOSE_NAS} - PROC_SITE=${PROC_SITE} + - SCHAIN_SITE=${SCHAIN_SITE} - SIRM_USER=${SIRM_USER} - SIRM_PASSWORD=${SIRM_PASSWORD} - SIRM_EMAIL=${SIRM_EMAIL} @@ -113,10 +114,12 @@ services: - backend_sirm labels: ofelia.enabled: "true" - ofelia.job-exec.adq-exp.schedule: "@every 10s" - ofelia.job-exec.adq-exp.command: "python manage.py adq_exp" - ofelia.job-exec.restart-exp.schedule: "0 0 0/17 ? * *" - ofelia.job-exec.restart-exp.command: "python manage.py restart_exp" + ofelia.job-exec.restart-reception.schedule: "0 1/5 * * * *" + ofelia.job-exec.restart-reception.command: "python manage.py restart_reception" + ofelia.job-exec.restart-pedestal.schedule: "0 2/10 * * * *" + ofelia.job-exec.restart-pedestal.command: "python manage.py restart_pedestal" + ofelia.job-exec.restart-experiment.schedule: "0 0 5 * * *" + ofelia.job-exec.restart-experiment.command: "python manage.py restart_experiment" logging: driver: "json-file" options: diff --git a/requirements.txt b/requirements.txt index 273d194..b8e0cc8 100644 --- a/requirements.txt +++ b/requirements.txt @@ -12,4 +12,5 @@ idna==3.3 urllib3==1.26.9 charset-normalizer==2.0.12 certifi==2021.10.8 -soupsieve==2.3.1 \ No newline at end of file +soupsieve==2.3.1 +h5py==3.7.0 \ No newline at end of file diff --git a/volumes/sirm/apps/main/management/commands/restart_exp.py b/volumes/sirm/apps/main/management/commands/restart_experiment.py similarity index 94% rename from volumes/sirm/apps/main/management/commands/restart_exp.py rename to volumes/sirm/apps/main/management/commands/restart_experiment.py index a5a96b7..3d872f5 100644 --- a/volumes/sirm/apps/main/management/commands/restart_exp.py +++ b/volumes/sirm/apps/main/management/commands/restart_experiment.py @@ -5,9 +5,9 @@ import time class Command(BaseCommand): """ - Restart experiment each 06 hours. + Restart experiment every night at 05:00 am. Example: - manage.py restart_exp --pk=1 + manage.py restart_experiment """ def handle(self, *args, **options): restart_experiment(self) diff --git a/volumes/sirm/apps/main/management/commands/restart_pedestal.py b/volumes/sirm/apps/main/management/commands/restart_pedestal.py new file mode 100644 index 0000000..414bfd3 --- /dev/null +++ b/volumes/sirm/apps/main/management/commands/restart_pedestal.py @@ -0,0 +1,200 @@ +from django.core.management.base import BaseCommand +from apps.main.models import Experiment +from django.shortcuts import get_object_or_404 +import os, fnmatch +import time +from datetime import datetime +import h5py +import numpy as np + +class Command(BaseCommand): + """ + Check pedestal acquisition each 10 minutes. + Example: + manage.py restart_pedestal + """ + def handle(self, *args, **options): + #start = time.time() + #time.sleep(1) + restart_pedestal(self) + #end = time.time() + #self.stdout.write(f'TIME: "{end - start}"') + +def check_experiment(): + if len(Experiment.objects.filter(status=2)) > 0: + return True + else: + return False + +def pedestal_start(self, id_exp): + all_status = Experiment.objects.filter(status=2) + check_id = False + + if len(all_status) > 0: + check_id = all_status[0].pk + + if check_id and check_id == id_exp: + exp = get_object_or_404(Experiment, pk=id_exp) + name = '{}-R@{}'.format(exp.name, datetime.now().strftime('%Y-%m-%dT%H-%M-%S')) + exp.pedestal.start_device(name_experiment=name) + self.stdout.write(f'"{exp.name}" experiment: Pedestal acquisition was restarted') + +def pedestal_stop(self, id_exp): + all_status = Experiment.objects.filter(status=2) + check_id = False + + if len(all_status) > 0: + check_id = all_status[0].pk + + if check_id and check_id == id_exp: + exp = get_object_or_404(Experiment, pk=id_exp) + exp.pedestal.stop_device() + self.stdout.write(f'"{exp.name}" experiment: Pedestal acquisition "{exp.name}" was stopped') + +def hdf5_list_content(get_file): + table_records = np.array(get_file).tolist() + table_dim = get_file.ndim + table_rows = get_file.shape[0] + + if table_dim == 1 and table_rows >= 1: + #type(table_records[0]) -> float + return table_records + else: + return False + +def hdf5_read(file): + dataspeed = {"ele_speed": False, "azi_speed": False} + + for key, value in dataspeed.items(): + with h5py.File(file, 'r') as hdf: + get = hdf.get('Data'+'/'+key) + if get is not None: + # 10 values + dataspeed[key] = hdf5_list_content(get)[-10:] + + return dataspeed + +def count_data(last_position): + pattern = "pos@*.h5" + count = 0 + list_data = [] + + list_position = os.listdir(last_position) + + for entry in sorted(list_position): + if fnmatch.fnmatch(entry, pattern): + count = count + 1 + list_data.append(os.path.join(last_position, entry)) + + if len(list_data) > 1: + list_data = list_data[-2] + else: + list_data = False + + return count, list_data + +def response_data(datadir, old_path_datetime, old_position, new_position): + path_position = True + path_datetime = False + read_files = False + + rootdir = os.path.join(datadir, 'position') + if os.path.isdir(rootdir): + path_datetime = path_data(os.path.join(datadir, 'position')) + + if path_datetime: + if not old_path_datetime or path_datetime != old_path_datetime: + old_position, read_files = count_data(path_datetime) + time.sleep(65) + new_position, read_files = count_data(path_datetime) + else: + time.sleep(65) + else: + path_position = False + + return path_position, path_datetime, old_position, new_position, read_files + +def path_data(rootdir): + list_=[] + for it in os.scandir(rootdir): + if it.is_dir(): + try: + datetime.strptime(it.path.split("/")[-1], "%Y-%m-%dT%H-00-00") + list_.append(it.path) + except ValueError: + pass + + list_ = sorted(list_, reverse=True) + try: + return list_[0] + except: + return False + +def check_count(datadir): + old_numbers = 0 + new_numbers = 0 + validation = False + path_datetime = False + speed = {"ele_speed": False, "azi_speed": False} + + path_position, path_datetime, old_numbers, new_numbers, read_files = response_data(datadir, path_datetime, old_numbers, new_numbers) + + for u in range(2): + if new_numbers > old_numbers: + validation = True + + data = hdf5_read(read_files) + for key, value in data.items(): + try: + if not max(data[key]) <= 0.1: + speed[key] = True + except: + pass + break + else: + if u < 1: + path_position, path_datetime, old_numbers, new_numbers, read_files = response_data(datadir, path_datetime, old_numbers, new_numbers) + + return path_position, path_datetime, validation, speed + +def restart_pedestal(self): + if check_experiment(): + + all_status = Experiment.objects.filter(status=2) + id_exp = all_status[0].pk + datadir_exp = all_status[0].reception_rx.datadir + datadir_exp = datadir_exp.replace(os.environ.get('EXPOSE_NAS', '/DATA_RM/DATA'), '/data') + datadir_exp = datadir_exp.replace('/rawdata', '') + + path_position, path_datetime, validation, speed = check_count(datadir_exp) + if path_position: + # Execute the process + if validation: + self.stdout.write(f'Acquisition pedestal is running') + if speed['ele_speed'] or speed['azi_speed']: + self.stdout.write(f'Pedestal speeds on Azimuth and Elevation are running') + else: + for key, value in speed.items(): + if not value: + self.stdout.write(f'Speed on {key} is <= 0.1, retry') + + pedestal_stop(self, id_exp) + time.sleep(14) + #pedestal_reset(self, id_exp) + #time.sleep(2) + pedestal_start(self, id_exp) + + else: + if not path_datetime: + self.stdout.write(f'No such directory with datetime format "%Y-%m-%dT%H-00-00", retry!') + else: + self.stdout.write(f'No file increment, retry') + + pedestal_stop(self, id_exp) + time.sleep(14) + #pedestal_reset(self, id_exp) + #time.sleep(2) + pedestal_start(self, id_exp) + + else: + self.stdout.write(f'No such directory: position, fail!') \ No newline at end of file diff --git a/volumes/sirm/apps/main/management/commands/adq_exp.py b/volumes/sirm/apps/main/management/commands/restart_reception.py similarity index 57% rename from volumes/sirm/apps/main/management/commands/adq_exp.py rename to volumes/sirm/apps/main/management/commands/restart_reception.py index 15a0a99..ec1c75b 100644 --- a/volumes/sirm/apps/main/management/commands/adq_exp.py +++ b/volumes/sirm/apps/main/management/commands/restart_reception.py @@ -4,15 +4,20 @@ from django.shortcuts import get_object_or_404 import os, fnmatch import time from datetime import datetime +import requests class Command(BaseCommand): """ - Check data acquisition each 10 seconds. + Check data acquisition each 05 minutes. Example: - manage.py adq_exp + manage.py restart_reception """ def handle(self, *args, **options): - restart_experiment(self) + #start = time.time() + time.sleep(15) + restart_acquisition(self) + #end = time.time() + #self.stdout.write(f'TIME: "{end - start}"') def check_experiment(): if len(Experiment.objects.filter(status=2)) > 0: @@ -32,6 +37,11 @@ def acquisition_start(self, id_exp): name = '{}-R@{}'.format(exp.name, datetime.now().strftime('%Y-%m-%dT%H-%M-%S')) exp.reception_rx.start_device(name_experiment = name, restart = True) self.stdout.write(f'"{exp.name}" experiment: Data acquisition was restarted') + self.stdout.write(f'Restarting schain...') + + r = requests.get('http://'+os.environ.get('SCHAIN_SITE', 'sophy-schain')+'/stop') + time.sleep(1) + r = requests.post('http://'+os.environ.get('SCHAIN_SITE', 'sophy-schain')+'/start', json={'name': exp.name}) def acquisition_stop(self, id_exp): all_status = Experiment.objects.filter(status=2) @@ -45,18 +55,36 @@ def acquisition_stop(self, id_exp): exp.reception_rx.stop_device() self.stdout.write(f'"{exp.name}" experiment: Data acquisition "{exp.name}" was stopped') -def count_data(datadir): +def count_data(last_channel): pattern = "rf@*.h5" - rawdata = {'ch0': 0, 'ch1': 0} + count = 0 + list_channel = os.listdir(last_channel) + + for entry in sorted(list_channel): + if fnmatch.fnmatch(entry, pattern): + count = count + 1 + return count - for key, value in rawdata.items(): - last_channel = path_data(os.path.join(datadir, key)) - if last_channel: - list_channel = os.listdir(last_channel) - for entry in sorted(list_channel): - if fnmatch.fnmatch(entry, pattern): - rawdata[key] = rawdata[key] + 1 - return rawdata +def response_data(datadir, old_channel, old_rawdata, new_rawdata, search): + path_channels = {'ch0': True, 'ch1': True} + channel = {'ch0': False, 'ch1': False} + + for key, value in path_channels.items(): + rootdir = os.path.join(datadir, key) + if os.path.isdir(rootdir): + channel[key] = path_data(os.path.join(datadir, key)) + if key in search: + if channel[key]: + if not old_channel[key] or channel[key] != old_channel[key]: + old_rawdata[key] = count_data(channel[key]) + time.sleep(1) + new_rawdata[key] = count_data(channel[key]) + else: + time.sleep(1) + else: + path_channels[key] = False + + return path_channels, channel, old_rawdata, new_rawdata def path_data(rootdir): list_=[] @@ -74,35 +102,52 @@ def path_data(rootdir): except: return False -def check_count(datadir, old_numbers): - diff = {} - numbers = count_data(datadir) +def check_count(datadir): + old_numbers = {'ch0': 0, 'ch1': 0} + new_numbers = {'ch0': 0, 'ch1': 0} + validation = {'ch0': False, 'ch1': False} + channel = {'ch0': False, 'ch1': False} - for key, value in numbers.items(): - if old_numbers[key] > numbers[key]: - old_numbers[key] = 0 - diff[key] = numbers[key] - old_numbers[key] - return numbers, diff + path_channels, channel, old_numbers, new_numbers = response_data(datadir, channel, old_numbers, new_numbers, ['ch0', 'ch1']) -def restart_experiment(self): - old_numbers={'ch0': 0, 'ch1': 0} - for count in range(5): - time.sleep(1) - if check_experiment(): - all_status = Experiment.objects.filter(status=2) - id_exp = all_status[0].pk - datadir_exp = all_status[0].reception_rx.datadir - datadir_exp = datadir_exp.replace(os.environ.get('EXPOSE_NAS', '/DATA_RM/DATA'), '/data') - - old_numbers, diff = check_count(datadir_exp, old_numbers) - if diff['ch0'] > 0 and diff['ch1'] > 0: - self.stdout.write(f'Data acquisition is running') + for key, value in validation.items(): + for _ in range(5): + if new_numbers[key] > old_numbers[key]: + validation[key] = True + break else: - count = count + 1 - if count == 5: - acquisition_stop(self, id_exp) - time.sleep(1) - acquisition_start(self, id_exp) - old_numbers={'ch0': 0, 'ch1': 0} + path_channels, channel, old_numbers, new_numbers = response_data(datadir, channel, old_numbers, new_numbers, [key]) + + return path_channels, channel, validation + +def restart_acquisition(self): + if check_experiment(): + all_status = Experiment.objects.filter(status=2) + id_exp = all_status[0].pk + datadir_exp = all_status[0].reception_rx.datadir + datadir_exp = datadir_exp.replace(os.environ.get('EXPOSE_NAS', '/DATA_RM/DATA'), '/data') + + path_channels, channel, validation = check_count(datadir_exp) + + if path_channels['ch0'] and path_channels['ch1']: + # Execute the process + if validation['ch0'] and validation['ch1']: + self.stdout.write(f'Data acquisition is running') + else: + if not channel['ch0'] or not channel['ch1']: + for key, value in channel.items(): + if not value: + self.stdout.write(f'No such directory with datetime format "%Y-%m-%dT%H-00-00": channel["{key}"], retry!') else: - self.stdout.write(f'An error ocurred while trying to read data acquisition, Retry!') \ No newline at end of file + for key, value in validation.items(): + if not value: + self.stdout.write(f'No file increment: channel["{key}"]') + + acquisition_stop(self, id_exp) + time.sleep(3) + acquisition_start(self, id_exp) + + else: + for key, value in path_channels.items(): + if not value: + self.stdout.write(f'No such directory: channel["{key}"], fail!') \ No newline at end of file