Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use publish_type method in localstack

Best Python code snippet using localstack_python

publish.py

Source:publish.py

1import re2import time3from PySide2 import QtCore, QtWidgets4from akapipe.core import database as akadatabase5from akapipe.core import db as akadb6from akapipe.core import templates as akatemplates7from . import base8from .. import actions, log9from .. import common10from .. import database11from .. import images12from .. import ui13from ..external import ffmpeg14AnimPublish = 015CompPublish = 116PRESETS = {17    AnimPublish: {18        'name': 'Work in Progress',19        'token': akatemplates.AV_AnimationPublish,20        'formats': (21            'png',22            'jpg',23        ),24    },25    CompPublish: {26        'name': 'Comp & Final',27        'token': akatemplates.AV_CompPublish,28        'formats': (29            'png',30            'jpg',31            'exr',32            'tif',33        ),34    },35}36def get_stamp_text(publish_type, source=None):37    v = 'Published on {time} by {user} to {publish_type}'.format(38        time=time.strftime('%d/%m/%Y %H:%M:%S'),39        publish_type=publish_type,40        user=common.get_username(),41    )42    if source:43        v += '\n Source:\n{}'.format(source)44    return v45@common.error46@common.debug47def stamp_publish(server, job, root, stamp_path, source, publish_type):48    # Write stamp file to publish dir49    with open(stamp_path, 'w') as f:50        v = get_stamp_text(publish_type, source=source)51        f.write(v)52    # Save the stamp info in the bookmark database53    db = database.get_db(server, job, root)54    common.proxy_path(source)55    with db.connection():56        db.setValue(57            common.proxy_path(source),58            'description',59            get_stamp_text(publish_type),60            table=database.AssetTable61        )62def get_progress_bar(frames):63    pbar = QtWidgets.QProgressDialog()64    common.set_stylesheet(pbar)65    pbar.setFixedWidth(common.size(common.DefaultWidth))66    pbar.setLabelText('Publishing files...')67    pbar.setMinimum(1)68    pbar.setMaximum(len(frames))69    pbar.setRange(1, len(frames))70    return pbar71def publish_footage(72        preset,73        source,74        frames,75        seq=None,76        shot=None,77        server=None,78        job=None,79        root=None,80        asset=None,81        task=None,82        make_movie=True,83        add_timecode=None,84        video_size=None,85        ffmpeg_preset=None,86        copy_path=True,87        reveal_publish=True,88):89    """Our footage publish script.90    Args:91        ref (weakref):  A reference to an item's data.92        publish_type (unicode):  The publish type.93        reveal (bool):  Show the publish folder in the file explorer.94    """95    is_collapsed = common.is_collapsed(source)96    if not is_collapsed:97        raise RuntimeError('Item is not a sequence.')98    if not frames:99        raise RuntimeError('Sequence seems to be empty.')100    ext = source.split('.')[-1]101    if make_movie and ext.lower() not in images.oiio_image_extensions:102        raise RuntimeError(f'{ext} is not a accepted image format.')103    # Let's check if the template file exists104    if not all((seq, shot)):105        seq, shot = base.get_seq_shot(source)106    if not all((seq, shot)):107        raise RuntimeError(108            'The item cannot be published because it is not in a shot folder.'109        )110    if ext.lower() not in PRESETS[preset]['formats']:111        raise RuntimeError('Not a valid publish format.')112    client, project = base.get_client_project()113    if not all((client, project)):114        raise RuntimeError(115            'Could not get the client and project names from the the current job.'116        )117    destination = akatemplates.expand_token(118        PRESETS[preset]['token'],119        PR_NetworkPath=server,120        client=client,121        project=project,122        sequence=seq,123        shot=shot,124    )125    if not QtCore.QFileInfo(destination).exists():126        if not QtCore.QDir(destination).mkpath('.'):127            raise RuntimeError(f'Could not create {destination}.')128    # Get the publish type from the path129    regex = r'|'.join(akatemplates.tokens[f] for f in akatemplates.publish_types)130    publish_type = re.search(regex, destination).group(0)131    publish_type = re.sub(r'[0-9_-]+', '', publish_type)132    # Initialize the aka database and get the client and project prefixes133    akadatabase.init_table_data(akadb.CL)134    akadatabase.init_table_data(akadb.PR)135    cl_abbrev = akadatabase.get_value(akadb.CL, client, akadb.CL_Abbreviation)136    pr_abbrev = akadatabase.get_value(akadb.PR, project, akadb.PR_Abbreviation)137    padding = akatemplates.tokens[akatemplates.AV_PublishSequenceFile].count('#')138    frames = sorted(frames, key=lambda x: int(x))139    pbar = get_progress_bar(frames)140    pbar.open()141    for idx, frame in enumerate(frames, 1):142        if pbar.wasCanceled():143            break144        pbar.setValue(idx)145        QtWidgets.QApplication.instance().processEvents()146        source_frame = f'{is_collapsed.group(1)}{frame}{is_collapsed.group(3)}'147        _framename = akatemplates.expand_token(148            akatemplates.AV_PublishSequenceFile,149            CL_Abbreviation=cl_abbrev,150            PR_Abbreviation=pr_abbrev,151            publish_type=publish_type,152            sequence=seq,153            shot=shot,154            ext=ext,155        )156        _framename = re.sub(r'[#]+', str(int(idx)).zfill(padding), _framename)157        destination_frame = f'{destination}/{_framename}'158        # Make thumbnail159        try:160            if idx == 1:161                # Now let's create a thumbnail for this publish162                images.ImageCache.oiio_make_thumbnail(163                    source_frame,164                    f'{destination}/thumbnail.png',165                    common.thumbnail_size166                )167        except:168            log.error('Could not make thumbnail.')169        # Create a movie170        if make_movie and idx == 1:171            try:172                movie_path = ffmpeg.convert(173                    source_frame,174                    ffmpeg_preset,175                    server=server,176                    job=job,177                    root=root,178                    asset=asset,179                    task=task,180                    size=video_size,181                    timecode=add_timecode182                )183                _moviename = akatemplates.expand_token(184                    akatemplates.AV_PublishFile,185                    CL_Abbreviation=cl_abbrev,186                    PR_Abbreviation=pr_abbrev,187                    publish_type=publish_type,188                    sequence=seq,189                    shot=shot,190                    ext=next(191                        v['output_extension'] for v in ffmpeg.PRESETS.values()192                        if v['preset'] == ffmpeg_preset193                    ),194                )195                destination_movie = f'{destination}/{_moviename}'196                destination_movie_file = QtCore.QFile(destination_movie)197                if destination_movie_file.exists():198                    if not destination_movie_file.remove():199                        raise RuntimeError('Could not remove movie file')200                if movie_path and not QtCore.QFile.copy(movie_path, destination_movie):201                    log.error(f'Could copy {movie_path}')202                if copy_path:203                    actions.copy_path(204                        destination_movie,205                        mode=common.WindowsPath206                    )207            except Exception as e:208                log.error('Failed to make movie.')209        destination_file = QtCore.QFile(destination_frame)210        # The actual copy operation211        if destination_file.exists():212            if not destination_file.remove():213                log.error(f'Could not remove {destination_frame}')214        QtCore.QFile.copy(source_frame, destination_frame)215    # Stamp the publish so we can backtrace if needed216    try:217        stampname = akatemplates.expand_token(218            akatemplates.AV_PublishFile,219            CL_Abbreviation=cl_abbrev,220            PR_Abbreviation=pr_abbrev,221            publish_type=publish_type,222            sequence=seq,223            shot=shot,224            ext='txt',225        )226        stamp_publish(227            server, job, root,228            f'{destination}/{stampname}',229            source,230            publish_type231        )232    except Exception as e:233        log.error(e)234    try:235        from ..teams import message236        db = database.get_db(server, job, root)237        webhook = db.value(db.source(), 'teamstoken', database.BookmarkTable)238        if webhook:239            payload = message.get_payload(240                message.PUBLISH_MESSAGE,241                thumbnail=f'{destination}/thumbnail.png',242                seq=seq,243                shot=shot,244                path=destination,245                date=time.strftime('%d/%m/%Y %H:%M:%S'),246                user=common.get_username(),247                publish_type=publish_type,248            )249            message.send(webhook, payload)250    except Exception as e:251        log.error(e)252    if copy_path:253        ui.OkBox(254            'Item published successfully.',255            'The path to the published movie file has been copied to the clipboard.'256        ).open()257    else:258        ui.OkBox(259            'Item published successfully.'260        ).open()261    if reveal_publish:...

