# Shift schedule client for saving and loading who is on shift # Derek Fujimoto # Nov 2025 import json import midas, midas.event import pandas as pd import numpy as np import os, datetime, time, glob, shutil import pytz.reference # separator for the string lists packed into the NAME, ROLE and NOTE banks. must # match BANK_DELIMITER in shiftschedule.js. ASCII unit separator, chosen because it # cannot appear in a shifter name, a role or a note typed by a user BANK_DELIMITER = '\x1f' # bytes Event.pack() prepends to the bank data, which is what midas measures against the # caller's max_len. all_bank_size_bytes covers only the banks themselves EVENT_OVERHEAD_BYTES = midas.event.event_header_size + midas.event.all_bank_header_size class Database(object): """Database put/get functions Attributes: data_type (str): csv|parquet, switch between database output types file_assign (str): filename, with path, for database shift assignment file_avail (str): filename, with path, for database shift availability file_notes (str): filename, with path, for database shift notes stateid (int|None): id number of this database state path (str): path to database files Args: logger (logger.Logger): logging object stateid (int|None): id number of this database state path (str): path to database files data_type (str): csv|parquet, switch between database output types """ def __init__(self, logger, stateid=None, path='.', data_type='csv'): self.stateid = stateid self.path = path self.data_type = data_type self.file_assign = os.path.join(path, f'shift_database') self.file_avail = os.path.join(path, f'avail_database') self.file_notes = os.path.join(path, f'notes_database') self.logger = logger self.logger.debug(f'Database object id {stateid} for "{data_type}" created with path {path}') def _read(self, filename): """Use pandas to read database. Infer data type based on filename extension Args: filename (str): path to the file Returns: pd.DataFrame """ # get filename with ambiguous file extension try: filename = glob.glob(f'{filename}.*')[0] self.logger.debug(f'Reading {filename}') except IndexError: msg = 'No database files found' self.logger.error(msg) raise FileNotFoundError(msg) if os.path.splitext(filename)[1] == '.csv': return pd.read_csv(filename) elif os.path.splitext(filename)[1] == '.parquet': df = pd.read_parquet(filename, engine='fastparquet') return df.reset_index() else: msg = f'Unknown data type {os.path.splitext(filename)[1]} for file {filename}.' self.logger.error(msg) raise RuntimeError(msg) def _write(self, df, filename): """Use pandas to write database Args: df (pd.DataFrame): dataframe to write filename (str): path to the file """ # check the type before touching anything on disk: an unknown type used to delete # the database first and raise afterwards, losing the data if self.data_type not in ('csv', 'parquet'): msg = f'Unknown data type {self.data_type}' self.logger.error(msg) raise RuntimeError(msg) target = f'{filename}.{self.data_type}' # write to a temporary file and swap it in, so an interruption partway through # cannot leave the database truncated or missing. the temporary name is chosen to # not match the "{filename}.*" glob used to read and to clean up head, tail = os.path.split(filename) temp = os.path.join(head, f'.tmp_{tail}.{self.data_type}') try: if self.data_type == 'csv': df.to_csv(temp) else: df.to_parquet(temp, engine='fastparquet', compression='snappy') except Exception: # leave no partial file behind: the existing database is still intact if os.path.exists(temp): os.remove(temp) raise os.replace(temp, target) self.logger.info(f'Writing {target}') self.logger.debug(f'Data written: \n{df.tail()}') # drop any database left over in the other format, ex: after switching data_type for name in glob.glob(f'{filename}.*'): if name != target: os.remove(name) self.logger.debug(f'Removing old file {name}') # and any temporary file a previous run was killed partway through. these are # dot-prefixed, so neither the glob above nor the one copy_state uses would ever # find them, and they would sit in the state directory forever for name in glob.glob(os.path.join(head, f'.tmp_{tail}.*')): os.remove(name) self.logger.debug(f'Removing stale temporary file {name}') def put_assign(self, names, tstart, roles): """Save shift to database. Multiple shifters can be assigned to the same shift. Args: names (list of str): shifter names to assign to that date and shift tstart (float): epoch time of the shift date in ms, with shift index encoded as ms role (str): role of the shifter in their shift Returns: None, writes database file """ self.logger.info(f'Assigning {names} for time {pd.to_datetime(tstart, unit="ms")} ({tstart}) in role {roles}') # make a dataframe with the new entires df_new = pd.DataFrame(columns=['name', 'time', 'role']) for i, (name, role) in enumerate(zip(names, roles)): df_new.loc[i, 'time'] = tstart df_new.loc[i, 'name'] = name df_new.loc[i, 'role'] = role df_new.set_index('time', inplace=True) # get the database if it exists try: df = self._read(self.file_assign) self.logger.debug(f'Read {self.file_assign}') except FileNotFoundError: df = pd.DataFrame(columns=['name', 'role']) df.index.name = 'time' self.logger.debug(f'No file {self.file_assign} found, creating new') else: # set index df.set_index('time', inplace=True) # fill empty roles df.role = df.role.fillna('') self.logger.debug(f'filled empty roles with blanks') # remove entries for this timestamp and all roles df.drop(index=tstart, inplace=True, errors='ignore') self.logger.debug(f'dropped indices {tstart}') # combine, accounting for empty if df.empty: df = df_new elif df_new.empty: pass else: df = pd.concat((df, df_new), axis='index') # write to file if df.empty: for path in glob.glob(f'{self.file_assign}.*'): os.remove(path) self.logger.info(f"No data left in assign database, removing {path}") else: self._write(df, self.file_assign) def get_assign(self, time_start=None, time_end=None): """Read shift assignments Args: time_start (int): day start epoch time in ms, inclusive time_end (int): day end epoch time in ms, inclusive Returns: pd.DataFrame: columns are shift names, indexed by epoch times """ # get the database try: df = self._read(self.file_assign) except FileNotFoundError: self.logger.info(f'File {self.file_assign} not found, returning blank dataframe') return pd.DataFrame(columns=['time', 'name', 'role']) # get range of dates - add one second to the end to ensure all shifts are accounted for idx = (df.time >= time_start) & (df.time <= time_end+1000) df = df.loc[idx] return df def put_avail(self, name, times, states): """record availabilities in database Args: name (str): shifter name times (iterable(float)): epoch times of the dates to set, with shift index encoded as miliseconds states (iterable(bool)): True if available, else False Returns: None, writes database file, indexed by time, columns are shifter names """ # make a new dataframe df_new = pd.DataFrame({name:states}, index=times) df_new.index.name = 'time' # get the database if it exists try: df = self._read(self.file_avail) self.logger.debug(f'Read availability database as \n{df.tail()}') except FileNotFoundError: df = pd.DataFrame({'time':times, name:states}) df.set_index('time', inplace=True) self.logger.info(f'No avail database found, making a new one: \n{df.tail()}') else: # set index df.set_index('time', inplace=True) # set the states if name not in df.columns: df.loc[:, name] = False for t, s in zip(times, states): if t not in df.index: df.loc[t] = False df.loc[t, name] = bool(s) # drop all false rows and columns idx = df.apply(any, axis='columns') cols = df.apply(any, axis='index') df = df.loc[idx, cols] self.logger.debug(f'Avail database to write to file:\n{df.tail()}') # write to file if df.empty: for path in glob.glob(f'{self.file_avail}.*'): os.remove(path) self.logger.info(f"No data left in avail database, removing {path}") else: self._write(df, self.file_avail) def get_avail(self, name, time_start=None, time_end=None): """Read availabilities from database Args: name (str): name of shifter | "all" time_start (int): day start epoch time, inclusive time_end (int): day end epoch time, inclusive Returns: pd.Series: boolean values, indexed by epoch times, with shift index encoded in the milliseconds term """ # get the database try: df = self._read(self.file_avail) self.logger.debug(f'Initial read avail database as \n{df.tail()}') except FileNotFoundError: # "all" is a request sentinel, not a shifter: with no database there are no # shifters to report, so it must not become a column cols = ['time'] if name == 'all' else ['time', name] return pd.DataFrame(columns=cols) # exception: return all names if name == 'all': cols = df.columns else: cols = ['time', name] # check that name is in the database. only meaningful for a single shifter: # for "all" the columns come from the database by definition, and indexing # cols[1] there would raise on a database with no shifters in it if name not in df.columns: return pd.DataFrame(columns=cols) # get range of dates - add one second to the end to ensure all shifts are accounted for idx = (df.time >= time_start) & (df.time <= time_end+1000) df = df.loc[idx, cols] df = df.fillna(False) self.logger.debug(f'Returning avail database as \n{df.tail()}') return df def put_notes(self, note, time): """Record notes in database Args: note (str) time (int): epoch time of the date to set Returns: None, writes database file, indexed by time, only one column (notes) """ # note must be string note = str(note) # get the database if it exists try: df = self._read(self.file_notes) self.logger.debug(f'Read notes database as \n{df.tail()}') except FileNotFoundError: df = pd.DataFrame({'note':[note], 'time':[time]}) df.set_index('time', inplace=True) # set the note else: df.set_index('time', inplace=True) df.note = df.note.astype(str) df.loc[time , 'note'] = note # drop empty notes df = df.loc[df.note.apply(bool)] # write to file self.logger.debug(f'Writing notes database as \n{df.tail()}') if df.empty: for path in glob.glob(f'{self.file_notes}.*'): os.remove(path) self.logger.info(f"No data left in note database, removing {path}") else: self._write(df, self.file_notes) def get_notes(self, time_start=None, time_end=None): """Get notes from database Args: time_start (int): day start epoch time, inclusive time_end (int): day end epoch time, inclusive Returns: pd.Series: strings for notes, indexed by epoch times """ # get the database try: df = self._read(self.file_notes) self.logger.debug(f'Read notes database as \n{df.tail()}') except FileNotFoundError: return pd.DataFrame(columns=['time', 'note']) # get range of dates - add one second to the end to ensure all shifts are accounted for idx = (df.time >= time_start) & (df.time <= time_end+1000) df = df.loc[idx] # drop nan df.dropna(inplace=True) # notes must be strings if df['note'].dtype != object: note = df['note'].astype(str) df.drop(columns=['note'], inplace=True) df['note'] = note self.logger.debug(f'Returning notes database as \n{df.tail()}') return df def get_assign_event(self, time_start=None, time_end=None): """Read shift assignments, return midas event Args: time_start (int): day start epoch time in ms, inclusive time_end (int): day end epoch time in ms, inclusive Returns: midas.event.Event: banks TIME, NAME, and ROLE Notes: TIME: array of ints NAME: string, deliminated by _ ROLES: string, deliminated by _ """ # fetch database df = self.get_assign(time_start=time_start, time_end=time_end) event = midas.event.Event() # add times to event as nparray of ints event.create_bank("TIME", midas.TID_INT64, df.time.values) # add names as "_"-separated string names = bytes((BANK_DELIMITER.join(df.name)).encode('utf-8')) event.create_bank("NAME", midas.TID_CHAR, names) # add roles as "_"-separated string df.role = df.role.fillna('') roles = bytes((BANK_DELIMITER.join(df.role)).encode('utf-8')) event.create_bank("ROLE", midas.TID_CHAR, roles) # setup header event.header.event_id = 1 event.header.serial_number = 1 return event def get_avail_event(self, name, time_start=None, time_end=None): """Read availabilities from database, return event Args: name (str): name of shifter | "all" time_start (int): day start epoch time, inclusive time_end (int): day end epoch time, inclusive Returns: midas.event.Event: banks TIME, NAME, N000, N001, N002, ... Notes: TIME: array of ints NAME: string, deliminated by "_" N***: array of booleans, if true, availabile for shift. The number *** corresponds to name index after splitting at deliminator """ df = self.get_avail(name = name, time_start=time_start, time_end=time_end) event = midas.event.Event() # add times to event as nparray of ints event.create_bank("TIME", midas.TID_INT64, df.time.values) # add names as "_"-separated string names = [n for n in df.columns if n != 'time'] bytename = bytes((BANK_DELIMITER.join(names)).encode('utf-8')) event.create_bank("NAME", midas.TID_CHAR, bytename) # make banks of booleans for each name. Number indicates index in name bank for i, name in enumerate(names): event.create_bank(f'N{i:03d}', midas.TID_BOOL, df[name].values) event.header.event_id = 1 event.header.serial_number = 1 return event def get_notes_event(self, time_start=None, time_end=None): """Get notes from database, return event Args: time_start (int): day start epoch time, inclusive time_end (int): day end epoch time, inclusive Returns: midas.event.Event: banks TIME, NOTE """ # get database df = self.get_notes(time_start=time_start, time_end=time_end) event = midas.event.Event() # add times to event as nparray of ints event.create_bank("TIME", midas.TID_INT64, df.time.values) # add notes as string bytename = bytes((BANK_DELIMITER.join(df.note)).encode('utf-8')) event.create_bank("NOTE", midas.TID_CHAR, bytename) event.header.event_id = 1 event.header.serial_number = 1 return event class Scheduler(object): """Interface with the database, track current state Args: client (midas.client.MidasClient) logger (logging.Logger) Attributes: database_type (str): CSV, fetched from ODB states_past (list): [Database], states older than current state. Last entry is the current state states_future (list): [Database], states newer than current state webdir (str): path to data directory containing dirs with database files """ path_setup = '/Shifts/ShiftSetup' path_stateid = '/Shifts/Variables/stateid' path_database = '/Shifts/Settings/database' path_atstart = '/Shifts/Variables/atstart' path_atend = '/Shifts/Variables/atend' history_len = 50 # max number of saved history steps def __init__(self, client, logger): self.client = client self.logger = logger # get all state objects self.webdir = client.odb_get('/Custom/Path') self.webdir = os.path.join(self.webdir, 'shiftschedule', 'data') paths = glob.glob(os.path.join(self.webdir, '*')) # state directories are named after their integer id, so ignore anything else that # may have been dropped in the data directory paths = [path for path in paths if os.path.basename(path).isdigit()] # sort numerically, not as strings: current_database, gen_id and delete_states_past # all assume ascending id order, and a string sort puts "10" before "2" paths.sort(key=lambda path: int(os.path.basename(path))) logger.info('Scheduler: read all databases') self.states_past = [Database(self.logger ,int(os.path.basename(path)), path, self.database_type) for path in paths] self.states_future = [] logger.debug(f'Database paths from {self.webdir}: {paths}') # ensure past states is not zero length: make new directory and database object if not len(self.states_past): logger.info('No past states detected, creating default') id = self.gen_id() path = os.path.join(self.webdir, str(id)) os.makedirs(path) self.states_past.append(Database(self.logger, id, path, self.database_type)) # ensure current state exists, update current stateid if not client.odb_exists(self.path_stateid) or len(self.states_past) == 1: self.set_current_id(self.states_past[-1].stateid) # split into past and future states: LIFO stacks # future states are the ones newer than the current id, which is where the user # left off after any undos current_id = self.current_id st = self.states_past.pop() while st.stateid > current_id and len(self.states_past): self.states_future.append(st) st = self.states_past.pop() self.states_past.append(st) # publish the state we actually landed on. set_current_id is the only writer of the # atstart/atend flags the undo and redo buttons read, so without this they keep # whatever values the previous run left behind self.set_current_id(self.current_database.stateid) logger.info(f'Past states: {[st.stateid for st in self.states_past]}') logger.info(f'Future states: {[st.stateid for st in self.states_future]}') logger.info('Scheduler initialized') def copy_state(self): """Copy the current database, assign a new id number, write to file.""" # delete future states self.delete_states_future() # new stateid id = self.gen_id() # make new directory new_path = os.path.dirname(self.current_database.path) new_path = os.path.join(new_path, str(id)) os.makedirs(new_path, exist_ok=True) # copy the databases to the new location for file in glob.glob(os.path.join(self.current_database.path, '*')): shutil.copy(file, new_path) self.logger.debug(f'Copied {file} -> {new_path}') # make a new database object db_new = self.current_database.__class__(self.logger, id, new_path, self.database_type) # set the current state self.states_past.append(db_new) self.set_current_id(id) # trim past states to length of self.history_len self.delete_states_past() @property def current_database(self): """Get current database Returns: Database: corresponds to current state id """ return self.states_past[-1] @property def current_id(self): """Get the current state id from the ODB Returns: int: state id """ try: id = self.client.odb_get(self.path_stateid) self.logger.debug(f'Found state id {id}') except KeyError: # fall back to the state we are actually on, not gen_id(), which returns the id # of the next state to be created and so names a state that does not exist id = self.current_database.stateid self.client.odb_set(self.path_stateid, id) self.logger.debug(f'No state id found, generating id {id}') return id @property def database_type(self): """Get the database type from the ODB""" # strip as well as lower: a trailing space is easy to introduce in the ODB browser # and would otherwise fail the csv/parquet comparison return self.client.odb_get(self.path_database).strip().lower() def delete_states_future(self): """Delete states_future list and corresponding directories""" self.logger.debug(f'Deleting future states {[st.stateid for st in self.states_future]}') for dat in self.states_future: shutil.rmtree(dat.path) self.states_future = [] def delete_states_past(self): """Ensure that there are not too many states in the past, at most self.history_len""" n_remove = max(len(self.states_past)-self.history_len, 0) while n_remove > 0: dat = self.states_past.pop(0) shutil.rmtree(dat.path) n_remove -= 1 self.logger.debug(f'Deleted past state {dat.stateid}') def gen_id(self): """Generate id for a new state""" # ids must be in increasing order and unique try: id = self.states_past[-1].stateid + 1 except IndexError: id = 0 self.logger.debug(f'Generated new stateid {id}') return id def move_state_backward(self): """Go backward in time""" if len(self.states_past) > 1: dat = self.states_past.pop() self.states_future.append(dat) self.logger.info('Moved state backwards by one') self.logger.debug(f'New past states {[st.stateid for st in self.states_past]}') self.logger.debug(f'New future states {[st.stateid for st in self.states_future]}') self.set_current_id(self.current_database.stateid) def move_state_forward(self): """Go forward in time""" if len(self.states_future) > 0: dat = self.states_future.pop() self.states_past.append(dat) self.logger.info('Moved state forwards by one') self.logger.debug(f'New past states {[st.stateid for st in self.states_past]}') self.logger.debug(f'New future states {[st.stateid for st in self.states_future]}') self.set_current_id(self.current_database.stateid) def rpc_handler(self, client, cmd, args, max_len): """ This is the function that will be called when something/someone triggers the "JRPC" for this client (e.g. by using the javascript code above). Arguments: * client (midas.client.MidasClient) * cmd (str) - The command user wants to execute * args (str) - Other arguments the user supplied * max_len (int) - The maximum string length the user accepts in the return value Returns: 2-tuple of (int, str) for status code, message. """ # get arguments as dict args = json.loads(args) self.logger.debug(f'RPC Handler {cmd} command') # save shifters assignment if cmd == "put": self.copy_state() try: self.current_database.put_assign(names = args["names"], tstart = int(args['time']), roles = args['roles']) except Exception as err: self.logger.error(str(err)) raise err from None # update onshift status self.set_onshift() # return event event = midas.event.Event() event.create_bank('STAT', midas.TID_INT32, [self.current_id]) # get shifter assignment elif cmd == "get": try: event = self.current_database.get_assign_event(time_start=args['time_start'], time_end=args['time_stop']) except Exception as err: self.logger.error(str(err)) raise err from None # put shifter availability elif cmd == "putavail": self.copy_state() try: self.current_database.put_avail(name=args['name'], times=args['times'], states=args['states']) except Exception as err: self.logger.error(str(err)) raise err from None # return event event = midas.event.Event() event.create_bank('STAT', midas.TID_INT32, [self.current_id]) elif cmd == "getavail": try: event = self.current_database.get_avail_event(name = args['name'], time_start = args['time_start'], time_end = args['time_stop']) except Exception as err: self.logger.error(str(err)) raise err from None # put notes elif cmd == "putnotes": self.copy_state() try: self.current_database.put_notes(time=args['time'], note=args['note']) except Exception as err: self.logger.error(str(err)) raise err from None # return event event = midas.event.Event() event.create_bank('STAT', midas.TID_INT32, [self.current_id]) elif cmd == "getnotes": try: event = self.current_database.get_notes_event(time_start = args['time_start'], time_end = args['time_stop']) except Exception as err: self.logger.error(str(err)) raise err from None elif cmd == 'forward': self.move_state_forward() # return event event = midas.event.Event() event.create_bank('STAT', midas.TID_INT32, [self.current_id]) elif cmd == 'backward': self.move_state_backward() # return event event = midas.event.Event() event.create_bank('STAT', midas.TID_INT32, [self.current_id]) # bad command word else: ret_str = "Unknown command '%s'" % cmd self.logger.error(ret_str) # the event has to be made here: the other branches each make their own event = midas.event.Event() event.create_bank('STNG', midas.TID_CHAR, bytes(ret_str.encode('utf-8'))) return (midas.status_codes["FE_ERR_DRIVER"], event) # midas truncates a reply longer than max_len with no error and no status change, # leaving the frontend to misparse a bank whose length runs past the buffer. fail # loudly instead. reachable via getavail for "all" over a wide enough date range # # bank sizes rather than pack(): the branches above do not all set an event id, # which pack() requires. add the headers pack() would prepend, or a reply landing # within one header of the limit clears this check and is truncated anyway. midas # writes at most max_len-1 bytes, so the last usable size is max_len-1 event.calculate_bank_sizes() size = event.all_bank_size_bytes + EVENT_OVERHEAD_BYTES if size > max_len - 1: msg = (f"Reply to '{cmd}' is {size} bytes, over the {max_len} byte limit. " "Narrow the date range.") self.logger.error(msg) event = midas.event.Event() event.create_bank('STNG', midas.TID_CHAR, bytes(msg.encode('utf-8'))) return (midas.status_codes["FE_ERR_DRIVER"], event) return (midas.status_codes["SUCCESS"], event) def set_current_id(self, stateid): """Set the current id in the ODB Args: stateid (int): state id number """ # set self.client.odb_set(self.path_stateid, stateid, explicit_new_midas_type=midas.TID_UINT32) self.logger.info(f'Set current id to {stateid}') # check endpoints self.client.odb_set(self.path_atstart, len(self.states_past) == 1) self.client.odb_set(self.path_atend, len(self.states_future) == 0) def set_onshift(self): """Set names of who is on shift and their role""" path = self.path_setup # get shift start and end times shift_start = np.array(self.client.odb_get(f'{path}/start_offset_mins')) shift_stop = np.array(self.client.odb_get(f'{path}/stop_offset_mins')) shiftids = self.client.odb_get(f'{path}/shiftids') # get today's date and time with timezone tz = pytz.reference.LocalTimezone() now = datetime.datetime.now() today = now.replace(hour=0, second=0, minute=0, microsecond=0) today = int(today.timestamp() + tz.utcoffset(today).total_seconds()) now = int(now.timestamp() + tz.utcoffset(now).total_seconds()) # to milliseconds today *= 1000 now *= 1000 shift_start *= 60000 shift_stop *= 60000 # get shifters for yesterday - tomorrow and the corresponding ids from the database try: shifters = self.current_database.get_assign(time_start=today-86400000, time_end=today+86401000) except Exception as err: self.logger.error(str(err)) raise err from None shifters['shiftid'] = shifters.time.apply(lambda x : int(round(x/1000 % 1 * 1000))) shifters['time'] = shifters.time.apply(lambda x : int(round(x/1000)*1000)) # determine who is on shift and the corresponding shiftid shifters_now = [] shifters_now_id = [] for _, row in shifters.iterrows(): # the shift this row was assigned to may have since been removed or renumbered # in the ODB, in which case there is no start/stop to compare against try: idx = shiftids.index(row.shiftid) starti = shift_start[idx] stopi = shift_stop[idx] except (ValueError, IndexError): self.logger.warning(f'Assignment for "{row["name"]}" has shiftid ' f'{row.shiftid}, which is not in /Shifts/ShiftSetup, ' 'skipping') continue # check if now is within the time if now >= row.time+starti and now <= row.time+stopi: shifters_now.append(row['name']) shifters_now_id.append(row['shiftid']) anyone_on_shift = len(shifters_now) > 0 # mhttpd renders an ODB array by joining every element, padding included, so # publish a display string for the web page with the empty entries dropped onshift_str = ', '.join(shifters_now) if anyone_on_shift else 'nobody' # these keys are arrays created by setup_odb, and odb_set resizes to fit by # default, so pad to the existing length rather than shrinking the key existing = self.client.odb_get('/Shifts/Variables/onshift_names') length = max(len(existing) if isinstance(existing, list) else 1, len(shifters_now)) shifters_now = shifters_now + [''] * (length - len(shifters_now)) shifters_now_id = shifters_now_id + [0] * (length - len(shifters_now_id)) # set in ODB self.client.odb_set('/Shifts/Variables/onshift_names', shifters_now) self.client.odb_set('/Shifts/Variables/onshift_shiftid', shifters_now_id) # create the display key here too: deployments which never re-run --setup # would otherwise not have it self.client.odb_set('/Shifts/Variables/onshift_names_str', onshift_str, create_if_needed=True) if anyone_on_shift: self.logger.info(f'Set on-shift shifters as "{shifters_now}" with shiftids {shifters_now_id}') else: self.logger.debug(f'Set on-shift shifters as "{shifters_now}" with shiftids {shifters_now_id}') def run_forever(client, logger): """Run the scheduler forever Args: client (midas.client.Client) """ logger.info('Started scheduler') scheduler = Scheduler(client, logger) # Register our function. client.register_brpc_callback(scheduler.rpc_handler) client.msg('Started shiftschedule client') # Spin forever. Program can be killed by Ctrl+C or # "Stop Program" through mhttpd. t0 = time.monotonic() try: while True: client.communicate(100) # ms # update the onshift shifter every 60 seconds t1 = time.monotonic() if t1-t0 > 60: t0 = t1 # a failure here must not end the process: the custom page would fall back # to a "start program" button and stop updating altogether try: scheduler.set_onshift() except Exception as err: logger.error(f'Failed to update on-shift shifters: {err}') finally: logger.info('Finished scheduler')