test_redunlive_models.py

Source:test_redunlive_models.py

1# -*- coding: utf-8 -*-2"""Tests for `models` in redunlive webapp."""3import httpretty4import pytest5from epipearl import Epipearl6from cadash.redunlive.models import CaptureAgent7from cadash.redunlive.models import CaLocation8epiphan_url = 'http://fake.example.edu'9class TestCaptureAgentModel(object):10    def setup(self):11        p = CaptureAgent('ABCD1111', 'fake1.example.edu')12        p.channels['live']['channel'] = '1'13        p.channels['live']['publish_type'] = '0'14        p.channels['lowBR']['channel'] = '2'15        p.channels['lowBR']['publish_type'] = '0'16        p.client = Epipearl(epiphan_url, 'user', 'passwd')17        self.ca = p18    @httpretty.activate19    def test_get_converging_live_status(self):20        live = self.ca.channels['live']21        lowBR = self.ca.channels['lowBR']22        assert live['publish_type'] == '0'23        httpretty.register_uri(24                httpretty.GET, '%s/admin/channel%s/get_params.cgi' %25                (epiphan_url, live['channel']), body='publish_type = 6')26        httpretty.register_uri(27                httpretty.GET, '%s/admin/channel%s/get_params.cgi' %28                (epiphan_url, lowBR['channel']), body='publish_type = 6')29        self.ca.sync_live_status()30        assert live['publish_type'] == lowBR['publish_type']31        assert live['publish_type'] == '6'32    @httpretty.activate33    def test_get_diverging_live_status(self):34        live = self.ca.channels['live']35        lowBR = self.ca.channels['lowBR']36        httpretty.register_uri(37                httpretty.GET, '%s/admin/channel%s/get_params.cgi' %38                (epiphan_url, live['channel']), body='publish_type = 0')39        httpretty.register_uri(40                httpretty.GET, '%s/admin/channel%s/get_params.cgi' %41                (epiphan_url, lowBR['channel']), body='publish_type = 6')42        httpretty.register_uri(43                httpretty.GET, '%s/admin/channel%s/set_params.cgi' %44                (epiphan_url, lowBR['channel']), body='', status=201)45        self.ca.sync_live_status()46        assert live['publish_type'] == lowBR['publish_type']47        assert live['publish_type'] == '0'48    @httpretty.activate49    def test_set_live_status(self):50        live = self.ca.channels['live']51        lowBR = self.ca.channels['lowBR']52        httpretty.register_uri(53                httpretty.GET, '%s/admin/channel%s/set_params.cgi' %54                (epiphan_url, live['channel']), body='', status=201)55        httpretty.register_uri(56                httpretty.GET, '%s/admin/channel%s/set_params.cgi' %57                (epiphan_url, lowBR['channel']), body='', status=201)58        new_publish_status = '6'59        self.ca.write_live_status(new_publish_status)60        assert live['publish_type'] == lowBR['publish_type']61        assert live['publish_type'] == new_publish_status62class TestCaLocationModel(object):63    def setup(self):64        p = CaptureAgent('ABCD1111', 'fake1.example.edu')65        p.channels['live']['channel'] = '1'66        p.channels['live']['publish_type'] = '0'67        p.channels['lowBR']['channel'] = '2'68        p.channels['lowBR']['publish_type'] = '0'69        self.primary = p70        s = CaptureAgent('ABCD2222', 'fake2.example.edu')71        s.channels['live']['channel'] = '7'72        s.channels['live']['publish_type'] = '0'73        s.channels['lowBR']['channel'] = '8'74        s.channels['lowBR']['publish_type'] = '0'75        self.secondary = s76        e1 = CaptureAgent('ABCD3333', 'fake3.example.edu')77        e1.channels['live']['channel'] = '3'78        e1.channels['live']['publish_type'] = '0'79        e1.channels['lowBR']['channel'] = '4'80        e1.channels['lowBR']['publish_type'] = '0'81        self.experimental1 = e182        e2 = CaptureAgent('ABCD4444', 'fake4.example.edu')83        e2.channels['live']['channel'] = '3'84        e2.channels['live']['publish_type'] = '0'85        e2.channels['lowBR']['channel'] = '4'86        e2.channels['lowBR']['publish_type'] = '0'87        self.experimental2 = e288        l = CaLocation('loc1')89        l.primary_ca = self.primary90        l.secondary_ca = self.secondary91        l.experimental_cas = [self.experimental1, self.experimental2]92        self.loc = l93    def test_simple_location(self):94        assert self.loc.primary_ca.name == 'fake1'95        assert self.loc.secondary_ca.name == 'fake2'96        assert len(self.loc.experimental_cas) == 297        assert self.loc.active_livestream is None98    def test_same_ca_for_secondary(self):99        with pytest.raises(ValueError):100            self.loc.secondary_ca = self.primary101    def test_same_ca_for_primary(self):102        with pytest.raises(ValueError):103            self.loc.primary_ca = self.secondary104    def test_toggle_active_livestream(self):105        assert self.loc.active_livestream is None106        self.primary.channels['live']['publish_type'] = '6'107        assert self.loc.active_livestream == 'primary'108        self.primary.channels['live']['publish_type'] = '0'109        self.secondary.channels['live']['publish_type'] = '6'...

scraper.py

Source:scraper.py

1import requests2from pprint import pprint3import json4from multiprocessing.dummy import Pool as ThreadPool5import time6def read_config(filename):7    data = None8    with open(filename, 'r') as f:9        data = json.loads(f.read())10    return data11config = read_config('config.json')12puzzle_ids = []13game_url_template = "https://nyt-games-prd.appspot.com/svc/crosswords/v6/game/"14security_header = config['s-header']15user_id = config['user-id']16game_data = []17# for multithreading18urls = [19    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-01-01&date_end=2019-02-28',20    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-03-01&date_end=2019-04-30',21    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-05-01&date_end=2019-06-30',22    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-07-01&date_end=2019-07-31',23    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-08-01&date_end=2019-10-31',24    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2019-11-01&date_end=2020-01-31',25    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-02-01&date_end=2020-03-31',26    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-04-01&date_end=2020-06-31',27    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-07-01&date_end=2020-08-31',28    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-09-01&date_end=2020-11-31',29    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2020-12-01&date_end=2021-02-28',30    'https://nyt-games-prd.appspot.com/svc/crosswords/v3/' + user_id + '/puzzles.json?publish_type=mini&sort_order=asc&sort_by=print_date&date_start=2021-03-01&date_end=2021-03-08'31]32def get_cword_data(cword):33    data_url = game_url_template + str(cword['puzzle_id']) + ".json"34    cword_response = requests.get(data_url, headers=security_header)35    cword_response_data = json.loads(cword_response.content)36    cword["board"] = cword_response_data["board"]37    return cword38def main():39    start = time.time()40    # scrape crossword lists for publish dates and game ids41    for url in urls:42        response = requests.get(url, headers=security_header)43        data = json.loads(response.content)['results']44        solved_cwords = [cword for cword in data if cword['solved']]45        with ThreadPool(20) as pool:46            results = pool.map(get_cword_data, solved_cwords)47            game_data.extend(results)48    # output the list of boards somehow (for now, writing to a json file)49    with open("data.json", "w") as f:50        f.write(json.dumps(game_data))51    end = time.time()52    print("script finished in", end - start, "s")53if __name__ == "__main__":...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.